From 100d0c6de58c30b2e27e69a596f51285a7d0b80f Mon Sep 17 00:00:00 2001 From: Guillaume Belanger Date: Fri, 11 Oct 2024 15:33:37 -0400 Subject: [PATCH] chore: use built-in testing from ops (#396) Signed-off-by: Guillaume Belanger --- tests/unit/fixtures.py | 4 +- .../sdcore_upf/v0/test_fiveg_n3_provider.py | 16 +- .../sdcore_upf/v0/test_fiveg_n3_requirer.py | 12 +- .../sdcore_upf/v0/test_fiveg_n4_provider.py | 12 +- .../sdcore_upf/v0/test_fiveg_n4_requirer.py | 12 +- tests/unit/test_charm_bessd_pebble_ready.py | 100 +++++----- tests/unit/test_charm_collect_status.py | 187 +++++++++--------- tests/unit/test_charm_config_changed.py | 22 +-- .../test_charm_pfcp_agent_pebble_ready.py | 12 +- tests/unit/test_charm_remove.py | 4 +- 10 files changed, 191 insertions(+), 190 deletions(-) diff --git a/tests/unit/fixtures.py b/tests/unit/fixtures.py index c47e1685..36036d67 100644 --- a/tests/unit/fixtures.py +++ b/tests/unit/fixtures.py @@ -4,7 +4,7 @@ from unittest.mock import patch import pytest -import scenario +from ops import testing from charm import UPFOperatorCharm from dpdk import DPDK @@ -52,6 +52,6 @@ def teardown() -> None: @pytest.fixture(autouse=True) def context(self): - self.ctx = scenario.Context( + self.ctx = testing.Context( charm_type=UPFOperatorCharm, ) diff --git a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_provider.py b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_provider.py index 26a22811..5a7f4d55 100644 --- a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_provider.py +++ b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_provider.py @@ -3,8 +3,8 @@ import pytest -import scenario from charms.sdcore_upf_k8s.v0.fiveg_n3 import FiveGN3RequestEvent, N3Provides +from ops import testing from ops.charm import ActionEvent, CharmBase @@ -31,7 +31,7 @@ def _on_publish_upf_information_action(self, event: ActionEvent): class TestN3Provides: @pytest.fixture(autouse=True) def context(self): - self.ctx = scenario.Context( + self.ctx = testing.Context( charm_type=N3Provider, meta={ "name": "n3-provider", @@ -50,11 +50,11 @@ def context(self): def test_given_fiveg_n3_relation_when_set_upf_information_then_info_added_to_relation_data( # noqa: E501 self, ): - fiveg_n3_relation = scenario.Relation( + fiveg_n3_relation = testing.Relation( endpoint="fiveg_n3", interface="fiveg_n3", ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n3_relation], ) @@ -74,11 +74,11 @@ def test_given_fiveg_n3_relation_when_set_upf_information_then_info_added_to_rel def test_given_invalid_upf_information_when_set_upf_information_then_error_raised( self, ): - fiveg_n3_relation = scenario.Relation( + fiveg_n3_relation = testing.Relation( endpoint="fiveg_n3", interface="fiveg_n3", ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n3_relation], ) @@ -96,11 +96,11 @@ def test_given_invalid_upf_information_when_set_upf_information_then_error_raise def test_given_when_relation_joined_then_fiveg_n3_request_event_emitted( self, ): - fiveg_n3_relation = scenario.Relation( + fiveg_n3_relation = testing.Relation( endpoint="fiveg_n3", interface="fiveg_n3", ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n3_relation], ) diff --git a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_requirer.py b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_requirer.py index 81959b28..7803662b 100644 --- a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_requirer.py +++ b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n3_requirer.py @@ -3,8 +3,8 @@ import pytest -import scenario from charms.sdcore_upf_k8s.v0.fiveg_n3 import N3AvailableEvent, N3Requires +from ops import testing from ops.charm import CharmBase @@ -17,7 +17,7 @@ def __init__(self, *args): class TestN3Provides: @pytest.fixture(autouse=True) def context(self): - self.ctx = scenario.Context( + self.ctx = testing.Context( charm_type=N3Requirer, meta={ "name": "n3-requirer", @@ -28,12 +28,12 @@ def context(self): def test_given_upf_ip_address_in_relation_data_when_relation_changed_then_fiveg_n3_request_event_emitted( # noqa: E501 self, ): - fiveg_n3_relation = scenario.Relation( + fiveg_n3_relation = testing.Relation( endpoint="fiveg_n3", interface="fiveg_n3", remote_app_data={"upf_ip_address": "1.2.3.4"}, ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n3_relation], ) @@ -46,12 +46,12 @@ def test_given_upf_ip_address_in_relation_data_when_relation_changed_then_fiveg_ def test_given_upf_ip_address_not_in_relation_data_when_relation_changed_then_fiveg_n3_request_event_emitted( # noqa: E501 self, ): - fiveg_n3_relation = scenario.Relation( + fiveg_n3_relation = testing.Relation( endpoint="fiveg_n3", interface="fiveg_n3", remote_app_data={}, ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n3_relation], ) diff --git a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_provider.py b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_provider.py index 682324f8..505e36d3 100644 --- a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_provider.py +++ b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_provider.py @@ -3,8 +3,8 @@ import pytest -import scenario from charms.sdcore_upf_k8s.v0.fiveg_n4 import FiveGN4RequestEvent, N4Provides +from ops import testing from ops.charm import ActionEvent, CharmBase @@ -34,7 +34,7 @@ def _on_publish_upf_information_action(self, event: ActionEvent): class TestN4Provides: @pytest.fixture(autouse=True) def context(self): - self.ctx = scenario.Context( + self.ctx = testing.Context( charm_type=N4Provider, meta={ "name": "n4-provider", @@ -54,11 +54,11 @@ def context(self): def test_given_fiveg_n4_relation_when_set_upf_information_then_info_added_to_relation_data( # noqa: E501 self, ): - fiveg_n4_relation = scenario.Relation( + fiveg_n4_relation = testing.Relation( endpoint="fiveg_n4", interface="fiveg_n4", ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n4_relation], ) @@ -80,11 +80,11 @@ def test_given_fiveg_n4_relation_when_set_upf_information_then_info_added_to_rel def test_given_when_relation_joined_then_fiveg_n4_request_event_emitted( self, ): - fiveg_n4_relation = scenario.Relation( + fiveg_n4_relation = testing.Relation( endpoint="fiveg_n4", interface="fiveg_n4", ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n4_relation], ) diff --git a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_requirer.py b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_requirer.py index a036a990..3722c061 100644 --- a/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_requirer.py +++ b/tests/unit/lib/charms/sdcore_upf/v0/test_fiveg_n4_requirer.py @@ -3,8 +3,8 @@ import pytest -import scenario from charms.sdcore_upf_k8s.v0.fiveg_n4 import N4AvailableEvent, N4Requires +from ops import testing from ops.charm import CharmBase @@ -17,7 +17,7 @@ def __init__(self, *args): class TestN4Provides: @pytest.fixture(autouse=True) def context(self): - self.ctx = scenario.Context( + self.ctx = testing.Context( charm_type=N4Requirer, meta={ "name": "n4-requirer", @@ -28,7 +28,7 @@ def context(self): def test_given_upf_hostname_in_relation_data_when_relation_changed_then_fiveg_n4_request_event_emitted( # noqa: E501 self, ): - fiveg_n4_relation = scenario.Relation( + fiveg_n4_relation = testing.Relation( endpoint="fiveg_n4", interface="fiveg_n4", remote_app_data={ @@ -36,7 +36,7 @@ def test_given_upf_hostname_in_relation_data_when_relation_changed_then_fiveg_n4 "upf_port": "1234", }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n4_relation], ) @@ -49,12 +49,12 @@ def test_given_upf_hostname_in_relation_data_when_relation_changed_then_fiveg_n4 def test_given_upf_hostname_not_in_relation_data_when_relation_changed_then_fiveg_n4_request_event_emitted( # noqa: E501 self, ): - fiveg_n4_relation = scenario.Relation( + fiveg_n4_relation = testing.Relation( endpoint="fiveg_n4", interface="fiveg_n4", remote_app_data={}, ) - state_in = scenario.State( + state_in = testing.State( leader=True, relations=[fiveg_n4_relation], ) diff --git a/tests/unit/test_charm_bessd_pebble_ready.py b/tests/unit/test_charm_bessd_pebble_ready.py index 6c21d8d6..f96dc1e5 100644 --- a/tests/unit/test_charm_bessd_pebble_ready.py +++ b/tests/unit/test_charm_bessd_pebble_ready.py @@ -5,7 +5,7 @@ import os import tempfile -import scenario +from ops import testing from ops.pebble import Layer, ServiceStatus from tests.unit.fixtures import UPFUnitTestFixtures @@ -20,11 +20,11 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_config_file_is access_gateway_ip = "2.1.1.1" self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, layers={ @@ -32,26 +32,26 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_config_file_is }, service_statuses={"pfcp-agent": ServiceStatus.ACTIVE}, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", mounts={"config": bessd_config_mount}, can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=0, stdout="RUNNING", ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -60,7 +60,7 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_config_file_is ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -71,7 +71,7 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_config_file_is ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -79,7 +79,7 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_config_file_is "access-gateway-ip": access_gateway_ip, "gnb-subnet": gnb_subnet, }, - model=scenario.Model(name="whatever"), + model=testing.Model(name="whatever"), ) self.ctx.run(self.ctx.on.pebble_ready(bessd_container), state_in) @@ -100,11 +100,11 @@ def test_given_config_file_already_written_when_bessd_pebble_ready_then_config_f access_gateway_ip = "2.1.1.1" self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, layers={ @@ -112,26 +112,26 @@ def test_given_config_file_already_written_when_bessd_pebble_ready_then_config_f }, service_statuses={"pfcp-agent": ServiceStatus.ACTIVE}, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", mounts={"config": bessd_config_mount}, can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=0, stdout="RUNNING", ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -140,7 +140,7 @@ def test_given_config_file_already_written_when_bessd_pebble_ready_then_config_f ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -151,7 +151,7 @@ def test_given_config_file_already_written_when_bessd_pebble_ready_then_config_f ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -159,7 +159,7 @@ def test_given_config_file_already_written_when_bessd_pebble_ready_then_config_f "access-gateway-ip": access_gateway_ip, "gnb-subnet": gnb_subnet, }, - model=scenario.Model(name="whatever"), + model=testing.Model(name="whatever"), ) with open("tests/unit/expected_upf.json", "r") as f: expected_upf_config = f.read() @@ -182,11 +182,11 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_pebble_layer_i access_gateway_ip = "2.1.1.1" self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, layers={ @@ -194,26 +194,26 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_pebble_layer_i }, service_statuses={"pfcp-agent": ServiceStatus.ACTIVE}, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", mounts={"config": bessd_config_mount}, can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=0, stdout="RUNNING", ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -222,7 +222,7 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_pebble_layer_i ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -233,7 +233,7 @@ def test_given_bessd_container_ready_when_bessd_pebble_ready_then_pebble_layer_i ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -298,11 +298,11 @@ def test_given_bessd_not_configured_when_bessd_pebble_ready_then_bessd_configure access_gateway_ip = "2.1.1.1" self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, layers={ @@ -310,16 +310,16 @@ def test_given_bessd_not_configured_when_bessd_pebble_ready_then_bessd_configure }, service_statuses={"pfcp-agent": ServiceStatus.ACTIVE}, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", mounts={"config": bessd_config_mount}, can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "run", @@ -330,7 +330,7 @@ def test_given_bessd_not_configured_when_bessd_pebble_ready_then_bessd_configure ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -354,11 +354,11 @@ def test_given_routes_not_created_when_bessd_pebble_ready_then_routes_created(se access_gateway_ip = "2.1.1.1" self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, layers={ @@ -366,21 +366,21 @@ def test_given_routes_not_created_when_bessd_pebble_ready_then_routes_created(se }, service_statuses={"pfcp-agent": ServiceStatus.ACTIVE}, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", mounts={"config": bessd_config_mount}, can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout="", # route not created ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "ip", "route", @@ -393,7 +393,7 @@ def test_given_routes_not_created_when_bessd_pebble_ready_then_routes_created(se ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "ip", "route", @@ -406,7 +406,7 @@ def test_given_routes_not_created_when_bessd_pebble_ready_then_routes_created(se ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -445,11 +445,11 @@ def test_given_iptables_rule_not_created_when_bessd_pebble_ready_then_rule_creat access_gateway_ip = "2.1.1.1" self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, layers={ @@ -457,16 +457,16 @@ def test_given_iptables_rule_not_created_when_bessd_pebble_ready_then_rule_creat }, service_statuses={"pfcp-agent": ServiceStatus.ACTIVE}, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", mounts={"config": bessd_config_mount}, can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "iptables-legacy", "-I", @@ -482,7 +482,7 @@ def test_given_iptables_rule_not_created_when_bessd_pebble_ready_then_rule_creat ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ diff --git a/tests/unit/test_charm_collect_status.py b/tests/unit/test_charm_collect_status.py index 95202e54..3b1b3b7c 100644 --- a/tests/unit/test_charm_collect_status.py +++ b/tests/unit/test_charm_collect_status.py @@ -4,8 +4,7 @@ import tempfile import pytest -import scenario -from ops import ActiveStatus, BlockedStatus, WaitingStatus +from ops import ActiveStatus, BlockedStatus, WaitingStatus, testing from ops.pebble import Layer, ServiceStatus from tests.unit.fixtures import UPFUnitTestFixtures @@ -15,7 +14,7 @@ class TestCharmCollectUnitStatus(UPFUnitTestFixtures): def test_given_unit_not_leader_when_collect_unit_status_then_status_is_blocked( self, ): - state_in = scenario.State( + state_in = testing.State( leader=False, ) @@ -50,7 +49,7 @@ def test_given_unit_not_leader_when_collect_unit_status_then_status_is_blocked( def test_given_invalid_config_when_collect_unit_status_then_status_is_blocked( self, config_param, value ): - state_in = scenario.State( + state_in = testing.State( leader=True, config={ config_param: value, @@ -66,7 +65,7 @@ def test_given_invalid_config_when_collect_unit_status_then_status_is_blocked( def test_given_upf_mode_set_to_dpdk_and_hugepages_enabled_but_mac_addresses_of_access_and_core_interfaces_not_set_when_collect_unit_status_then_status_is_blocked( # noqa: E501 self, ): - state_in = scenario.State( + state_in = testing.State( leader=True, config={"upf-mode": "dpdk"}, ) @@ -81,7 +80,7 @@ def test_given_cpu_incompatible_when_collect_unit_status_then_status_is_blocked( self, ): self.mock_check_output.return_value = b"Flags: ssse3 fma cx16 rdrand" - state_in = scenario.State( + state_in = testing.State( leader=True, ) @@ -96,7 +95,7 @@ def test_given_hugepages_unavailable_when_collect_unit_status_then_status_is_blo ): self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand pdpe1gb" self.mock_client_list.return_value = [] - state_in = scenario.State( + state_in = testing.State( leader=True, config={ "upf-mode": "dpdk", @@ -113,7 +112,7 @@ def test_given_multus_not_available_when_collect_unit_status_then_status_is_bloc self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" self.mock_client_list.return_value = [] self.mock_multus_is_available.return_value = False - state_in = scenario.State( + state_in = testing.State( leader=True, config={}, ) @@ -128,11 +127,11 @@ def test_given_cant_connect_to_bessd_container_when_collect_unit_status_then_sta self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" self.mock_client_list.return_value = [] self.mock_multus_is_available.return_value = True - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=False, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container], ) @@ -146,15 +145,15 @@ def test_given_multus_not_ready_when_collect_unit_status_then_status_is_waiting( self.mock_client_list.return_value = [] self.mock_multus_is_available.return_value = True self.mock_multus_is_ready.return_value = False - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], ) @@ -168,22 +167,22 @@ def test_given_default_route_not_created_when_collect_unit_status_then_status_is self.mock_client_list.return_value = [] self.mock_multus_is_available.return_value = True self.mock_multus_is_ready.return_value = True - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout="", ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], ) @@ -198,22 +197,22 @@ def test_given_gnb_route_not_created_when_collect_unit_status_then_status_is_wai self.mock_multus_is_available.return_value = True self.mock_multus_is_ready.return_value = True core_gateway_ip = "1.2.3.4" - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}", ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -234,22 +233,22 @@ def test_given_storage_not_attached_when_collect_unit_status_then_status_is_wait gnb_subnet = "2.2.2.0/24" core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -274,36 +273,36 @@ def test_given_bessd_service_not_running_when_collect_unit_status_then_status_is core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_config_mount = scenario.Mount( + pfcp_agent_config_mount = testing.Mount( location="/tmp/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, mounts={ "config": pfcp_agent_config_mount, }, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={ "config": bessd_config_mount, }, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -326,22 +325,22 @@ def test_given_grpc_service_not_ready_when_collect_unit_status_then_status_is_wa core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_config_mount = scenario.Mount( + pfcp_agent_config_mount = testing.Mount( location="/tmp/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, mounts={ "config": pfcp_agent_config_mount, }, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={ @@ -350,18 +349,18 @@ def test_given_grpc_service_not_ready_when_collect_unit_status_then_status_is_wa layers={"bessd": Layer({"services": {"bessd": {}}})}, service_statuses={"bessd": ServiceStatus.ACTIVE}, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=1, ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -386,22 +385,22 @@ def test_given_bessd_not_configured_when_collect_unit_status_then_status_is_wait core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_config_mount = scenario.Mount( + pfcp_agent_config_mount = testing.Mount( location="/tmp/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, mounts={ "config": pfcp_agent_config_mount, }, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={ @@ -410,22 +409,22 @@ def test_given_bessd_not_configured_when_collect_unit_status_then_status_is_wait layers={"bessd": Layer({"services": {"bessd": {}}})}, service_statuses={"bessd": ServiceStatus.ACTIVE}, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=1, ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -452,22 +451,22 @@ def test_given_routectl_service_not_running_when_collect_unit_status_then_status core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_config_mount = scenario.Mount( + pfcp_agent_config_mount = testing.Mount( location="/tmp/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, mounts={ "config": pfcp_agent_config_mount, }, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={ @@ -476,21 +475,21 @@ def test_given_routectl_service_not_running_when_collect_unit_status_then_status layers={"bessd": Layer({"services": {"bessd": {}}})}, service_statuses={"bessd": ServiceStatus.ACTIVE}, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=0, stdout="RUNNING", ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -499,7 +498,7 @@ def test_given_routectl_service_not_running_when_collect_unit_status_then_status ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -510,7 +509,7 @@ def test_given_routectl_service_not_running_when_collect_unit_status_then_status ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -524,7 +523,9 @@ def test_given_routectl_service_not_running_when_collect_unit_status_then_status assert state_out.unit_status == WaitingStatus("Waiting for routectl service to run") - def test_given_pfcp_agent_storage_not_attached_when_collect_unit_status_then_status_is_waiting(self): # noqa: E501 + def test_given_pfcp_agent_storage_not_attached_when_collect_unit_status_then_status_is_waiting( + self, + ): # noqa: E501 self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" self.mock_client_list.return_value = [] self.mock_multus_is_available.return_value = True @@ -533,15 +534,15 @@ def test_given_pfcp_agent_storage_not_attached_when_collect_unit_status_then_sta core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={ @@ -552,21 +553,21 @@ def test_given_pfcp_agent_storage_not_attached_when_collect_unit_status_then_sta }, service_statuses={"bessd": ServiceStatus.ACTIVE, "routectl": ServiceStatus.ACTIVE}, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=0, stdout="RUNNING", ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -575,7 +576,7 @@ def test_given_pfcp_agent_storage_not_attached_when_collect_unit_status_then_sta ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -586,7 +587,7 @@ def test_given_pfcp_agent_storage_not_attached_when_collect_unit_status_then_sta ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -611,22 +612,22 @@ def test_given_pfcp_service_not_running_when_collect_unit_status_then_status_is_ core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_config_mount = scenario.Mount( + pfcp_agent_config_mount = testing.Mount( location="/tmp/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, mounts={ "config": pfcp_agent_config_mount, }, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={ @@ -637,21 +638,21 @@ def test_given_pfcp_service_not_running_when_collect_unit_status_then_status_is_ }, service_statuses={"bessd": ServiceStatus.ACTIVE, "routectl": ServiceStatus.ACTIVE}, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=0, stdout="RUNNING", ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -660,7 +661,7 @@ def test_given_pfcp_service_not_running_when_collect_unit_status_then_status_is_ ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -671,7 +672,7 @@ def test_given_pfcp_service_not_running_when_collect_unit_status_then_status_is_ ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -694,15 +695,15 @@ def test_given_services_are_running_when_collect_unit_status_then_status_is_wait core_gateway_ip = "1.2.3.4" access_gateway_ip = "2.1.1.1" with tempfile.TemporaryDirectory() as temp_file: - bessd_config_mount = scenario.Mount( + bessd_config_mount = testing.Mount( location="/etc/bess/conf/", source=temp_file, ) - pfcp_agent_config_mount = scenario.Mount( + pfcp_agent_config_mount = testing.Mount( location="/tmp/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, mounts={ @@ -713,7 +714,7 @@ def test_given_services_are_running_when_collect_unit_status_then_status_is_wait }, service_statuses={"pfcp-agent": ServiceStatus.ACTIVE}, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={ @@ -724,21 +725,21 @@ def test_given_services_are_running_when_collect_unit_status_then_status_is_wait }, service_statuses={"bessd": ServiceStatus.ACTIVE, "routectl": ServiceStatus.ACTIVE}, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout=f"default via {core_gateway_ip}\n {gnb_subnet} via {access_gateway_ip}", # noqa: E501 ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "version"], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=["/opt/bess/bessctl/bessctl", "show", "worker"], return_code=0, stdout="RUNNING", ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -747,7 +748,7 @@ def test_given_services_are_running_when_collect_unit_status_then_status_is_wait ], return_code=0, ), - scenario.Exec( + testing.Exec( command_prefix=[ "/opt/bess/bessctl/bessctl", "show", @@ -758,7 +759,7 @@ def test_given_services_are_running_when_collect_unit_status_then_status_is_wait ), }, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ @@ -780,15 +781,15 @@ def test_given_no_workload_version_when_collect_unit_status_then_workload_versio self.mock_multus_is_available.return_value = True self.mock_multus_is_ready.return_value = False - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], ) @@ -806,21 +807,21 @@ def test_given_workload_version_file_when_collect_unit_status_then_workload_vers self.mock_multus_is_ready.return_value = False with tempfile.TemporaryDirectory() as temp_file: - workload_version_mount = scenario.Mount( + workload_version_mount = testing.Mount( location="/etc", source=temp_file, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, mounts={"workload-version": workload_version_mount}, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], ) diff --git a/tests/unit/test_charm_config_changed.py b/tests/unit/test_charm_config_changed.py index 37d52003..7c25e8b3 100644 --- a/tests/unit/test_charm_config_changed.py +++ b/tests/unit/test_charm_config_changed.py @@ -2,7 +2,7 @@ # See LICENSE file for licensing details. -import scenario +from ops import testing from tests.unit.fixtures import UPFUnitTestFixtures @@ -14,12 +14,12 @@ def test_given_dpdk_when_config_changed_then_bessd_configured_for_dpdk( self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand pdpe1gb" self.mock_k8s_service.is_created.return_value = True self.mock_dpdk.is_configured.return_value = False - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=False, ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container], config={ @@ -38,20 +38,20 @@ def test_given_n3_integration_when_config_changed_then_n3_information_published( ): self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand pdpe1gb" self.mock_k8s_service.is_created.return_value = True - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - n3_relation = scenario.Relation( + n3_relation = testing.Relation( endpoint="fiveg_n3", interface="fiveg_n3", ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], relations=[n3_relation], @@ -71,20 +71,20 @@ def test_given_n4_integration_when_config_changed_then_n4_information_published( self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand pdpe1gb" self.mock_k8s_service.is_created.return_value = True self.mock_k8s_service.get_hostname.return_value = "my-hostname" - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, ) - n4_relation = scenario.Relation( + n4_relation = testing.Relation( endpoint="fiveg_n4", interface="fiveg_n4", ) - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], relations=[n4_relation], diff --git a/tests/unit/test_charm_pfcp_agent_pebble_ready.py b/tests/unit/test_charm_pfcp_agent_pebble_ready.py index 0fa56338..52d060b2 100644 --- a/tests/unit/test_charm_pfcp_agent_pebble_ready.py +++ b/tests/unit/test_charm_pfcp_agent_pebble_ready.py @@ -2,7 +2,7 @@ # See LICENSE file for licensing details. import tempfile -import scenario +from ops import testing from ops.pebble import Layer, ServiceStatus from tests.unit.fixtures import UPFUnitTestFixtures @@ -17,22 +17,22 @@ def test_given_pfcp_agent_container_ready_when_pfcp_agent_pebble_ready_then_pebb access_gateway_ip = "2.1.1.1" self.mock_check_output.return_value = b"Flags: avx2 ssse3 fma cx16 rdrand" with tempfile.TemporaryDirectory() as temp_file: - pfcp_agent_config_mount = scenario.Mount( + pfcp_agent_config_mount = testing.Mount( location="/tmp/conf/", source=temp_file, ) - pfcp_agent_container = scenario.Container( + pfcp_agent_container = testing.Container( name="pfcp-agent", can_connect=True, mounts={ "config": pfcp_agent_config_mount, }, ) - bessd_container = scenario.Container( + bessd_container = testing.Container( name="bessd", can_connect=True, execs={ - scenario.Exec( + testing.Exec( command_prefix=["ip", "route", "show"], return_code=0, stdout="", @@ -48,7 +48,7 @@ def test_given_pfcp_agent_container_ready_when_pfcp_agent_pebble_ready_then_pebb }, ) self.mock_k8s_service.is_created.return_value = True - state_in = scenario.State( + state_in = testing.State( leader=True, containers=[bessd_container, pfcp_agent_container], config={ diff --git a/tests/unit/test_charm_remove.py b/tests/unit/test_charm_remove.py index 276205d8..f20b46da 100644 --- a/tests/unit/test_charm_remove.py +++ b/tests/unit/test_charm_remove.py @@ -2,7 +2,7 @@ # See LICENSE file for licensing details. -import scenario +from ops import testing from tests.unit.fixtures import UPFUnitTestFixtures @@ -12,7 +12,7 @@ def test_given_k8s_service_is_created_when_remove_then_service_is_deleted( self, ): self.mock_k8s_service.is_created.return_value = True - state_in = scenario.State( + state_in = testing.State( leader=True, )