File: test_flow_mod.py

package info (click to toggle)
python-openflow 2021.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,224 kB
  • sloc: python: 6,906; sh: 4; makefile: 4
file content (58 lines) | stat: -rw-r--r-- 1,925 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
"""Flow modification (add/delete) message tests."""
from pyof.v0x01.common.action import ActionOutput
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.unit.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)


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)
        super().set_minimum_size(72)


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,
            'actions': _get_actions()}


def _get_match():
    """Return a Match object."""
    return Match()


def _get_actions():
    """Return a List of actions registered by flow object."""
    action = ActionOutput(port=65533, max_length=65535)
    return [action]