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
|
# This file is part of cloud-init. See LICENSE file for license information.
from datetime import datetime
from unittest import mock
import pytest
from cloudinit import version
from cloudinit.sources.azure import errors, kvp
@pytest.fixture()
def fake_utcnow():
timestamp = datetime.utcnow()
with mock.patch.object(kvp, "datetime", autospec=True) as m:
m.utcnow.return_value = timestamp
yield timestamp
@pytest.fixture()
def fake_vm_id():
vm_id = "fake-vm-id"
with mock.patch.object(kvp.identity, "query_vm_id", autospec=True) as m:
m.return_value = vm_id
yield vm_id
@pytest.fixture
def telemetry_reporter(tmp_path):
kvp_file_path = tmp_path / "kvp_pool_file"
kvp_file_path.write_bytes(b"")
reporter = kvp.handlers.HyperVKvpReportingHandler(
kvp_file_path=str(kvp_file_path)
)
kvp.instantiated_handler_registry.register_item("telemetry", reporter)
yield reporter
kvp.instantiated_handler_registry.unregister_item("telemetry")
class TestReportFailureToHost:
def test_report_failure_to_host(self, caplog, telemetry_reporter):
error = errors.ReportableError(reason="test")
assert kvp.report_failure_to_host(error) is True
assert (
"KVP handler not enabled, skipping host report." not in caplog.text
)
report = {
"key": "PROVISIONING_REPORT",
"value": error.as_encoded_report(),
}
assert report in list(telemetry_reporter._iterate_kvps(0))
def test_report_skipped_without_telemetry(self, caplog):
error = errors.ReportableError(reason="test")
assert kvp.report_failure_to_host(error) is False
assert "KVP handler not enabled, skipping host report." in caplog.text
class TestReportSuccessToHost:
def test_report_success_to_host(
self, caplog, fake_utcnow, fake_vm_id, telemetry_reporter
):
assert kvp.report_success_to_host() is True
assert (
"KVP handler not enabled, skipping host report." not in caplog.text
)
report_value = errors.encode_report(
[
"result=success",
f"agent=Cloud-Init/{version.version_string()}",
f"timestamp={fake_utcnow.isoformat()}",
f"vm_id={fake_vm_id}",
]
)
report = {
"key": "PROVISIONING_REPORT",
"value": report_value,
}
assert report in list(telemetry_reporter._iterate_kvps(0))
def test_report_skipped_without_telemetry(self, caplog):
assert kvp.report_success_to_host() is False
assert "KVP handler not enabled, skipping host report." in caplog.text
|
