File: test_flow_mod.py

package info (click to toggle)
python-openflow 1.1.0~alpha2-1
  • links: PTS
  • area: main
  • in suites: stretch
  • size: 692 kB
  • ctags: 1,027
  • sloc: python: 2,713; makefile: 207; sh: 8
file content (72 lines) | stat: -rw-r--r-- 2,281 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
"""Flow modification (add/delete) message tests."""
import unittest

from pyof.v0x01.common.flow_match import Match
from pyof.v0x01.common.phy_port import Port
from pyof.v0x01.controller2switch.flow_mod import FlowMod, FlowModCommand
from tests.test_struct import TestStruct


class TestFlowAdd(TestStruct):
    """Flow addition message tests (also those in :class:`.TestDump`)."""

    @classmethod
    def setUpClass(cls):
        """Configure raw file and its object in parent class (TestDump)."""
        super().setUpClass()
        super().set_raw_dump_file('v0x01', 'ofpt_flow_add')
        kwargs = _get_flowmod_kwargs(FlowModCommand.OFPFC_ADD)
        super().set_raw_dump_object(FlowMod, **kwargs)
        super().set_minimum_size(72)

    @unittest.skip('Need to recover dump contents.')
    def test_pack(self):
        pass

    @unittest.skip('Need to recover dump contents.')
    def test_unpack(self):
        pass


class TestFlowDelete(TestStruct):
    """Flow deletion message tests (also those in :class:`.TestDump`)."""

    @classmethod
    def setUpClass(cls):
        """Configure raw file and its object in parent class (TestDump)."""
        super().setUpClass()
        super().set_raw_dump_file('v0x01', 'ofpt_flow_delete')
        kwargs = _get_flowmod_kwargs(FlowModCommand.OFPFC_DELETE)
        super().set_raw_dump_object(FlowMod, **kwargs)
        # No need to test minimum size again.


def _get_flowmod_kwargs(command):
    """Return parameters for FlowMod object."""
    return {'xid': 4,
            'command': command,
            'match': _get_match(),
            'cookie': 0,
            'idle_timeout': 0,
            'hard_timeout': 0,
            'priority': 32768,
            'buffer_id': 4294967295,
            'out_port': Port.OFPP_NONE,
            'flags': 0}


def _get_match():
    """Return a Match object."""
    return Match(wildcards=0x000000000010001f,
                 in_port=0,
                 dl_src='00:00:00:00:00:00',
                 dl_dst='00:00:00:00:00:00',
                 dl_vlan=0,
                 dl_vlan_pcp=0,
                 dl_type=0,
                 nw_tos=0,
                 nw_proto=0,
                 nw_src=0,
                 nw_dst=0,
                 tp_src=0,
                 tp_dst=0)