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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
import errno
import time
import pytest
from pr2test.context_manager import make_test_matrix, skip_if_not_supported
from pr2test.marks import require_root
from pyroute2 import NetlinkError
pytestmark = [require_root()]
wait_timeout = 30
test_matrix = make_test_matrix(targets=['local', 'netns'])
@pytest.mark.parametrize('context', test_matrix, indirect=True)
@skip_if_not_supported
def test_addr_add(context):
index, ifname = context.default_interface
ipaddr = context.new_ipaddr
ipr = context.ipr
ndb = context.ndb
ipr.addr('add', index=index, address=ipaddr, prefixlen=24)
ndb.addresses.wait(index=index, address=ipaddr, timeout=wait_timeout)
@pytest.mark.parametrize('context', test_matrix, indirect=True)
@skip_if_not_supported
def test_addr_replace(context):
index, ifname = context.default_interface
ipaddr1 = context.new_ipaddr
ipaddr2 = context.new_ipaddr
ipr = context.ipr
ndb = context.ndb
ipr.addr('add', index=index, address=ipaddr1, prefixlen=24)
ndb.addresses.wait(index=index, address=ipaddr1, timeout=wait_timeout)
ipr.addr('replace', index=index, address=ipaddr2, prefixlen=24)
ndb.addresses.wait(index=index, address=ipaddr2, timeout=wait_timeout)
@pytest.mark.parametrize('context', test_matrix, indirect=True)
@skip_if_not_supported
def test_addr_add_local(context):
index, ifname = context.default_interface
ipaddr1 = context.new_ipaddr
ipaddr2 = context.new_ipaddr
ipr = context.ipr
ndb = context.ndb
ipr.addr('add', index=index, address=ipaddr1, local=ipaddr2, prefixlen=24)
ndb.addresses.wait(
index=index, address=ipaddr1, local=ipaddr2, timeout=wait_timeout
)
@pytest.mark.parametrize('context', test_matrix, indirect=True)
@skip_if_not_supported
def test_addr_add_broadcast(context):
index, ifname = context.default_interface
ipaddr1 = context.new_ipaddr
ipaddr2 = context.new_ipaddr
ipr = context.ipr
ndb = context.ndb
ipr.addr(
'add', index=index, address=ipaddr1, broadcast=ipaddr2, prefixlen=24
)
ndb.addresses.wait(
index=index, address=ipaddr1, broadcast=ipaddr2, timeout=wait_timeout
)
@pytest.mark.parametrize('context', test_matrix, indirect=True)
@skip_if_not_supported
def test_addr_add_broadcast_default(context):
index, ifname = context.default_interface
ipaddr = context.new_ipaddr
ipr = context.ipr
ndb = context.ndb
ipr.addr('add', index=index, address=ipaddr, broadcast=True, prefixlen=24)
interface = ndb.addresses.wait(
index=index, address=ipaddr, timeout=wait_timeout
)
assert interface['broadcast'] is not None
@pytest.mark.parametrize('context', test_matrix, indirect=True)
@skip_if_not_supported
def test_addr_filter(context):
index, ifname = context.default_interface
ipaddr1 = context.new_ipaddr
ipaddr2 = context.new_ipaddr
ipaddrB = context.new_ipaddr
ipr = context.ipr
ndb = context.ndb
ipr.addr(
'add', index=index, address=ipaddr1, broadcast=ipaddrB, prefixlen=24
)
ipr.addr(
'add', index=index, address=ipaddr2, broadcast=ipaddrB, prefixlen=24
)
ndb.addresses.wait(index=index, address=ipaddr1, timeout=wait_timeout)
ndb.addresses.wait(index=index, address=ipaddr2, timeout=wait_timeout)
assert (
len(tuple(ipr.get_addr(index=index))) >= 2
) # remember link-local IPv6
assert len(tuple(ipr.get_addr(address=ipaddr1))) == 1
assert len(tuple(ipr.get_addr(broadcast=ipaddrB))) == 2
assert len(tuple(ipr.get_addr(match=lambda x: x['index'] == index))) >= 2
@pytest.mark.parametrize('context', test_matrix, indirect=True)
@skip_if_not_supported
def test_addr_flush(context):
index, ifname = context.default_interface
addresses = [
context.new_ipaddr,
context.new_ipaddr,
context.new_ipaddr,
context.new_ipaddr,
]
ipr = context.ipr
ndb = context.ndb
counter = 5
for ipaddr in addresses:
ipr.addr('add', index=index, address=ipaddr, prefixlen=24)
for ipaddr in addresses:
ndb.addresses.wait(index=index, address=ipaddr, timeout=wait_timeout)
ipr.flush_addr(index=index)
while counter:
for ipaddr in tuple(addresses):
if ipaddr not in ndb.addresses:
addresses.remove(ipaddr)
if not addresses:
break
time.sleep(1)
counter -= 1
else:
raise Exception()
@pytest.mark.parametrize('context', test_matrix, indirect=True)
def test_fail_no_such_device(context):
ifaddr = context.new_ipaddr
index = sorted([i['index'] for i in context.ipr.get_links()])[-1] + 10
with pytest.raises(NetlinkError) as e:
context.ipr.addr('add', index=index, address=ifaddr, prefixlen=24)
assert e.value.code == errno.ENODEV
@pytest.mark.parametrize('context', test_matrix, indirect=True)
def test_symbolic_flags(context):
ipaddr = context.new_ipaddr
index, ifname = context.default_interface
context.ipr.link('set', index=index, state='up')
context.ipr.addr('add', index=index, address=ipaddr, prefixlen=24)
addr = [
x for x in context.ipr.get_addr() if x.get_attr('IFA_LOCAL') == ipaddr
][0]
assert 'IFA_F_PERMANENT' in addr.flags2names(addr['flags'])
|
