Coverage for mlos_bench/mlos_bench/tests/environments/local/local_env_telemetry_test.py: 100%
51 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-14 00:55 +0000
1#
2# Copyright (c) Microsoft Corporation.
3# Licensed under the MIT License.
4#
5"""Unit tests for telemetry and status of LocalEnv benchmark environment."""
6from datetime import datetime, timedelta, tzinfo
8import pytest
9from pytz import UTC
11from mlos_bench.tests import ZONE_INFO
12from mlos_bench.tests.environments import check_env_fail_telemetry, check_env_success
13from mlos_bench.tests.environments.local import create_local_env
14from mlos_bench.tunables.tunable_groups import TunableGroups
17def _format_str(zone_info: tzinfo | None) -> str:
18 if zone_info is not None:
19 return "%Y-%m-%d %H:%M:%S.%f %z"
20 return "%Y-%m-%d %H:%M:%S.%f"
23# FIXME: This fails with zone_info = None when run with `TZ="America/Chicago pytest -n0 ...`
24@pytest.mark.parametrize(("zone_info"), ZONE_INFO)
25def test_local_env_telemetry(tunable_groups: TunableGroups, zone_info: tzinfo | None) -> None:
26 """Produce benchmark and telemetry data in a local script and read it."""
27 ts1 = datetime.now(zone_info)
28 ts2 = ts1 + timedelta(minutes=1)
30 format_str = _format_str(zone_info)
31 time_str1 = ts1.strftime(format_str)
32 time_str2 = ts2.strftime(format_str)
34 local_env = create_local_env(
35 tunable_groups,
36 {
37 "run": [
38 "echo 'metric,value' > output.csv",
39 "echo 'latency,4.1' >> output.csv",
40 "echo 'throughput,512' >> output.csv",
41 "echo 'score,0.95' >> output.csv",
42 "echo '-------------------'", # This output does not go anywhere
43 "echo 'timestamp,metric,value' > telemetry.csv",
44 f"echo {time_str1},cpu_load,0.65 >> telemetry.csv",
45 f"echo {time_str1},mem_usage,10240 >> telemetry.csv",
46 f"echo {time_str2},cpu_load,0.8 >> telemetry.csv",
47 f"echo {time_str2},mem_usage,20480 >> telemetry.csv",
48 ],
49 "read_results_file": "output.csv",
50 "read_telemetry_file": "telemetry.csv",
51 },
52 )
54 check_env_success(
55 local_env,
56 tunable_groups,
57 expected_results={
58 "latency": 4.1,
59 "throughput": 512.0,
60 "score": 0.95,
61 },
62 expected_telemetry=[
63 (ts1.astimezone(UTC), "cpu_load", 0.65),
64 (ts1.astimezone(UTC), "mem_usage", 10240.0),
65 (ts2.astimezone(UTC), "cpu_load", 0.8),
66 (ts2.astimezone(UTC), "mem_usage", 20480.0),
67 ],
68 )
71# FIXME: This fails with zone_info = None when run with `TZ="America/Chicago pytest -n0 ...`
72@pytest.mark.parametrize(("zone_info"), ZONE_INFO)
73def test_local_env_telemetry_no_header(
74 tunable_groups: TunableGroups,
75 zone_info: tzinfo | None,
76) -> None:
77 """Read the telemetry data with no header."""
78 ts1 = datetime.now(zone_info)
79 ts2 = ts1 + timedelta(minutes=1)
81 format_str = _format_str(zone_info)
82 time_str1 = ts1.strftime(format_str)
83 time_str2 = ts2.strftime(format_str)
85 local_env = create_local_env(
86 tunable_groups,
87 {
88 "run": [
89 f"echo {time_str1},cpu_load,0.65 > telemetry.csv",
90 f"echo {time_str1},mem_usage,10240 >> telemetry.csv",
91 f"echo {time_str2},cpu_load,0.8 >> telemetry.csv",
92 f"echo {time_str2},mem_usage,20480 >> telemetry.csv",
93 ],
94 "read_telemetry_file": "telemetry.csv",
95 },
96 )
98 check_env_success(
99 local_env,
100 tunable_groups,
101 expected_results={},
102 expected_telemetry=[
103 (ts1.astimezone(UTC), "cpu_load", 0.65),
104 (ts1.astimezone(UTC), "mem_usage", 10240.0),
105 (ts2.astimezone(UTC), "cpu_load", 0.8),
106 (ts2.astimezone(UTC), "mem_usage", 20480.0),
107 ],
108 )
111@pytest.mark.filterwarnings(
112 "ignore:.*(Could not infer format, so each element will be parsed individually, "
113 "falling back to `dateutil`).*:UserWarning::0"
114) # pylint: disable=line-too-long # noqa
115@pytest.mark.parametrize(("zone_info"), ZONE_INFO)
116def test_local_env_telemetry_wrong_header(
117 tunable_groups: TunableGroups,
118 zone_info: tzinfo | None,
119) -> None:
120 """Read the telemetry data with incorrect header."""
121 ts1 = datetime.now(zone_info)
122 ts2 = ts1 + timedelta(minutes=1)
124 format_str = _format_str(zone_info)
125 time_str1 = ts1.strftime(format_str)
126 time_str2 = ts2.strftime(format_str)
128 local_env = create_local_env(
129 tunable_groups,
130 {
131 "run": [
132 # Error: the data is correct, but the header has unexpected column names
133 "echo 'ts,metric_name,metric_value' > telemetry.csv",
134 f"echo {time_str1},cpu_load,0.65 >> telemetry.csv",
135 f"echo {time_str1},mem_usage,10240 >> telemetry.csv",
136 f"echo {time_str2},cpu_load,0.8 >> telemetry.csv",
137 f"echo {time_str2},mem_usage,20480 >> telemetry.csv",
138 ],
139 "read_telemetry_file": "telemetry.csv",
140 },
141 )
143 check_env_fail_telemetry(local_env, tunable_groups)
146def test_local_env_telemetry_invalid(tunable_groups: TunableGroups) -> None:
147 """Fail when the telemetry data has wrong format."""
148 zone_info = UTC
149 ts1 = datetime.now(zone_info)
150 ts2 = ts1 + timedelta(minutes=1)
152 format_str = _format_str(zone_info)
153 time_str1 = ts1.strftime(format_str)
154 time_str2 = ts2.strftime(format_str)
156 local_env = create_local_env(
157 tunable_groups,
158 {
159 "run": [
160 # Error: too many columns
161 f"echo {time_str1},EXTRA,cpu_load,0.65 > telemetry.csv",
162 f"echo {time_str1},EXTRA,mem_usage,10240 >> telemetry.csv",
163 f"echo {time_str2},EXTRA,cpu_load,0.8 >> telemetry.csv",
164 f"echo {time_str2},EXTRA,mem_usage,20480 >> telemetry.csv",
165 ],
166 "read_telemetry_file": "telemetry.csv",
167 },
168 )
170 check_env_fail_telemetry(local_env, tunable_groups)
173def test_local_env_telemetry_invalid_ts(tunable_groups: TunableGroups) -> None:
174 """Fail when the telemetry data has wrong format."""
175 local_env = create_local_env(
176 tunable_groups,
177 {
178 "run": [
179 # Error: field 1 must be a timestamp
180 "echo 1,cpu_load,0.65 > telemetry.csv",
181 "echo 2,mem_usage,10240 >> telemetry.csv",
182 "echo 3,cpu_load,0.8 >> telemetry.csv",
183 "echo 4,mem_usage,20480 >> telemetry.csv",
184 ],
185 "read_telemetry_file": "telemetry.csv",
186 },
187 )
189 check_env_fail_telemetry(local_env, tunable_groups)