#2688: refactor test classes
This commit is contained in:
@@ -51,11 +51,11 @@ class TestService(Service):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class TestDummyApplication(Application, identifier="TestDummyApplication"):
|
class DummyApplication(Application, identifier="DummyApplication"):
|
||||||
"""Test Application class"""
|
"""Test Application class"""
|
||||||
|
|
||||||
def __init__(self, **kwargs):
|
def __init__(self, **kwargs):
|
||||||
kwargs["name"] = "TestDummyApplication"
|
kwargs["name"] = "DummyApplication"
|
||||||
kwargs["port"] = Port.HTTP
|
kwargs["port"] = Port.HTTP
|
||||||
kwargs["protocol"] = IPProtocol.TCP
|
kwargs["protocol"] = IPProtocol.TCP
|
||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
@@ -85,9 +85,9 @@ def service_class():
|
|||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def application(file_system) -> TestDummyApplication:
|
def application(file_system) -> DummyApplication:
|
||||||
return TestDummyApplication(
|
return DummyApplication(
|
||||||
name="TestDummyApplication",
|
name="DummyApplication",
|
||||||
port=Port.ARP,
|
port=Port.ARP,
|
||||||
file_system=file_system,
|
file_system=file_system,
|
||||||
sys_log=SysLog(hostname="dummy_application"),
|
sys_log=SysLog(hostname="dummy_application"),
|
||||||
@@ -96,7 +96,7 @@ def application(file_system) -> TestDummyApplication:
|
|||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def application_class():
|
def application_class():
|
||||||
return TestDummyApplication
|
return DummyApplication
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ from tests import TEST_ASSETS_ROOT
|
|||||||
TEST_CONFIG = TEST_ASSETS_ROOT / "configs/software_fix_duration.yaml"
|
TEST_CONFIG = TEST_ASSETS_ROOT / "configs/software_fix_duration.yaml"
|
||||||
ONE_ITEM_CONFIG = TEST_ASSETS_ROOT / "configs/fix_duration_one_item.yaml"
|
ONE_ITEM_CONFIG = TEST_ASSETS_ROOT / "configs/fix_duration_one_item.yaml"
|
||||||
|
|
||||||
TestApplications = ["TestDummyApplication", "TestBroadcastClient"]
|
TestApplications = ["DummyApplication", "BroadcastTestClient"]
|
||||||
|
|
||||||
|
|
||||||
def load_config(config_path: Union[str, Path]) -> PrimaiteGame:
|
def load_config(config_path: Union[str, Path]) -> PrimaiteGame:
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ from primaite.simulator.system.applications.application import Application
|
|||||||
from primaite.simulator.system.services.service import Service
|
from primaite.simulator.system.services.service import Service
|
||||||
|
|
||||||
|
|
||||||
class TestBroadcastService(Service):
|
class BroadcastTestService(Service):
|
||||||
"""A service for sending broadcast and unicast messages over a network."""
|
"""A service for sending broadcast and unicast messages over a network."""
|
||||||
|
|
||||||
def __init__(self, **kwargs):
|
def __init__(self, **kwargs):
|
||||||
@@ -41,14 +41,14 @@ class TestBroadcastService(Service):
|
|||||||
super().send(payload="broadcast", dest_ip_address=ip_network, dest_port=Port.HTTP, ip_protocol=self.protocol)
|
super().send(payload="broadcast", dest_ip_address=ip_network, dest_port=Port.HTTP, ip_protocol=self.protocol)
|
||||||
|
|
||||||
|
|
||||||
class TestBroadcastClient(Application, identifier="TestBroadcastClient"):
|
class BroadcastTestClient(Application, identifier="BroadcastTestClient"):
|
||||||
"""A client application to receive broadcast and unicast messages."""
|
"""A client application to receive broadcast and unicast messages."""
|
||||||
|
|
||||||
payloads_received: List = []
|
payloads_received: List = []
|
||||||
|
|
||||||
def __init__(self, **kwargs):
|
def __init__(self, **kwargs):
|
||||||
# Set default client properties
|
# Set default client properties
|
||||||
kwargs["name"] = "TestBroadcastClient"
|
kwargs["name"] = "BroadcastTestClient"
|
||||||
kwargs["port"] = Port.HTTP
|
kwargs["port"] = Port.HTTP
|
||||||
kwargs["protocol"] = IPProtocol.TCP
|
kwargs["protocol"] = IPProtocol.TCP
|
||||||
super().__init__(**kwargs)
|
super().__init__(**kwargs)
|
||||||
@@ -75,8 +75,8 @@ def broadcast_network() -> Network:
|
|||||||
start_up_duration=0,
|
start_up_duration=0,
|
||||||
)
|
)
|
||||||
client_1.power_on()
|
client_1.power_on()
|
||||||
client_1.software_manager.install(TestBroadcastClient)
|
client_1.software_manager.install(BroadcastTestClient)
|
||||||
application_1 = client_1.software_manager.software["TestBroadcastClient"]
|
application_1 = client_1.software_manager.software["BroadcastTestClient"]
|
||||||
application_1.run()
|
application_1.run()
|
||||||
|
|
||||||
client_2 = Computer(
|
client_2 = Computer(
|
||||||
@@ -87,8 +87,8 @@ def broadcast_network() -> Network:
|
|||||||
start_up_duration=0,
|
start_up_duration=0,
|
||||||
)
|
)
|
||||||
client_2.power_on()
|
client_2.power_on()
|
||||||
client_2.software_manager.install(TestBroadcastClient)
|
client_2.software_manager.install(BroadcastTestClient)
|
||||||
application_2 = client_2.software_manager.software["TestBroadcastClient"]
|
application_2 = client_2.software_manager.software["BroadcastTestClient"]
|
||||||
application_2.run()
|
application_2.run()
|
||||||
|
|
||||||
server_1 = Server(
|
server_1 = Server(
|
||||||
@@ -100,8 +100,8 @@ def broadcast_network() -> Network:
|
|||||||
)
|
)
|
||||||
server_1.power_on()
|
server_1.power_on()
|
||||||
|
|
||||||
server_1.software_manager.install(TestBroadcastService)
|
server_1.software_manager.install(BroadcastTestService)
|
||||||
service: TestBroadcastService = server_1.software_manager.software["BroadcastService"]
|
service: BroadcastTestService = server_1.software_manager.software["BroadcastService"]
|
||||||
service.start()
|
service.start()
|
||||||
|
|
||||||
switch_1 = Switch(hostname="switch_1", num_ports=6, start_up_duration=0)
|
switch_1 = Switch(hostname="switch_1", num_ports=6, start_up_duration=0)
|
||||||
@@ -117,14 +117,14 @@ def broadcast_network() -> Network:
|
|||||||
@pytest.fixture(scope="function")
|
@pytest.fixture(scope="function")
|
||||||
def broadcast_service_and_clients(
|
def broadcast_service_and_clients(
|
||||||
broadcast_network,
|
broadcast_network,
|
||||||
) -> Tuple[TestBroadcastService, TestBroadcastClient, TestBroadcastClient]:
|
) -> Tuple[BroadcastTestService, BroadcastTestClient, BroadcastTestClient]:
|
||||||
client_1: TestBroadcastClient = broadcast_network.get_node_by_hostname("client_1").software_manager.software[
|
client_1: BroadcastTestClient = broadcast_network.get_node_by_hostname("client_1").software_manager.software[
|
||||||
"TestBroadcastClient"
|
"BroadcastTestClient"
|
||||||
]
|
]
|
||||||
client_2: TestBroadcastClient = broadcast_network.get_node_by_hostname("client_2").software_manager.software[
|
client_2: BroadcastTestClient = broadcast_network.get_node_by_hostname("client_2").software_manager.software[
|
||||||
"TestBroadcastClient"
|
"BroadcastTestClient"
|
||||||
]
|
]
|
||||||
service: TestBroadcastService = broadcast_network.get_node_by_hostname("server_1").software_manager.software[
|
service: BroadcastTestService = broadcast_network.get_node_by_hostname("server_1").software_manager.software[
|
||||||
"BroadcastService"
|
"BroadcastService"
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|||||||
@@ -21,7 +21,7 @@ def populated_node(application_class) -> Tuple[Application, Computer]:
|
|||||||
computer.power_on()
|
computer.power_on()
|
||||||
computer.software_manager.install(application_class)
|
computer.software_manager.install(application_class)
|
||||||
|
|
||||||
app = computer.software_manager.software.get("TestDummyApplication")
|
app = computer.software_manager.software.get("DummyApplication")
|
||||||
app.run()
|
app.run()
|
||||||
|
|
||||||
return app, computer
|
return app, computer
|
||||||
@@ -39,7 +39,7 @@ def test_application_on_offline_node(application_class):
|
|||||||
)
|
)
|
||||||
computer.software_manager.install(application_class)
|
computer.software_manager.install(application_class)
|
||||||
|
|
||||||
app: Application = computer.software_manager.software.get("TestDummyApplication")
|
app: Application = computer.software_manager.software.get("DummyApplication")
|
||||||
|
|
||||||
computer.power_off()
|
computer.power_off()
|
||||||
|
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ from primaite.simulator.network.hardware.node_operating_state import NodeOperati
|
|||||||
from primaite.simulator.network.hardware.nodes.host.host_node import HostNode
|
from primaite.simulator.network.hardware.nodes.host.host_node import HostNode
|
||||||
from primaite.simulator.network.hardware.nodes.network.router import ACLAction, Router
|
from primaite.simulator.network.hardware.nodes.network.router import ACLAction, Router
|
||||||
from primaite.simulator.network.transmission.transport_layer import Port
|
from primaite.simulator.network.transmission.transport_layer import Port
|
||||||
from tests.conftest import TestDummyApplication, TestService
|
from tests.conftest import DummyApplication, TestService
|
||||||
|
|
||||||
|
|
||||||
def test_successful_node_file_system_creation_request(example_network):
|
def test_successful_node_file_system_creation_request(example_network):
|
||||||
@@ -47,14 +47,14 @@ def test_successful_application_requests(example_network):
|
|||||||
net = example_network
|
net = example_network
|
||||||
|
|
||||||
client_1 = net.get_node_by_hostname("client_1")
|
client_1 = net.get_node_by_hostname("client_1")
|
||||||
client_1.software_manager.install(TestDummyApplication)
|
client_1.software_manager.install(DummyApplication)
|
||||||
client_1.software_manager.software.get("TestDummyApplication").run()
|
client_1.software_manager.software.get("DummyApplication").run()
|
||||||
|
|
||||||
resp_1 = net.apply_request(["node", "client_1", "application", "TestDummyApplication", "scan"])
|
resp_1 = net.apply_request(["node", "client_1", "application", "DummyApplication", "scan"])
|
||||||
assert resp_1 == RequestResponse(status="success", data={})
|
assert resp_1 == RequestResponse(status="success", data={})
|
||||||
resp_2 = net.apply_request(["node", "client_1", "application", "TestDummyApplication", "fix"])
|
resp_2 = net.apply_request(["node", "client_1", "application", "DummyApplication", "fix"])
|
||||||
assert resp_2 == RequestResponse(status="success", data={})
|
assert resp_2 == RequestResponse(status="success", data={})
|
||||||
resp_3 = net.apply_request(["node", "client_1", "application", "TestDummyApplication", "compromise"])
|
resp_3 = net.apply_request(["node", "client_1", "application", "DummyApplication", "compromise"])
|
||||||
assert resp_3 == RequestResponse(status="success", data={})
|
assert resp_3 == RequestResponse(status="success", data={})
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user