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
|
import struct
import pytest
from pyroute2 import IPRoute
from pyroute2.netlink import NLM_F_DUMP, NLM_F_REQUEST
from pyroute2.netlink.rtnl import (
RTM_GETLINK,
RTM_GETROUTE,
RTM_NEWLINK,
RTM_NEWROUTE,
)
@pytest.fixture
def ipr():
with IPRoute() as iproute:
yield iproute
test_config = (
'name,argv,kwarg,msg_type,msg_flags',
(
(
'link',
('get',),
{'index': 1},
(RTM_GETLINK, RTM_NEWLINK),
NLM_F_REQUEST,
),
(
'link',
('dump',),
{},
(RTM_GETLINK, RTM_NEWLINK),
NLM_F_DUMP | NLM_F_REQUEST,
),
(
'route',
('dump',),
{},
(RTM_GETROUTE, RTM_NEWROUTE),
NLM_F_DUMP | NLM_F_REQUEST,
),
),
)
@pytest.mark.parametrize(*test_config)
def test_compile_call(ipr, name, argv, kwarg, msg_type, msg_flags):
compiler_context = ipr.compile()
data = getattr(ipr, name)(*argv, **kwarg)
assert msg_type[0], msg_flags == struct.unpack_from(
'HH', data[0], offset=4
)
compiler_context.close()
assert ipr.compiled is None
for msg in getattr(ipr, name)(*argv, **kwarg):
assert msg['header']['type'] == msg_type[1]
@pytest.mark.parametrize(*test_config)
def test_compile_context_manager(ipr, name, argv, kwarg, msg_type, msg_flags):
with ipr.compile():
data = getattr(ipr, name)(*argv, **kwarg)
assert msg_type[0], msg_flags == struct.unpack_from(
'HH', data[0], offset=4
)
assert ipr.compiled is None
for msg in getattr(ipr, name)(*argv, **kwarg):
assert msg['header']['type'] == msg_type[1]
|
