Merge branch 'dev' into bugfix/2299-check_hash_function_corrupts_files_and_folders

tests/unit_tests/_primaite/_session/test_episode_schedule.py

@@ -0,0 +1,50 @@
+import pytest
+import yaml
+
+from primaite.session.episode_schedule import ConstantEpisodeScheduler, EpisodeListScheduler
+
+
+def test_episode_list_scheduler():
+    # Initialize an instance of EpisodeListScheduler
+
+    # Define a schedule and episode data for testing
+    schedule = {0: ["episode1"], 1: ["episode2"]}
+    episode_data = {"episode1": "data1: 1", "episode2": "data2: 2"}
+    base_scenario = """agents: []"""
+
+    scheduler = EpisodeListScheduler(schedule=schedule, episode_data=episode_data, base_scenario=base_scenario)
+    # Test when episode number is within the schedule
+    result = scheduler(0)
+    assert isinstance(result, dict)
+    assert yaml.safe_load("data1: 1\nagents: []") == result
+
+    # Test next episode
+    result = scheduler(1)
+    assert isinstance(result, dict)
+    assert yaml.safe_load("data2: 2\nagents: []") == result
+
+    # Test when episode number exceeds the schedule
+    result = scheduler(2)
+    assert isinstance(result, dict)
+    assert yaml.safe_load("data1: 1\nagents: []") == result
+    assert scheduler._exceeded_episode_list
+
+    # Test when episode number is a sequence
+    scheduler.schedule = {0: ["episode1", "episode2"]}
+    result = scheduler(0)
+    assert isinstance(result, dict)
+    assert yaml.safe_load("data1: 1\ndata2: 2\nagents: []") == result
+
+
+def test_constant_episode_scheduler():
+    # Initialize an instance of ConstantEpisodeScheduler
+    config = {"key": "value"}
+    scheduler = ConstantEpisodeScheduler(config=config)
+
+    result = scheduler(0)
+    assert isinstance(result, dict)
+    assert {"key": "value"} == result
+
+    result = scheduler(1)
+    assert isinstance(result, dict)
+    assert {"key": "value"} == result

tests/unit_tests/_primaite/_simulator/_system/_applications/_red_applications/test_data_manipulation_bot.py

@@ -70,7 +70,7 @@ def test_dm_bot_perform_data_manipulation_success(dm_bot):
     dm_bot._perform_data_manipulation(p_of_success=1.0)
 
     assert dm_bot.attack_stage in (DataManipulationAttackStage.SUCCEEDED, DataManipulationAttackStage.FAILED)
-    assert len(dm_bot._host_db_client.connections)
+    assert len(dm_bot._host_db_client.client_connections)
 
 
 def test_dm_bot_fails_without_db_client(dm_client):

tests/unit_tests/_primaite/_simulator/_system/_applications/test_database_client.py

@@ -56,7 +56,11 @@ def test_connect_to_database_fails_on_reattempt(database_client_on_computer):
     database_client, computer = database_client_on_computer
 
     database_client.connected = False
-    assert database_client._connect(server_ip_address=IPv4Address("192.168.0.1"), is_reattempt=True) is False
+
+    database_connection = database_client._connect(
+        server_ip_address=IPv4Address("192.168.0.1"), connection_request_id="", is_reattempt=True
+    )
+    assert database_connection is None
 
 
 def test_disconnect_when_client_is_closed(database_client_on_computer):
@@ -79,21 +83,20 @@ def test_disconnect(database_client_on_computer):
     """Database client should remove the connection."""
     database_client, computer = database_client_on_computer
 
-    assert not database_client.connected
+    assert database_client.connected is False
 
     database_client.connect()
 
-    assert database_client.connected
+    assert database_client.connected is True
 
     database_client.disconnect()
 
-    assert not database_client.connected
+    assert database_client.connected is False
 
 
 def test_query_when_client_is_closed(database_client_on_computer):
     """Database client should return False when it is not running."""
     database_client, computer = database_client_on_computer
-
     database_client.close()
     assert database_client.operating_state is ApplicationOperatingState.CLOSED
 

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

@@ -1,5 +1,7 @@
 from uuid import uuid4
 
+import pytest
+
 from primaite.simulator.system.services.service import ServiceOperatingState
 from primaite.simulator.system.software import SoftwareHealthState
 
@@ -179,7 +181,8 @@ def test_overwhelm_service(service):
     assert service.health_state_actual is SoftwareHealthState.OVERWHELMED
 
 
-def test_create_and_remove_connections(service):
+@pytest.mark.xfail(reason="Fails as it's now too simple. Needs to be be refactored so that uses a service on a node.")
+def test_create_and_terminate_connections(service):
     service.start()
     uuid = str(uuid4())
 
@@ -187,6 +190,6 @@ def test_create_and_remove_connections(service):
     assert len(service.connections) == 1
     assert service.health_state_actual is SoftwareHealthState.GOOD
 
-    assert service.remove_connection(connection_id=uuid)  # should be true
+    assert service.terminate_connection(connection_id=uuid)  # should be true
     assert len(service.connections) == 0
     assert service.health_state_actual is SoftwareHealthState.GOOD

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

@@ -0,0 +1,126 @@
+from uuid import uuid4
+
+import pytest
+
+from primaite.simulator import LogLevel, SIM_OUTPUT
+from primaite.simulator.system.core.sys_log import SysLog
+
+
+@pytest.fixture(scope="function")
+def syslog() -> SysLog:
+    return SysLog(hostname="test")
+
+
+def test_debug_sys_log_level(syslog, capsys):
+    """Test that the debug log level logs debug syslogs and above."""
+    SIM_OUTPUT.sys_log_level = LogLevel.DEBUG
+    SIM_OUTPUT.write_sys_log_to_terminal = True
+
+    test_string = str(uuid4())
+
+    syslog.debug(test_string)
+    syslog.info(test_string)
+    syslog.warning(test_string)
+    syslog.error(test_string)
+    syslog.critical(test_string)
+
+    captured = "".join(capsys.readouterr())
+
+    assert test_string in captured
+    assert "DEBUG" in captured
+    assert "INFO" in captured
+    assert "WARNING" in captured
+    assert "ERROR" in captured
+    assert "CRITICAL" in captured
+
+
+def test_info_sys_log_level(syslog, capsys):
+    """Test that the debug log level logs debug syslogs and above."""
+    SIM_OUTPUT.sys_log_level = LogLevel.INFO
+    SIM_OUTPUT.write_sys_log_to_terminal = True
+
+    test_string = str(uuid4())
+
+    syslog.debug(test_string)
+    syslog.info(test_string)
+    syslog.warning(test_string)
+    syslog.error(test_string)
+    syslog.critical(test_string)
+
+    captured = "".join(capsys.readouterr())
+
+    assert test_string in captured
+    assert "DEBUG" not in captured
+    assert "INFO" in captured
+    assert "WARNING" in captured
+    assert "ERROR" in captured
+    assert "CRITICAL" in captured
+
+
+def test_warning_sys_log_level(syslog, capsys):
+    """Test that the debug log level logs debug syslogs and above."""
+    SIM_OUTPUT.sys_log_level = LogLevel.WARNING
+    SIM_OUTPUT.write_sys_log_to_terminal = True
+
+    test_string = str(uuid4())
+
+    syslog.debug(test_string)
+    syslog.info(test_string)
+    syslog.warning(test_string)
+    syslog.error(test_string)
+    syslog.critical(test_string)
+
+    captured = "".join(capsys.readouterr())
+
+    assert test_string in captured
+    assert "DEBUG" not in captured
+    assert "INFO" not in captured
+    assert "WARNING" in captured
+    assert "ERROR" in captured
+    assert "CRITICAL" in captured
+
+
+def test_error_sys_log_level(syslog, capsys):
+    """Test that the debug log level logs debug syslogs and above."""
+    SIM_OUTPUT.sys_log_level = LogLevel.ERROR
+    SIM_OUTPUT.write_sys_log_to_terminal = True
+
+    test_string = str(uuid4())
+
+    syslog.debug(test_string)
+    syslog.info(test_string)
+    syslog.warning(test_string)
+    syslog.error(test_string)
+    syslog.critical(test_string)
+
+    captured = "".join(capsys.readouterr())
+
+    assert test_string in captured
+    assert "DEBUG" not in captured
+    assert "INFO" not in captured
+    assert "WARNING" not in captured
+    assert "ERROR" in captured
+    assert "CRITICAL" in captured
+
+
+def test_critical_sys_log_level(syslog, capsys):
+    """Test that the debug log level logs debug syslogs and above."""
+    SIM_OUTPUT.sys_log_level = LogLevel.CRITICAL
+    SIM_OUTPUT.write_sys_log_to_terminal = True
+
+    test_string = str(uuid4())
+
+    syslog.debug(test_string)
+    syslog.info(test_string)
+    syslog.warning(test_string)
+    syslog.error(test_string)
+    syslog.critical(test_string)
+
+    captured = "".join(capsys.readouterr())
+
+    assert test_string in captured
+    assert "DEBUG" not in captured
+    assert "INFO" not in captured
+    assert "WARNING" not in captured
+    assert "ERROR" not in captured
+    assert "CRITICAL" in captured