Newer
Older
# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
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
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