-
Notifications
You must be signed in to change notification settings - Fork 7
/
evalConfig.py
162 lines (130 loc) · 4.75 KB
/
evalConfig.py
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
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from ktoolbox import common
from ktoolbox.common import StructParseBase
from ktoolbox.common import strict_dataclass
import testType
from tftbase import TestCaseType
from tftbase import TestType
@strict_dataclass
@dataclass(frozen=True, kw_only=True)
class TestItem(StructParseBase):
threshold: float
@staticmethod
def parse(yamlidx: int, yamlpath: str, arg: Any) -> "TestItem":
with common.structparse_with_strdict(arg, yamlpath) as varg:
threshold = common.structparse_pop_float(*varg.for_key("threshold"))
return TestItem(
yamlidx=yamlidx,
yamlpath=yamlpath,
threshold=threshold,
)
def serialize(self) -> dict[str, Any]:
return {"threshold": self.threshold}
@strict_dataclass
@dataclass(frozen=True, kw_only=True)
class TestCaseData(StructParseBase):
test_case_type: TestCaseType
normal: TestItem
reverse: TestItem
def get_threshold(self, *, is_reverse: bool) -> float:
if is_reverse:
return self.reverse.threshold
return self.normal.threshold
@staticmethod
def parse(yamlidx: int, yamlpath: str, arg: Any) -> "TestCaseData":
with common.structparse_with_strdict(arg, yamlpath) as varg:
test_case_type = common.structparse_pop_enum(
*varg.for_key("id"),
enum_type=TestCaseType,
)
normal = common.structparse_pop_obj(
*varg.for_key("Normal"),
construct=TestItem.parse,
)
reverse = common.structparse_pop_obj(
*varg.for_key("Reverse"),
construct=TestItem.parse,
)
return TestCaseData(
yamlidx=yamlidx,
yamlpath=yamlpath,
test_case_type=test_case_type,
normal=normal,
reverse=reverse,
)
def serialize(self) -> dict[str, Any]:
return {
"id": self.test_case_type.name,
"Normal": self.normal.serialize(),
"Reverse": self.reverse.serialize(),
}
@strict_dataclass
@dataclass(frozen=True, kw_only=True)
class TestTypeData(StructParseBase):
test_type: TestType
test_cases: Mapping[TestCaseType, TestCaseData]
test_type_handler: testType.TestTypeHandler
@staticmethod
def parse(yamlidx: int, yamlpath_base: str, key: Any, arg: Any) -> "TestTypeData":
try:
test_type = common.enum_convert(TestType, key)
except Exception:
raise ValueError(
f'"{yamlpath_base}.[{yamlidx}]": expects a test type as key like iperf-tcp but got {key}'
)
yamlpath = f"{yamlpath_base}.{test_type.name}"
try:
test_type_handler = testType.TestTypeHandler.get(test_type)
except Exception:
raise ValueError(
f'"{yamlpath}": test type "{test_type}" has no handler implementation'
)
if not isinstance(arg, list):
raise ValueError(f'"{yamlpath}": expects a list of test cases')
test_cases = {}
for yamlidx2, arg2 in enumerate(arg):
c = TestCaseData.parse(yamlidx2, f"{yamlpath}[{yamlidx}]", arg2)
if c.test_case_type in test_cases:
raise ValueError(
f'"{yamlpath}[{yamlidx2}]": duplicate key {c.test_case_type.name}'
)
test_cases[c.test_case_type] = c
return TestTypeData(
yamlidx=yamlidx,
yamlpath=yamlpath,
test_type=test_type,
test_type_handler=test_type_handler,
test_cases=test_cases,
)
def serialize(self) -> list[Any]:
return [v.serialize() for k, v in self.test_cases.items()]
@strict_dataclass
@dataclass(frozen=True, kw_only=True)
class Config(StructParseBase):
configs: Mapping[TestType, TestTypeData]
@staticmethod
def parse(arg: Any) -> "Config":
yamlpath = ""
vdict = common.structparse_check_strdict(arg, yamlpath)
configs = {}
for yamlidx2, key in enumerate(vdict):
c = TestTypeData.parse(
yamlidx=yamlidx2,
yamlpath_base=yamlpath,
key=key,
arg=vdict[key],
)
if c.test_type in configs:
raise ValueError(
f'"{yamlpath}.[{yamlidx2}]": duplicate key {c.test_type.name}'
)
configs[c.test_type] = c
return Config(
yamlidx=0,
yamlpath=yamlpath,
configs=configs,
)
def serialize(self) -> dict[str, Any]:
return {k.name: v.serialize() for k, v in self.configs.items()}