<commit_before>#!/usr/bin/python

import os
import re
import shutil
import tempfile
import time
import unittest
from mininet.log import setLogLevel
from mininet.net import Mininet
from mininet.node import Controller
from mininet.node import Host
from mininet.topo import Topo
from mininet.util import dumpNodeConnections, pmonitor

FAUCET_DIR = '../src/ryu_faucet/org/onfsdn/faucet'


class VLANHost(Host):

    def config(self, vlan=100, **params):
        """Configure VLANHost according to (optional) parameters:
           vlan: VLAN ID for default interface"""
        r = super(Host, self).config(**params)
        intf = self.defaultIntf()
        self.cmd('ifconfig %s inet 0' % intf)
        self.cmd('vconfig add %s %d' % (intf, vlan))
        self.cmd('ifconfig %s.%d inet %s' % (intf, vlan, params['ip']))
        newName = '%s.%d' % (intf, vlan)
        intf.name = newName
        self.nameToIntf[newName] = intf
        return r


class FAUCET(Controller):

    def __init__(self, name, cdir=FAUCET_DIR,
                 command='ryu-manager faucet.py',
                 cargs='--ofp-tcp-listen-port=%s --verbose',
                 **kwargs):
        Controller.__init__(self, name, cdir=cdir,
                            command=command,
                            cargs=cargs, **kwargs)


class FaucetSwitchTopo(Topo):

    def build(self, n_tagged=0, tagged_vid=100, n_untagged=0):
        switch = self.addSwitch('s1')
        for h in range(n_tagged):
            host = self.addHost('ht_%s' % (h + 1),
                cls=VLANHost, vlan=tagged_vid)
            self.addLink(host, switch)
        for h in range(n_untagged):
            host = self.addHost('hu_%s' % (h + 1))
            self.addLink(host, switch)


class FaucetTest(unittest.TestCase):

    CONFIG = ""

    def setUp(self):
        self.tmpdir = tempfile.mkdtemp()
        os.environ['FAUCET_CONFIG'] = os.path.join(self.tmpdir,
             'faucet.yaml')
        os.environ['FAUCET_LOG'] = os.path.join(self.tmpdir,
             'faucet.log')
        os.environ['FAUCET_EXCEPTION_LOG'] = os.path.join(self.tmpdir,
            'faucet-exception.log')
        open(os.environ['FAUCET_CONFIG'], 'w').write(self.CONFIG)

    def tearDown(self):
        shutil.rmtree(self.tmpdir)


class FaucetUntaggedTest(FaucetTest):

    CONFIG = """
---
dp_id: 0x1
name: "untagged-faucet-1"
hardware: "Allied-Telesis"
interfaces:
    1:
        native_vlan: 100
        description: "b1"
    2:
        native_vlan: 100
        description: "b2"
    3:
        native_vlan: 100
        description: "b3"
    4:
        native_vlan: 100
        description: "b4"
vlans:
    100:
        description: "untagged"
"""

    def setUp(self):
        super(FaucetUntaggedTest, self).setUp()
        self.topo = FaucetSwitchTopo(n_untagged=4)
        self.net = Mininet(self.topo, controller=FAUCET)
        self.net.start()
        dumpNodeConnections(self.net.hosts)
        self.net.waitConnected()

    def test_untagged(self):
        self.assertEquals(0, self.net.pingAll())

    def tearDown(self):
        self.net.stop()
        super(FaucetUntaggedTest, self).tearDown()

<commit_msg>Test that disabling unicast_flood on a VLAN doesn't break forwarding.<commit_after>#!/usr/bin/python

import os
import re
import shutil
import tempfile
import time
import unittest
from mininet.log import setLogLevel
from mininet.net import Mininet
from mininet.node import Controller
from mininet.node import Host
from mininet.topo import Topo
from mininet.util import dumpNodeConnections, pmonitor

FAUCET_DIR = '../src/ryu_faucet/org/onfsdn/faucet'


class VLANHost(Host):

    def config(self, vlan=100, **params):
        """Configure VLANHost according to (optional) parameters:
           vlan: VLAN ID for default interface"""
        r = super(Host, self).config(**params)
        intf = self.defaultIntf()
        self.cmd('ifconfig %s inet 0' % intf)
        self.cmd('vconfig add %s %d' % (intf, vlan))
        self.cmd('ifconfig %s.%d inet %s' % (intf, vlan, params['ip']))
        newName = '%s.%d' % (intf, vlan)
        intf.name = newName
        self.nameToIntf[newName] = intf
        return r


class FAUCET(Controller):

    def __init__(self, name, cdir=FAUCET_DIR,
                 command='ryu-manager faucet.py',
                 cargs='--ofp-tcp-listen-port=%s --verbose',
                 **kwargs):
        Controller.__init__(self, name, cdir=cdir,
                            command=command,
                            cargs=cargs, **kwargs)


class FaucetSwitchTopo(Topo):

    def build(self, n_tagged=0, tagged_vid=100, n_untagged=0):
        switch = self.addSwitch('s1')
        for h in range(n_tagged):
            host = self.addHost('ht_%s' % (h + 1),
                cls=VLANHost, vlan=tagged_vid)
            self.addLink(host, switch)
        for h in range(n_untagged):
            host = self.addHost('hu_%s' % (h + 1))
            self.addLink(host, switch)


class FaucetTest(unittest.TestCase):

    CONFIG = ""

    def setUp(self):
        self.tmpdir = tempfile.mkdtemp()
        os.environ['FAUCET_CONFIG'] = os.path.join(self.tmpdir,
             'faucet.yaml')
        os.environ['FAUCET_LOG'] = os.path.join(self.tmpdir,
             'faucet.log')
        os.environ['FAUCET_EXCEPTION_LOG'] = os.path.join(self.tmpdir,
            'faucet-exception.log')
        open(os.environ['FAUCET_CONFIG'], 'w').write(self.CONFIG)

    def tearDown(self):
        shutil.rmtree(self.tmpdir)


class FaucetUntaggedTest(FaucetTest):

    CONFIG = """
---
dp_id: 0x1
name: "untagged-faucet-1"
hardware: "Allied-Telesis"
interfaces:
    1:
        native_vlan: 100
        description: "b1"
    2:
        native_vlan: 100
        description: "b2"
    3:
        native_vlan: 100
        description: "b3"
    4:
        native_vlan: 100
        description: "b4"
vlans:
    100:
        description: "untagged"
"""

    def setUp(self):
        super(FaucetUntaggedTest, self).setUp()
        self.topo = FaucetSwitchTopo(n_untagged=4)
        self.net = Mininet(self.topo, controller=FAUCET)
        self.net.start()
        dumpNodeConnections(self.net.hosts)
        self.net.waitConnected()

    def test_untagged(self):
        self.assertEquals(0, self.net.pingAll())

    def tearDown(self):
        self.net.stop()
        super(FaucetUntaggedTest, self).tearDown()


class FaucetUntaggedNoVLanUnicaseFloodTest(FaucetUntaggedTest):

    CONFIG = """
---
dp_id: 0x1
name: "untagged-faucet-1"
hardware: "Allied-Telesis"
interfaces:
    1:
        native_vlan: 100
        description: "b1"
    2:
        native_vlan: 100
        description: "b2"
    3:
        native_vlan: 100
        description: "b3"
    4:
        native_vlan: 100
        description: "b4"
vlans:
    100:
        description: "untagged"
        unicast_flood: False
"""

    def test_untagged(self):
        self.assertEquals(0, self.net.pingAll())


{'fraction_non_alphanumeric': 0.1247116493656286, 'fraction_numerical': 0.015426758938869665, 'mean_word_length': 2.370748299319728, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4796529', 'n_tokens_mistral': 2506, 'n_tokens_neox': 2348, 'n_words': 479}