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
|
# SPDX-FileCopyrightText: Copyright 2022, Arm Limited and/or its affiliates.
# SPDX-License-Identifier: Apache-2.0
"""Tests for the output parsing."""
from __future__ import annotations
import base64
import json
from typing import Any
import pytest
from mlia.backend.output_consumer import Base64OutputConsumer
from mlia.backend.output_consumer import OutputConsumer
OUTPUT_MATCH_ALL = bytearray(
"""
String1: My awesome string!
String2: STRINGS_ARE_GREAT!!!
Int: 12
Float: 3.14
""",
encoding="utf-8",
)
OUTPUT_NO_MATCH = bytearray(
"""
This contains no matches...
Test1234567890!"£$%^&*()_+@~{}[]/.,<>?|
""",
encoding="utf-8",
)
OUTPUT_PARTIAL_MATCH = bytearray(
"String1: My awesome string!",
encoding="utf-8",
)
REGEX_CONFIG = {
"FirstString": {"pattern": r"String1.*: (.*)", "type": "str"},
"SecondString": {"pattern": r"String2.*: (.*)!!!", "type": "str"},
"IntegerValue": {"pattern": r"Int.*: (.*)", "type": "int"},
"FloatValue": {"pattern": r"Float.*: (.*)", "type": "float"},
}
EMPTY_REGEX_CONFIG: dict[str, dict[str, Any]] = {}
EXPECTED_METRICS_ALL = {
"FirstString": "My awesome string!",
"SecondString": "STRINGS_ARE_GREAT",
"IntegerValue": 12,
"FloatValue": 3.14,
}
EXPECTED_METRICS_PARTIAL = {
"FirstString": "My awesome string!",
}
@pytest.mark.parametrize(
"expected_metrics",
[
EXPECTED_METRICS_ALL,
EXPECTED_METRICS_PARTIAL,
],
)
def test_base64_output_consumer(expected_metrics: dict) -> None:
"""
Make sure the Base64OutputConsumer yields valid results.
I.e. return an empty dict if either the input or the config is empty and
return the parsed metrics otherwise.
"""
parser = Base64OutputConsumer()
assert isinstance(parser, OutputConsumer)
def create_base64_output(expected_metrics: dict) -> bytearray:
json_str = json.dumps(expected_metrics, indent=4)
json_b64 = base64.b64encode(json_str.encode("utf-8"))
return (
OUTPUT_MATCH_ALL # Should not be matched by the Base64OutputConsumer
+ f"<{Base64OutputConsumer.TAG_NAME}>".encode()
+ bytearray(json_b64)
+ f"</{Base64OutputConsumer.TAG_NAME}>".encode()
+ OUTPUT_NO_MATCH # Just to add some difficulty...
)
output = create_base64_output(expected_metrics)
consumed = False
for line in output.splitlines():
if parser.feed(line.decode("utf-8")):
consumed = True
assert consumed # we should have consumed at least one line
res = parser.parsed_output
assert len(res) == 1
assert isinstance(res, list)
for val in res:
assert val == expected_metrics
|
