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
|
from ipaddress import IPv4Network
from vpp_object import VppObject
from vpp_papi import VppEnum
from vpp_ip import INVALID_INDEX
from vpp_papi_provider import UnexpectedApiReturnValueError
class SessionSdl:
"""Session SDL"""
def __init__(
self,
rmt,
action_index,
tag,
):
self.action_index = action_index
self.rmt = rmt
self.tag = tag
def encode(self):
return {
"rmt": self.rmt,
"action_index": self.action_index,
"tag": self.tag,
}
class VppSessionSdl(VppObject):
"""VPP Session SDL"""
def __init__(self, test, rules, is_add, appns_index):
self._test = test
self._rules = rules
self.is_add = is_add
self.appns_index = appns_index
@property
def rules(self):
return self._rules
@property
def count(self):
return len(self._rules)
def encode_rules(self):
rules = []
for rule in self._rules:
rules.append(rule.encode())
return rules
def add_vpp_config(self, expect_error=False):
try:
reply = self._test.vapi.session_sdl_add_del_v2(
is_add=self.is_add,
appns_index=self.appns_index,
count=self.count,
r=self.encode_rules(),
)
self._test.registry.register(self, self._test.logger)
if expect_error:
self._test.fail("Unexpected api reply")
return self
except UnexpectedApiReturnValueError:
if not expect_error:
self._test.fail("Unexpected api reply")
return None
def query_vpp_config(self):
pass
def remove_vpp_config(self, expect_error=False):
pass
|
