Merge remote-tracking branch 'origin/dev' into feature/1812-traverse-actions-dict

tests/e2e_integration_tests/test_uc2_data_manipulation_scenario.py

@@ -1,11 +1,12 @@
 from primaite.simulator.network.hardware.nodes.computer import Computer
 from primaite.simulator.network.hardware.nodes.server import Server
 from primaite.simulator.system.applications.database_client import DatabaseClient
-from primaite.simulator.system.services.database_service import DatabaseService
+from primaite.simulator.system.services.database.database_service import DatabaseService
 from primaite.simulator.system.services.red_services.data_manipulation_bot import DataManipulationBot
 
 
 def test_data_manipulation(uc2_network):
+    """Tests the UC2 data manipulation scenario end-to-end. Is a work in progress."""
     client_1: Computer = uc2_network.get_node_by_hostname("client_1")
     db_manipulation_bot: DataManipulationBot = client_1.software_manager.software["DataManipulationBot"]
 
@@ -15,6 +16,8 @@ def test_data_manipulation(uc2_network):
     web_server: Server = uc2_network.get_node_by_hostname("web_server")
     db_client: DatabaseClient = web_server.software_manager.software["DatabaseClient"]
 
+    db_service.backup_database()
+
     # First check that the DB client on the web_server can successfully query the users table on the database
     assert db_client.query("SELECT * FROM user;")
 
@@ -23,3 +26,9 @@ def test_data_manipulation(uc2_network):
 
     # Now check that the DB client on the web_server cannot query the users table on the database
     assert not db_client.query("SELECT * FROM user;")
+
+    # Now restore the database
+    db_service.restore_backup()
+
+    # Now check that the DB client on the web_server can successfully query the users table on the database
+    assert db_client.query("SELECT * FROM user;")

tests/integration_tests/system/test_database_on_node.py

@@ -2,7 +2,8 @@ from ipaddress import IPv4Address
 
 from primaite.simulator.network.hardware.nodes.server import Server
 from primaite.simulator.system.applications.database_client import DatabaseClient
-from primaite.simulator.system.services.database_service import DatabaseService
+from primaite.simulator.system.services.database.database_service import DatabaseService
+from primaite.simulator.system.services.ftp.ftp_server import FTPServer
 
 
 def test_database_client_server_connection(uc2_network):
@@ -57,3 +58,37 @@ def test_database_client_query(uc2_network):
     db_client.connect()
 
     assert db_client.query("SELECT * FROM user;")
+
+
+def test_create_database_backup(uc2_network):
+    """Run the backup_database method and check if the FTP server has the relevant file."""
+    db_server: Server = uc2_network.get_node_by_hostname("database_server")
+    db_service: DatabaseService = db_server.software_manager.software["DatabaseService"]
+
+    # back up should be created
+    assert db_service.backup_database() is True
+
+    backup_server: Server = uc2_network.get_node_by_hostname("backup_server")
+    ftp_server: FTPServer = backup_server.software_manager.software["FTPServer"]
+
+    # backup file should exist in the backup server
+    assert ftp_server.file_system.get_file(folder_name=db_service.uuid, file_name="database.db") is not None
+
+
+def test_restore_backup(uc2_network):
+    """Run the restore_backup method and check if the backup is properly restored."""
+    db_server: Server = uc2_network.get_node_by_hostname("database_server")
+    db_service: DatabaseService = db_server.software_manager.software["DatabaseService"]
+
+    # create a back up
+    assert db_service.backup_database() is True
+
+    # delete database locally
+    db_service.file_system.delete_file(folder_name="database", file_name="database.db")
+
+    assert db_service.file_system.get_file(folder_name="database", file_name="database.db") is None
+
+    # back up should be restored
+    assert db_service.restore_backup() is True
+
+    assert db_service.file_system.get_file(folder_name="database", file_name="database.db") is not None

tests/integration_tests/system/test_dns_client_server.py

@@ -1,9 +1,7 @@
-from ipaddress import IPv4Address
-
 from primaite.simulator.network.hardware.nodes.computer import Computer
 from primaite.simulator.network.hardware.nodes.server import Server
-from primaite.simulator.system.services.dns_client import DNSClient
-from primaite.simulator.system.services.dns_server import DNSServer
+from primaite.simulator.system.services.dns.dns_client import DNSClient
+from primaite.simulator.system.services.dns.dns_server import DNSServer
 from primaite.simulator.system.services.service import ServiceOperatingState
 
 

tests/integration_tests/system/test_ftp_client_server.py

@@ -0,0 +1,62 @@
+from ipaddress import IPv4Address
+
+from primaite.simulator.network.hardware.nodes.computer import Computer
+from primaite.simulator.network.hardware.nodes.server import Server
+from primaite.simulator.system.services.ftp.ftp_client import FTPClient
+from primaite.simulator.system.services.ftp.ftp_server import FTPServer
+from primaite.simulator.system.services.service import ServiceOperatingState
+
+
+def test_ftp_client_store_file_in_server(uc2_network):
+    """
+    Test checks to see if the client is able to store files in the backup server.
+    """
+    client_1: Computer = uc2_network.get_node_by_hostname("client_1")
+    backup_server: Server = uc2_network.get_node_by_hostname("backup_server")
+
+    ftp_client: FTPClient = client_1.software_manager.software["FTPClient"]
+    ftp_server: FTPServer = backup_server.software_manager.software["FTPServer"]
+
+    assert ftp_client.operating_state == ServiceOperatingState.RUNNING
+    assert ftp_server.operating_state == ServiceOperatingState.RUNNING
+
+    # create file on ftp client
+    ftp_client.file_system.create_file(file_name="test_file.txt")
+
+    assert ftp_client.send_file(
+        src_folder_name="root",
+        src_file_name="test_file.txt",
+        dest_folder_name="client_1_backup",
+        dest_file_name="test_file.txt",
+        dest_ip_address=backup_server.nics.get(next(iter(backup_server.nics))).ip_address,
+    )
+
+    assert ftp_server.file_system.get_file(folder_name="client_1_backup", file_name="test_file.txt")
+
+
+def test_ftp_client_retrieve_file_from_server(uc2_network):
+    """
+    Test checks to see if the client is able to retrieve files from the backup server.
+    """
+    client_1: Computer = uc2_network.get_node_by_hostname("client_1")
+    backup_server: Server = uc2_network.get_node_by_hostname("backup_server")
+
+    ftp_client: FTPClient = client_1.software_manager.software["FTPClient"]
+    ftp_server: FTPServer = backup_server.software_manager.software["FTPServer"]
+
+    assert ftp_client.operating_state == ServiceOperatingState.RUNNING
+    assert ftp_server.operating_state == ServiceOperatingState.RUNNING
+
+    # create file on ftp server
+    ftp_server.file_system.create_file(file_name="test_file.txt", folder_name="file_share")
+
+    assert ftp_client.request_file(
+        src_folder_name="file_share",
+        src_file_name="test_file.txt",
+        dest_folder_name="downloads",
+        dest_file_name="test_file.txt",
+        dest_ip_address=backup_server.nics.get(next(iter(backup_server.nics))).ip_address,
+    )
+
+    # client should have retrieved the file
+    assert ftp_client.file_system.get_file(folder_name="downloads", file_name="test_file.txt")

tests/unit_tests/_primaite/_simulator/_system/_services/test_database.py

@@ -1,9 +1,7 @@
-import json
-
 import pytest
 
 from primaite.simulator.network.hardware.base import Node
-from primaite.simulator.system.services.database_service import DatabaseService
+from primaite.simulator.system.services.database.database_service import DatabaseService
 
 
 @pytest.fixture(scope="function")

tests/unit_tests/_primaite/_simulator/_system/_services/test_dns.py

@@ -3,16 +3,20 @@ from ipaddress import IPv4Address
 import pytest
 
 from primaite.simulator.network.hardware.base import Node
+from primaite.simulator.network.hardware.nodes.computer import Computer
+from primaite.simulator.network.hardware.nodes.server import Server
 from primaite.simulator.network.protocols.dns import DNSPacket, DNSReply, DNSRequest
 from primaite.simulator.network.transmission.network_layer import IPProtocol
 from primaite.simulator.network.transmission.transport_layer import Port
-from primaite.simulator.system.services.dns_client import DNSClient
-from primaite.simulator.system.services.dns_server import DNSServer
+from primaite.simulator.system.services.dns.dns_client import DNSClient
+from primaite.simulator.system.services.dns.dns_server import DNSServer
 
 
 @pytest.fixture(scope="function")
 def dns_server() -> Node:
-    node = Node(hostname="dns_server")
+    node = Server(
+        hostname="dns_server", ip_address="192.168.1.10", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
+    )
     node.software_manager.install(software_class=DNSServer)
     node.software_manager.software["DNSServer"].start()
     return node
@@ -20,9 +24,9 @@ def dns_server() -> Node:
 
 @pytest.fixture(scope="function")
 def dns_client() -> Node:
-    node = Node(hostname="dns_client")
-    node.software_manager.install(software_class=DNSClient)
-    node.software_manager.software["DNSClient"].start()
+    node = Computer(
+        hostname="dns_client", ip_address="192.168.1.11", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
+    )
     return node
 
 

tests/unit_tests/_primaite/_simulator/_system/_services/test_ftp.py

@@ -0,0 +1,86 @@
+from ipaddress import IPv4Address
+
+import pytest
+
+from primaite.simulator.network.hardware.base import Node
+from primaite.simulator.network.hardware.nodes.computer import Computer
+from primaite.simulator.network.hardware.nodes.server import Server
+from primaite.simulator.network.protocols.ftp import FTPCommand, FTPPacket
+from primaite.simulator.network.transmission.network_layer import IPProtocol
+from primaite.simulator.network.transmission.transport_layer import Port
+from primaite.simulator.system.services.ftp.ftp_client import FTPClient
+from primaite.simulator.system.services.ftp.ftp_server import FTPServer
+
+
+@pytest.fixture(scope="function")
+def ftp_server() -> Node:
+    node = Server(
+        hostname="ftp_server", ip_address="192.168.1.10", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
+    )
+    node.software_manager.install(software_class=FTPServer)
+    node.software_manager.software["FTPServer"].start()
+    return node
+
+
+@pytest.fixture(scope="function")
+def ftp_client() -> Node:
+    node = Computer(
+        hostname="ftp_client", ip_address="192.168.1.11", subnet_mask="255.255.255.0", default_gateway="192.168.1.1"
+    )
+    return node
+
+
+def test_create_ftp_server(ftp_server):
+    assert ftp_server is not None
+    ftp_server_service: FTPServer = ftp_server.software_manager.software["FTPServer"]
+    assert ftp_server_service.name is "FTPServer"
+    assert ftp_server_service.port is Port.FTP
+    assert ftp_server_service.protocol is IPProtocol.TCP
+
+
+def test_create_ftp_client(ftp_client):
+    assert ftp_client is not None
+    ftp_client_service: FTPClient = ftp_client.software_manager.software["FTPClient"]
+    assert ftp_client_service.name is "FTPClient"
+    assert ftp_client_service.port is Port.FTP
+    assert ftp_client_service.protocol is IPProtocol.TCP
+
+
+def test_ftp_server_store_file(ftp_server):
+    """Test to make sure the FTP Server knows how to deal with request responses."""
+    assert ftp_server.file_system.get_file(folder_name="downloads", file_name="file.txt") is None
+
+    response: FTPPacket = FTPPacket(
+        ftp_command=FTPCommand.STOR,
+        ftp_command_args={
+            "dest_folder_name": "downloads",
+            "dest_file_name": "file.txt",
+            "file_size": 24,
+        },
+        packet_payload_size=24,
+    )
+
+    ftp_server_service: FTPServer = ftp_server.software_manager.software["FTPServer"]
+    ftp_server_service.receive(response)
+
+    assert ftp_server.file_system.get_file(folder_name="downloads", file_name="file.txt")
+
+
+def test_ftp_client_store_file(ftp_client):
+    """Test to make sure the FTP Client knows how to deal with request responses."""
+    assert ftp_client.file_system.get_file(folder_name="downloads", file_name="file.txt") is None
+
+    response: FTPPacket = FTPPacket(
+        ftp_command=FTPCommand.STOR,
+        ftp_command_args={
+            "dest_folder_name": "downloads",
+            "dest_file_name": "file.txt",
+            "file_size": 24,
+        },
+        packet_payload_size=24,
+    )
+
+    ftp_client_service: FTPClient = ftp_client.software_manager.software["FTPClient"]
+    ftp_client_service.receive(response)
+
+    assert ftp_client.file_system.get_file(folder_name="downloads", file_name="file.txt")