Newer
Older
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
import logging
import time
import pytest
from telemetry.backend.drivers.emulated.EmulatedDriver import EmulatedDriver
from telemetry.backend.tests.messages_emulated import (
create_test_configuration,
create_specific_config_keys,
create_config_for_delete,
create_test_subscriptions,
)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@pytest.fixture
def setup_driver():
"""Sets up an EmulatedDriver instance for testing."""
yield EmulatedDriver(address="127.0.0.1", port=8080)
@pytest.fixture
def connected_configured_driver(setup_driver):
driver = setup_driver # EmulatedDriver(address="127.0.0.1", port=8080)
driver.Connect()
driver.SetConfig(create_test_configuration())
yield driver
driver.Disconnect()
def test_connect(setup_driver):
logger.info(">>> test_connect <<<")
driver = setup_driver
assert driver.Connect() is True
assert driver.connected is True
def test_disconnect(setup_driver):
logger.info(">>> test_disconnect <<<")
driver = setup_driver
driver.Connect()
assert driver.Disconnect() is True
assert driver.connected is False
def test_set_config(setup_driver):
logger.info(">>> test_set_config <<<")
driver = setup_driver
driver.Connect()
config = create_test_configuration()
results = driver.SetConfig(config)
assert all(result is True for result in results)
def test_get_config(connected_configured_driver):
logger.info(">>> test_get_config <<<")
resource_keys = create_specific_config_keys()
results = connected_configured_driver.GetConfig(resource_keys)
for key, value in results:
assert key in create_specific_config_keys()
assert value is not None
def test_delete_config(connected_configured_driver):
logger.info(">>> test_delete_config <<<")
resource_keys = create_config_for_delete()
results = connected_configured_driver.DeleteConfig(resource_keys)
assert all(result is True for result in results)
def test_subscribe_state(connected_configured_driver):
logger.info(">>> test_subscribe_state <<<")
subscriptions = create_test_subscriptions()
results = connected_configured_driver.SubscribeState(subscriptions)
assert all(result is True for result in results)
def test_unsubscribe_state(connected_configured_driver):
logger.info(">>> test_unsubscribe_state <<<")
subscriptions = create_test_subscriptions()
connected_configured_driver.SubscribeState(subscriptions)
results = connected_configured_driver.UnsubscribeState(subscriptions)
assert all(result is True for result in results)
def test_get_state(connected_configured_driver):
logger.info(">>> test_get_state <<<")
subscriptions = create_test_subscriptions()
connected_configured_driver.SubscribeState(subscriptions)
logger.info(f"Subscribed to state: {subscriptions}. waiting for 3 seconds ...")
time.sleep(3)
state_iterator = connected_configured_driver.GetState(blocking=False)
states = list(state_iterator)
assert len(states) > 0