Merge branch 'dev' into feature/2457-Set_link_bandwidth_via_config

tests/conftest.py

@@ -1,11 +1,8 @@
 # © Crown-owned copyright 2023, Defence Science and Technology Laboratory UK
-from datetime import datetime
-from pathlib import Path
-from typing import Any, Dict, Optional, Tuple, Union
+from typing import Any, Dict, Tuple
 
 import pytest
 import yaml
-from _pytest.monkeypatch import MonkeyPatch
 
 from primaite import getLogger, PRIMAITE_PATHS
 from primaite.game.agent.actions import ActionManager
@@ -13,7 +10,6 @@ from primaite.game.agent.interface import AbstractAgent
 from primaite.game.agent.observations.observation_manager import NestedObservation, ObservationManager
 from primaite.game.agent.rewards import RewardFunction
 from primaite.game.game import PrimaiteGame
-from primaite.simulator import SIM_OUTPUT
 from primaite.simulator.file_system.file_system import FileSystem
 from primaite.simulator.network.container import Network
 from primaite.simulator.network.hardware.nodes.host.computer import Computer
@@ -32,7 +28,6 @@ from primaite.simulator.system.services.dns.dns_server import DNSServer
 from primaite.simulator.system.services.service import Service
 from primaite.simulator.system.services.web_server.web_server import WebServer
 from tests import TEST_ASSETS_ROOT
-from tests.mock_and_patch.get_session_path_mock import temp_user_sessions_path
 
 ACTION_SPACE_NODE_VALUES = 1
 ACTION_SPACE_NODE_ACTION_VALUES = 1
@@ -40,21 +35,6 @@ ACTION_SPACE_NODE_ACTION_VALUES = 1
 _LOGGER = getLogger(__name__)
 
 
-@pytest.fixture(scope="function", autouse=True)
-def set_syslog_output_to_true():
-    """Will be run before each test."""
-    monkeypatch = MonkeyPatch()
-    monkeypatch.setattr(
-        SIM_OUTPUT,
-        "path",
-        Path(TEST_ASSETS_ROOT.parent.parent / "simulation_output" / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")),
-    )
-    monkeypatch.setattr(SIM_OUTPUT, "save_pcap_logs", False)
-    monkeypatch.setattr(SIM_OUTPUT, "save_sys_logs", False)
-
-    yield
-
-
 class TestService(Service):
     """Test Service class"""
 
@@ -86,7 +66,10 @@ class TestApplication(Application):
 
 @pytest.fixture(scope="function")
 def uc2_network() -> Network:
-    return arcd_uc2_network()
+    with open(PRIMAITE_PATHS.user_config_path / "example_config" / "data_manipulation.yaml") as f:
+        cfg = yaml.safe_load(f)
+    game = PrimaiteGame.from_config(cfg)
+    return game.simulation.network
 
 
 @pytest.fixture(scope="function")

tests/e2e_integration_tests/environments/test_rllib_single_agent_environment.py

@@ -38,3 +38,5 @@ def test_rllib_single_agent_compatibility():
     save_file = Path(tempfile.gettempdir()) / "ray/"
     algo.save(save_file)
     assert save_file.exists()
+
+    save_file.unlink()  # clean up

tests/e2e_integration_tests/environments/test_sb3_environment.py

@@ -25,3 +25,4 @@ def test_sb3_compatibility():
     model.save(save_path)
 
     assert (save_path).exists()
+    save_path.unlink()  # clean up

tests/e2e_integration_tests/test_uc2_data_manipulation_scenario.py

@@ -26,6 +26,9 @@ def test_data_manipulation(uc2_network):
     # First check that the DB client on the web_server can successfully query the users table on the database
     assert db_connection.query("SELECT")
 
+    db_manipulation_bot.data_manipulation_p_of_success = 1.0
+    db_manipulation_bot.port_scan_p_of_success = 1.0
+
     # Now we run the DataManipulationBot
     db_manipulation_bot.attack()
 
@@ -59,6 +62,11 @@ def test_application_install_uninstall_on_uc2():
     _, _, _, _, info = env.step(78)
     assert "DoSBot" in domcon.software_manager.software
 
+    # installing takes 3 steps so let's wait for 3 steps
+    env.step(0)
+    env.step(0)
+    env.step(0)
+
     # Test we can now execute the DoSBot app
     _, _, _, _, info = env.step(81)
     assert info["agent_actions"]["defender"].response.status == "success"

tests/integration_tests/cli/__init__.py

@@ -0,0 +1,11 @@
+from typing import List
+
+from typer.testing import CliRunner, Result
+
+from primaite.cli import app
+
+
+def cli(args: List[str]) -> Result:
+    """Pass in a list of arguments and it will return the result."""
+    runner = CliRunner()
+    return runner.invoke(app, args)

tests/integration_tests/cli/test_dev_cli.py

@@ -0,0 +1,171 @@
+import os
+import shutil
+import tempfile
+from pathlib import Path
+
+import pkg_resources
+import pytest
+import yaml
+
+from primaite import PRIMAITE_CONFIG
+from primaite.utils.cli.primaite_config_utils import update_primaite_application_config
+from tests.integration_tests.cli import cli
+
+
+@pytest.fixture(autouse=True)
+def test_setup():
+    """
+    Setup this test by using the default primaite app config in package
+    """
+    global PRIMAITE_CONFIG
+    current_config = PRIMAITE_CONFIG.copy()  # store the config before test
+
+    pkg_config_path = Path(pkg_resources.resource_filename("primaite", "setup/_package_data/primaite_config.yaml"))
+
+    with open(pkg_config_path, "r") as file:
+        # load from config
+        config_dict = yaml.safe_load(file)
+
+    PRIMAITE_CONFIG["developer_mode"] = config_dict["developer_mode"]
+
+    yield
+
+    PRIMAITE_CONFIG["developer_mode"] = current_config["developer_mode"]  # restore config to prevent being yelled at
+    update_primaite_application_config(config=PRIMAITE_CONFIG)
+
+
+def test_dev_mode_enable_disable():
+    """Test dev mode enable and disable."""
+    # check defaults
+    assert PRIMAITE_CONFIG["developer_mode"]["enabled"] is False  # not enabled by default
+
+    result = cli(["dev-mode", "show"])
+    assert "Production" in result.output  # should print that it is in Production mode by default
+
+    result = cli(["dev-mode", "enable"])
+
+    assert "Development" in result.output  # should print that it is in Development mode
+
+    assert PRIMAITE_CONFIG["developer_mode"]["enabled"]  # config should reflect that dev mode is enabled
+
+    result = cli(["dev-mode", "show"])
+    assert "Development" in result.output  # should print that it is in Development mode
+
+    result = cli(["dev-mode", "disable"])
+
+    assert "Production" in result.output  # should print that it is in Production mode
+
+    assert PRIMAITE_CONFIG["developer_mode"]["enabled"] is False  # config should reflect that dev mode is disabled
+
+    result = cli(["dev-mode", "show"])
+    assert "Production" in result.output  # should print that it is in Production mode
+
+
+def test_dev_mode_config_sys_log_level():
+    """Check that the system log level can be changed via CLI."""
+    # check defaults
+    assert PRIMAITE_CONFIG["developer_mode"]["sys_log_level"] == "DEBUG"  # DEBUG by default
+
+    result = cli(["dev-mode", "config", "-level", "WARNING"])
+
+    assert "sys_log_level=WARNING" in result.output  # should print correct value
+
+    # config should reflect that log level is WARNING
+    assert PRIMAITE_CONFIG["developer_mode"]["sys_log_level"] == "WARNING"
+
+    result = cli(["dev-mode", "config", "--sys-log-level", "INFO"])
+
+    assert "sys_log_level=INFO" in result.output  # should print correct value
+
+    # config should reflect that log level is WARNING
+    assert PRIMAITE_CONFIG["developer_mode"]["sys_log_level"] == "INFO"
+
+
+def test_dev_mode_config_sys_logs_enable_disable():
+    """Test that the system logs output can be enabled or disabled."""
+    # check defaults
+    assert PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"] is False  # False by default
+
+    result = cli(["dev-mode", "config", "--output-sys-logs"])
+    assert "output_sys_logs=True" in result.output  # should print correct value
+
+    # config should reflect that output_sys_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"]
+
+    result = cli(["dev-mode", "config", "--no-sys-logs"])
+    assert "output_sys_logs=False" in result.output  # should print correct value
+
+    # config should reflect that output_sys_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"] is False
+
+    result = cli(["dev-mode", "config", "-sys"])
+    assert "output_sys_logs=True" in result.output  # should print correct value
+
+    # config should reflect that output_sys_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"]
+
+    result = cli(["dev-mode", "config", "-nsys"])
+    assert "output_sys_logs=False" in result.output  # should print correct value
+
+    # config should reflect that output_sys_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_sys_logs"] is False
+
+
+def test_dev_mode_config_pcap_logs_enable_disable():
+    """Test that the pcap logs output can be enabled or disabled."""
+    # check defaults
+    assert PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"] is False  # False by default
+
+    result = cli(["dev-mode", "config", "--output-pcap-logs"])
+    assert "output_pcap_logs=True" in result.output  # should print correct value
+
+    # config should reflect that output_pcap_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"]
+
+    result = cli(["dev-mode", "config", "--no-pcap-logs"])
+    assert "output_pcap_logs=False" in result.output  # should print correct value
+
+    # config should reflect that output_pcap_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"] is False
+
+    result = cli(["dev-mode", "config", "-pcap"])
+    assert "output_pcap_logs=True" in result.output  # should print correct value
+
+    # config should reflect that output_pcap_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"]
+
+    result = cli(["dev-mode", "config", "-npcap"])
+    assert "output_pcap_logs=False" in result.output  # should print correct value
+
+    # config should reflect that output_pcap_logs is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_pcap_logs"] is False
+
+
+def test_dev_mode_config_output_to_terminal_enable_disable():
+    """Test that the output to terminal can be enabled or disabled."""
+    # check defaults
+    assert PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"] is False  # False by default
+
+    result = cli(["dev-mode", "config", "--output-to-terminal"])
+    assert "output_to_terminal=True" in result.output  # should print correct value
+
+    # config should reflect that output_to_terminal is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"]
+
+    result = cli(["dev-mode", "config", "--no-terminal"])
+    assert "output_to_terminal=False" in result.output  # should print correct value
+
+    # config should reflect that output_to_terminal is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"] is False
+
+    result = cli(["dev-mode", "config", "-t"])
+    assert "output_to_terminal=True" in result.output  # should print correct value
+
+    # config should reflect that output_to_terminal is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"]
+
+    result = cli(["dev-mode", "config", "-nt"])
+    assert "output_to_terminal=False" in result.output  # should print correct value
+
+    # config should reflect that output_to_terminal is True
+    assert PRIMAITE_CONFIG["developer_mode"]["output_to_terminal"] is False

tests/integration_tests/network/test_multi_lan_internet_example_network.py

@@ -83,7 +83,7 @@ def test_sometech_webserver_cannot_access_ftp_on_sometech_storage_server():
     some_tech_storage_srv.file_system.create_file(file_name="test.png")
 
     web_server: Server = network.get_node_by_hostname("some_tech_web_srv")
-
+    web_server.software_manager.install(FTPClient)
     web_ftp_client: FTPClient = web_server.software_manager.software["FTPClient"]
 
     assert not web_ftp_client.request_file(

tests/integration_tests/system/test_database_on_node.py

@@ -101,7 +101,7 @@ def test_database_client_native_connection_query(uc2_network):
     """Tests DB query across the network returns HTTP status 200 and date."""
     web_server: Server = uc2_network.get_node_by_hostname("web_server")
     db_client: DatabaseClient = web_server.software_manager.software["DatabaseClient"]
-
+    db_client.connect()
     assert db_client.query(sql="SELECT")
     assert db_client.query(sql="INSERT")
 
@@ -222,6 +222,7 @@ def test_database_client_cannot_query_offline_database_server(uc2_network):
 
     web_server: Server = uc2_network.get_node_by_hostname("web_server")
     db_client: DatabaseClient = web_server.software_manager.software.get("DatabaseClient")
+    db_client.connect()
     assert len(db_client.client_connections)
 
     # Establish a new connection to the DatabaseService

tests/unit_tests/_primaite/_simulator/_file_system/test_file.py

@@ -57,20 +57,6 @@ def test_simulated_file_check_hash(file_system):
     assert file.health_status == FileSystemItemHealthStatus.CORRUPT
 
 
-@pytest.mark.skip(reason="NODE_FILE_CHECKHASH not implemented")
-def test_real_file_check_hash(file_system):
-    file: File = file_system.create_file(file_name="test_file.txt", real=True)
-
-    file.check_hash()
-    assert file.health_status == FileSystemItemHealthStatus.GOOD
-    # change file content
-    with open(file.sim_path, "a") as f:
-        f.write("get hacked scrub lol xD\n")
-
-    file.check_hash()
-    assert file.health_status == FileSystemItemHealthStatus.CORRUPT
-
-
 def test_file_corrupt_repair_restore(file_system):
     """Test the ability to corrupt and repair files."""
     file: File = file_system.create_file(file_name="test_file.txt", folder_name="test_folder")

tests/unit_tests/_primaite/_simulator/_file_system/test_file_system.py

@@ -191,7 +191,7 @@ def test_copy_file(file_system):
     file_system.create_folder(folder_name="src_folder")
     file_system.create_folder(folder_name="dst_folder")
 
-    file = file_system.create_file(file_name="test_file.txt", size=10, folder_name="src_folder", real=True)
+    file = file_system.create_file(file_name="test_file.txt", size=10, folder_name="src_folder")
     assert file_system.num_file_creations == 1
     original_uuid = file.uuid
 

tests/unit_tests/_primaite/_simulator/_file_system/test_folder.py

@@ -132,21 +132,3 @@ def test_simulated_folder_check_hash(file_system):
     file.sim_size = 0
     folder.check_hash()
     assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
-
-
-@pytest.mark.skip(reason="NODE_FILE_CHECKHASH not implemented")
-def test_real_folder_check_hash(file_system):
-    folder: Folder = file_system.create_folder(folder_name="test_folder")
-    file_system.create_file(file_name="test_file.txt", folder_name="test_folder", real=True)
-
-    folder.check_hash()
-    assert folder.health_status == FileSystemItemHealthStatus.GOOD
-    # change simulated file size
-    file = folder.get_file(file_name="test_file.txt")
-
-    # change file content
-    with open(file.sim_path, "a") as f:
-        f.write("get hacked scrub lol xD\n")
-
-    folder.check_hash()
-    assert folder.health_status == FileSystemItemHealthStatus.CORRUPT

tests/unit_tests/_primaite/_simulator/_system/core/test_sys_log.py

@@ -2,10 +2,20 @@ from uuid import uuid4
 
 import pytest
 
+from primaite import PRIMAITE_CONFIG
 from primaite.simulator import LogLevel, SIM_OUTPUT
 from primaite.simulator.system.core.sys_log import SysLog
 
 
+@pytest.fixture(autouse=True)
+def override_dev_mode_temporarily():
+    """Temporarily turn off dev mode for this test."""
+    primaite_dev_mode = PRIMAITE_CONFIG["developer_mode"]["enabled"]
+    PRIMAITE_CONFIG["developer_mode"]["enabled"] = False
+    yield  # run tests
+    PRIMAITE_CONFIG["developer_mode"]["enabled"] = primaite_dev_mode
+
+
 @pytest.fixture(scope="function")
 def syslog() -> SysLog:
     return SysLog(hostname="test")