#2248 - synced wth dev
This commit is contained in:
@@ -15,6 +15,7 @@ def populated_node(application_class) -> Tuple[Application, Computer]:
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.1.1",
|
||||
start_up_duration=0,
|
||||
shut_down_duration=0,
|
||||
)
|
||||
computer.power_on()
|
||||
computer.software_manager.install(application_class)
|
||||
@@ -33,6 +34,7 @@ def test_application_on_offline_node(application_class):
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.1.1",
|
||||
start_up_duration=0,
|
||||
shut_down_duration=0,
|
||||
)
|
||||
computer.software_manager.install(application_class)
|
||||
|
||||
@@ -40,9 +42,6 @@ def test_application_on_offline_node(application_class):
|
||||
|
||||
computer.power_off()
|
||||
|
||||
for i in range(computer.shut_down_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.OFF
|
||||
assert app.operating_state is ApplicationOperatingState.CLOSED
|
||||
|
||||
@@ -59,9 +58,6 @@ def test_server_turns_off_application(populated_node):
|
||||
|
||||
computer.power_off()
|
||||
|
||||
for i in range(computer.shut_down_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.OFF
|
||||
assert app.operating_state is ApplicationOperatingState.CLOSED
|
||||
|
||||
@@ -75,9 +71,6 @@ def test_application_cannot_be_turned_on_when_computer_is_off(populated_node):
|
||||
|
||||
computer.power_off()
|
||||
|
||||
for i in range(computer.shut_down_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.OFF
|
||||
assert app.operating_state is ApplicationOperatingState.CLOSED
|
||||
|
||||
@@ -96,28 +89,20 @@ def test_computer_runs_applications(populated_node):
|
||||
|
||||
computer.power_off()
|
||||
|
||||
for i in range(computer.shut_down_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.OFF
|
||||
assert app.operating_state is ApplicationOperatingState.CLOSED
|
||||
|
||||
computer.power_on()
|
||||
|
||||
for i in range(computer.start_up_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.ON
|
||||
assert app.operating_state is ApplicationOperatingState.RUNNING
|
||||
|
||||
computer.power_off()
|
||||
for i in range(computer.start_up_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.OFF
|
||||
assert app.operating_state is ApplicationOperatingState.CLOSED
|
||||
|
||||
computer.power_on()
|
||||
for i in range(computer.start_up_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.ON
|
||||
assert app.operating_state is ApplicationOperatingState.RUNNING
|
||||
|
||||
@@ -12,7 +12,13 @@ from primaite.simulator.system.services.service import Service, ServiceOperating
|
||||
def populated_node(
|
||||
service_class,
|
||||
) -> Tuple[Server, Service]:
|
||||
server = Server(hostname="server", ip_address="192.168.0.1", subnet_mask="255.255.255.0", start_up_duration=0)
|
||||
server = Server(
|
||||
hostname="server",
|
||||
ip_address="192.168.0.1",
|
||||
subnet_mask="255.255.255.0",
|
||||
start_up_duration=0,
|
||||
shut_down_duration=0,
|
||||
)
|
||||
server.power_on()
|
||||
server.software_manager.install(service_class)
|
||||
|
||||
@@ -30,6 +36,7 @@ def test_service_on_offline_node(service_class):
|
||||
subnet_mask="255.255.255.0",
|
||||
default_gateway="192.168.1.1",
|
||||
start_up_duration=0,
|
||||
shut_down_duration=0,
|
||||
)
|
||||
computer.power_on()
|
||||
computer.software_manager.install(service_class)
|
||||
@@ -38,9 +45,6 @@ def test_service_on_offline_node(service_class):
|
||||
|
||||
computer.power_off()
|
||||
|
||||
for i in range(computer.shut_down_duration + 1):
|
||||
computer.apply_timestep(timestep=i)
|
||||
|
||||
assert computer.operating_state is NodeOperatingState.OFF
|
||||
assert service.operating_state is ServiceOperatingState.STOPPED
|
||||
|
||||
@@ -66,9 +70,6 @@ def test_server_turns_off_service(populated_node):
|
||||
|
||||
server.power_off()
|
||||
|
||||
for i in range(server.shut_down_duration + 1):
|
||||
server.apply_timestep(timestep=i)
|
||||
|
||||
assert server.operating_state is NodeOperatingState.OFF
|
||||
assert service.operating_state is ServiceOperatingState.STOPPED
|
||||
|
||||
@@ -82,9 +83,6 @@ def test_service_cannot_be_turned_on_when_server_is_off(populated_node):
|
||||
|
||||
server.power_off()
|
||||
|
||||
for i in range(server.shut_down_duration + 1):
|
||||
server.apply_timestep(timestep=i)
|
||||
|
||||
assert server.operating_state is NodeOperatingState.OFF
|
||||
assert service.operating_state is ServiceOperatingState.STOPPED
|
||||
|
||||
@@ -103,28 +101,20 @@ def test_server_turns_on_service(populated_node):
|
||||
|
||||
server.power_off()
|
||||
|
||||
for i in range(server.shut_down_duration + 1):
|
||||
server.apply_timestep(timestep=i)
|
||||
|
||||
assert server.operating_state is NodeOperatingState.OFF
|
||||
assert service.operating_state is ServiceOperatingState.STOPPED
|
||||
|
||||
server.power_on()
|
||||
|
||||
for i in range(server.start_up_duration + 1):
|
||||
server.apply_timestep(timestep=i)
|
||||
|
||||
assert server.operating_state is NodeOperatingState.ON
|
||||
assert service.operating_state is ServiceOperatingState.RUNNING
|
||||
|
||||
server.power_off()
|
||||
for i in range(server.start_up_duration + 1):
|
||||
server.apply_timestep(timestep=i)
|
||||
|
||||
assert server.operating_state is NodeOperatingState.OFF
|
||||
assert service.operating_state is ServiceOperatingState.STOPPED
|
||||
|
||||
server.power_on()
|
||||
for i in range(server.start_up_duration + 1):
|
||||
server.apply_timestep(timestep=i)
|
||||
|
||||
assert server.operating_state is NodeOperatingState.ON
|
||||
assert service.operating_state is ServiceOperatingState.RUNNING
|
||||
|
||||
@@ -3,6 +3,7 @@ from typing import Tuple
|
||||
|
||||
import pytest
|
||||
|
||||
from primaite.simulator.network.container import Network
|
||||
from primaite.simulator.network.hardware.base import Link
|
||||
from primaite.simulator.network.hardware.nodes.host.computer import Computer
|
||||
from primaite.simulator.network.hardware.nodes.host.server import Server
|
||||
@@ -17,7 +18,7 @@ from primaite.simulator.system.services.web_server.web_server import WebServer
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def web_client_web_server_database(example_network) -> Tuple[Computer, Server, Server]:
|
||||
def web_client_web_server_database(example_network) -> Tuple[Network, Computer, Server, Server]:
|
||||
# add rules to network router
|
||||
router_1: Router = example_network.get_node_by_hostname("router_1")
|
||||
router_1.acl.add_rule(
|
||||
@@ -97,12 +98,52 @@ def web_client_web_server_database(example_network) -> Tuple[Computer, Server, S
|
||||
assert dns_client.check_domain_exists("arcd.com")
|
||||
assert db_client.connect()
|
||||
|
||||
return computer, web_server, db_server
|
||||
return example_network, computer, web_server, db_server
|
||||
|
||||
|
||||
def test_web_client_requests_users(web_client_web_server_database):
|
||||
computer, web_server, db_server = web_client_web_server_database
|
||||
_, computer, _, _ = web_client_web_server_database
|
||||
|
||||
web_browser: WebBrowser = computer.software_manager.software.get("WebBrowser")
|
||||
|
||||
assert web_browser.get_webpage()
|
||||
|
||||
|
||||
class TestWebBrowserHistory:
|
||||
def test_populating_history(self, web_client_web_server_database):
|
||||
network, computer, _, _ = web_client_web_server_database
|
||||
|
||||
web_browser: WebBrowser = computer.software_manager.software.get("WebBrowser")
|
||||
assert web_browser.history == []
|
||||
web_browser.get_webpage()
|
||||
assert len(web_browser.history) == 1
|
||||
web_browser.get_webpage()
|
||||
assert len(web_browser.history) == 2
|
||||
assert web_browser.history[-1].status == WebBrowser.BrowserHistoryItem._HistoryItemStatus.LOADED
|
||||
assert web_browser.history[-1].response_code == 200
|
||||
|
||||
router = network.get_node_by_hostname("router_1")
|
||||
router.acl.add_rule(action=ACLAction.DENY, src_port=Port.HTTP, dst_port=Port.HTTP, position=0)
|
||||
assert not web_browser.get_webpage()
|
||||
assert len(web_browser.history) == 3
|
||||
# with current NIC behaviour, even if you block communication, you won't get SERVER_UNREACHABLE because
|
||||
# application.send always returns true, even if communication fails. we should change what is returned from NICs
|
||||
assert web_browser.history[-1].status == WebBrowser.BrowserHistoryItem._HistoryItemStatus.LOADED
|
||||
assert web_browser.history[-1].response_code == 404
|
||||
|
||||
def test_history_in_state(self, web_client_web_server_database):
|
||||
network, computer, _, _ = web_client_web_server_database
|
||||
web_browser: WebBrowser = computer.software_manager.software.get("WebBrowser")
|
||||
|
||||
state = computer.describe_state()
|
||||
assert "history" in state["applications"]["WebBrowser"]
|
||||
assert len(state["applications"]["WebBrowser"]["history"]) == 0
|
||||
|
||||
web_browser.get_webpage()
|
||||
router = network.get_node_by_hostname("router_1")
|
||||
router.acl.add_rule(action=ACLAction.DENY, src_port=Port.HTTP, dst_port=Port.HTTP, position=0)
|
||||
web_browser.get_webpage()
|
||||
|
||||
state = computer.describe_state()
|
||||
assert state["applications"]["WebBrowser"]["history"][0]["outcome"] == 200
|
||||
assert state["applications"]["WebBrowser"]["history"][1]["outcome"] == 404
|
||||
|
||||
Reference in New Issue
Block a user