Add tests for request success/fail

This commit is contained in:
Marek Wolan
2024-03-09 23:06:53 +00:00
parent 31ae4672ac
commit 359777f4f8
2 changed files with 76 additions and 4 deletions

View File

@@ -146,9 +146,12 @@ def arcd_uc2_network() -> Network:
)
client_1.power_on()
network.connect(endpoint_b=client_1.network_interface[1], endpoint_a=switch_2.network_interface[1])
db_client_1 = client_1.software_manager.install(DatabaseClient)
db_client_1 = client_1.software_manager.software.get("DatabaseClient")
client_1.software_manager.install(DatabaseClient)
db_client_1: DatabaseClient = client_1.software_manager.software.get("DatabaseClient")
db_client_1.configure(server_ip_address=IPv4Address("192.168.1.14"))
db_client_1.run()
web_browser_1 = client_1.software_manager.software.get("WebBrowser")
web_browser_1.target_url = "http://arcd.com/users/"
client_1.software_manager.install(DataManipulationBot)
db_manipulation_bot: DataManipulationBot = client_1.software_manager.software.get("DataManipulationBot")
db_manipulation_bot.configure(
@@ -170,9 +173,10 @@ def arcd_uc2_network() -> Network:
client_2.power_on()
client_2.software_manager.install(DatabaseClient)
db_client_2 = client_2.software_manager.software.get("DatabaseClient")
db_client_2.configure(server_ip_address=IPv4Address("192.168.1.14"))
db_client_2.run()
web_browser = client_2.software_manager.software.get("WebBrowser")
web_browser.target_url = "http://arcd.com/users/"
web_browser_2 = client_2.software_manager.software.get("WebBrowser")
web_browser_2.target_url = "http://arcd.com/users/"
network.connect(endpoint_b=client_2.network_interface[1], endpoint_a=switch_2.network_interface[2])
# Domain Controller

View File

@@ -9,6 +9,8 @@ import pytest
from primaite.interface.request import RequestResponse
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
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.transmission.transport_layer import Port
from tests.conftest import TestApplication, TestService
@@ -90,3 +92,69 @@ def test_request_fails_if_node_off(example_network, node_request):
assert client_1.operating_state == NodeOperatingState.OFF
resp_2 = net.apply_request(node_request)
assert resp_2.status == "failure"
class TestDataManipulationGreenRequests:
def test_node_off(self, uc2_network):
"""Test that green requests succeed when the node is on and fail if the node is off."""
net = uc2_network
client_1_browser_execute = net.apply_request(["node", "client_1", "application", "WebBrowser", "execute"])
client_1_db_client_execute = net.apply_request(["node", "client_1", "application", "DatabaseClient", "execute"])
client_2_browser_execute = net.apply_request(["node", "client_2", "application", "WebBrowser", "execute"])
client_2_db_client_execute = net.apply_request(["node", "client_2", "application", "DatabaseClient", "execute"])
assert client_1_browser_execute.status == "success"
assert client_1_db_client_execute.status == "success"
assert client_2_browser_execute.status == "success"
assert client_2_db_client_execute.status == "success"
client_1 = net.get_node_by_hostname("client_1")
client_2 = net.get_node_by_hostname("client_2")
client_1.shut_down_duration = 0
client_1.power_off()
client_2.shut_down_duration = 0
client_2.power_off()
client_1_browser_execute_off = net.apply_request(["node", "client_1", "application", "WebBrowser", "execute"])
client_1_db_client_execute_off = net.apply_request(
["node", "client_1", "application", "DatabaseClient", "execute"]
)
client_2_browser_execute_off = net.apply_request(["node", "client_2", "application", "WebBrowser", "execute"])
client_2_db_client_execute_off = net.apply_request(
["node", "client_2", "application", "DatabaseClient", "execute"]
)
assert client_1_browser_execute_off.status == "failure"
assert client_1_db_client_execute_off.status == "failure"
assert client_2_browser_execute_off.status == "failure"
assert client_2_db_client_execute_off.status == "failure"
def test_acl_block(self, uc2_network):
"""Test that green requests succeed when not blocked by ACLs but fail when blocked."""
net = uc2_network
router: Router = net.get_node_by_hostname("router_1")
client_1: HostNode = net.get_node_by_hostname("client_1")
client_2: HostNode = net.get_node_by_hostname("client_2")
client_1_browser_execute = net.apply_request(["node", "client_1", "application", "WebBrowser", "execute"])
client_2_browser_execute = net.apply_request(["node", "client_2", "application", "WebBrowser", "execute"])
assert client_1_browser_execute.status == "success"
assert client_2_browser_execute.status == "success"
router.acl.add_rule(ACLAction.DENY, src_port=Port.HTTP, dst_port=Port.HTTP, position=3)
client_1_browser_execute = net.apply_request(["node", "client_1", "application", "WebBrowser", "execute"])
client_2_browser_execute = net.apply_request(["node", "client_2", "application", "WebBrowser", "execute"])
assert client_1_browser_execute.status == "failure"
assert client_2_browser_execute.status == "failure"
client_1_db_client_execute = net.apply_request(["node", "client_1", "application", "DatabaseClient", "execute"])
client_2_db_client_execute = net.apply_request(["node", "client_2", "application", "DatabaseClient", "execute"])
assert client_1_db_client_execute.status == "success"
assert client_2_db_client_execute.status == "success"
router.acl.add_rule(ACLAction.DENY, src_port=Port.POSTGRES_SERVER, dst_port=Port.POSTGRES_SERVER)
client_1_db_client_execute = net.apply_request(["node", "client_1", "application", "DatabaseClient", "execute"])
client_2_db_client_execute = net.apply_request(["node", "client_2", "application", "DatabaseClient", "execute"])
assert client_1_db_client_execute.status == "failure"
assert client_2_db_client_execute.status == "failure"