From 32c2ea0b100e39e6db28b14e1f939852c7ca2c21 Mon Sep 17 00:00:00 2001 From: Charlie Crane Date: Mon, 15 Jul 2024 08:22:18 +0100 Subject: [PATCH] #2710 - Pre-commit run ahead of raising PR --- .../services/database/database_service.py | 4 +- .../system/services/terminal/terminal.py | 38 +++++++++---------- .../_system/_services/test_terminal.py | 31 ++++++++++----- 3 files changed, 41 insertions(+), 32 deletions(-) diff --git a/src/primaite/simulator/system/services/database/database_service.py b/src/primaite/simulator/system/services/database/database_service.py index d6feafbd..f061b3c7 100644 --- a/src/primaite/simulator/system/services/database/database_service.py +++ b/src/primaite/simulator/system/services/database/database_service.py @@ -19,9 +19,9 @@ _LOGGER = getLogger(__name__) class DatabaseService(Service): """ -A class for simulating a generic SQL Server service. + A class for simulating a generic SQL Server service. - This class inherits from the `Service` class and provides methods to simulate a SQL database. + This class inherits from the `Service` class and provides methods to simulate a SQL database. """ password: Optional[str] = None diff --git a/src/primaite/simulator/system/services/terminal/terminal.py b/src/primaite/simulator/system/services/terminal/terminal.py index 3324c4e4..589492ba 100644 --- a/src/primaite/simulator/system/services/terminal/terminal.py +++ b/src/primaite/simulator/system/services/terminal/terminal.py @@ -117,14 +117,13 @@ class Terminal(Service): def login(self, dest_ip_address: IPv4Address, **kwargs) -> bool: """Process User request to login to Terminal. - - :param dest_ip_address: The IP address of the node we want to connect to. + + :param dest_ip_address: The IP address of the node we want to connect to. :return: True if successful, False otherwise. """ if self.operating_state != ServiceOperatingState.RUNNING: self.sys_log.warning("Cannot process login as service is not running") return False - user_account = f"Username: placeholder, Password: placeholder" if self.connection_uuid in self.user_connections: self.sys_log.debug("User authentication passed") return True @@ -133,9 +132,9 @@ class Terminal(Service): # TODO: Refactor with UserManager changes to provide correct credentials and validate. transport_message = SSHTransportMessage.SSH_MSG_USERAUTH_REQUEST connection_message = SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN - payload: SSHPacket = SSHPacket(payload="login", - transport_message=transport_message, - connection_message=connection_message) + payload: SSHPacket = SSHPacket( + payload="login", transport_message=transport_message, connection_message=connection_message + ) self.sys_log.debug(f"Sending login request to {dest_ip_address}") self.send(payload=payload, dest_ip_address=dest_ip_address) @@ -158,8 +157,11 @@ class Terminal(Service): self.sys_log.info(f"{self.name}: Connect request for ID: {self.connection_uuid} authorised") transport_message = SSHTransportMessage.SSH_MSG_USERAUTH_SUCCESS connection_message = SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN_CONFIRMATION - new_connection = TerminalClientConnection(parent_node = self.software_manager.node, - connection_id=self.connection_uuid, dest_ip_address=dest_ip_address) + new_connection = TerminalClientConnection( + parent_node=self.software_manager.node, + connection_id=self.connection_uuid, + dest_ip_address=dest_ip_address, + ) self.user_connections[self.connection_uuid] = new_connection self.is_connected = True @@ -170,7 +172,7 @@ class Terminal(Service): def _ssh_process_logoff(self, session_id: str, *args, **kwargs) -> bool: """Process the logoff attempt. Return a bool if succesful or unsuccessful.""" - # TODO: Should remove + # TODO: Should remove def receive(self, payload: SSHPacket, session_id: str, **kwargs) -> bool: """Receive Payload and process for a response.""" @@ -178,7 +180,7 @@ class Terminal(Service): return False if self.operating_state != ServiceOperatingState.RUNNING: - self.sys_log.warning(f"Cannot process message as not running") + self.sys_log.warning("Cannot process message as not running") return False self.sys_log.debug(f"Received payload: {payload} from session: {session_id}") @@ -212,7 +214,7 @@ class Terminal(Service): """Remote login to terminal via SSH.""" if not user_account: # TODO: Generic hardcoded info, will need to be updated with UserManager. - user_account = f"Username: placeholder, Password: placeholder" + user_account = "Username: placeholder, Password: placeholder" # something like self.user_manager.get_user_details ? # Implement SSHPacket class @@ -242,8 +244,8 @@ class Terminal(Service): return False def disconnect(self, dest_ip_address: IPv4Address) -> bool: - """Disconnect from remote connection. - + """Disconnect from remote connection. + :param dest_ip_address: The IP address fo the connection we are terminating. :return: True if successful, False otherwise. """ @@ -263,15 +265,13 @@ class Terminal(Service): software_manager.send_payload_to_session_manager( payload={"type": "disconnect", "connection_id": self.connection_uuid}, dest_ip_address=dest_ip_address, - dest_port=self.port + dest_port=self.port, ) connection = self.user_connections.pop(self.connection_uuid) connection.is_active = False - self.sys_log.info( - f"{self.name}: Disconnected {self.connection_uuid}" - ) + self.sys_log.info(f"{self.name}: Disconnected {self.connection_uuid}") return True def send( @@ -289,6 +289,4 @@ class Terminal(Service): self.sys_log.warning(f"Cannot send commands when Operating state is {self.operating_state}!") return False self.sys_log.debug(f"Sending payload: {payload}") - return super().send( - payload=payload, dest_ip_address=dest_ip_address, dest_port=self.port - ) + return super().send(payload=payload, dest_ip_address=dest_ip_address, dest_port=self.port) diff --git a/tests/unit_tests/_primaite/_simulator/_system/_services/test_terminal.py b/tests/unit_tests/_primaite/_simulator/_system/_services/test_terminal.py index 62933b5c..6b0365ce 100644 --- a/tests/unit_tests/_primaite/_simulator/_system/_services/test_terminal.py +++ b/tests/unit_tests/_primaite/_simulator/_system/_services/test_terminal.py @@ -11,14 +11,18 @@ from primaite.simulator.system.services.service import ServiceOperatingState from primaite.simulator.system.services.terminal.terminal import Terminal from primaite.simulator.system.software import SoftwareHealthState + @pytest.fixture(scope="function") def terminal_on_computer() -> Tuple[Terminal, Computer]: - computer: Computer = Computer(hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0) + computer: Computer = Computer( + hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0 + ) computer.power_on() terminal: Terminal = computer.software_manager.software.get("Terminal") return [terminal, computer] + @pytest.fixture(scope="function") def basic_network() -> Network: network = Network() @@ -37,6 +41,7 @@ def test_terminal_creation(terminal_on_computer): terminal, computer = terminal_on_computer terminal.describe_state() + def test_terminal_install_default(): """Terminal should be auto installed onto Nodes""" computer = Computer(hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0) @@ -44,22 +49,25 @@ def test_terminal_install_default(): assert computer.software_manager.software.get("Terminal") + def test_terminal_not_on_switch(): """Ensure terminal does not auto-install to switch""" test_switch = Switch(hostname="Test") assert not test_switch.software_manager.software.get("Terminal") + def test_terminal_send(basic_network): - """Check that Terminal can send """ + """Check that Terminal can send""" network: Network = basic_network computer_a: Computer = network.get_node_by_hostname("node_a") terminal_a: Terminal = computer_a.software_manager.software.get("Terminal") - payload: SSHPacket = SSHPacket(payload="Test_Payload", - transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST, - connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN) - + payload: SSHPacket = SSHPacket( + payload="Test_Payload", + transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST, + connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_OPEN, + ) assert terminal_a.send(payload=payload, dest_ip_address="192.168.0.11") @@ -91,6 +99,7 @@ def test_terminal_disconnect(basic_network): assert terminal.is_connected is False + def test_terminal_ignores_when_off(basic_network): """Terminal should ignore commands when not running""" network: Network = basic_network @@ -99,14 +108,16 @@ def test_terminal_ignores_when_off(basic_network): computer_b: Computer = network.get_node_by_hostname("node_b") - terminal_a.login(dest_ip_address="192.168.0.11") # login to computer_b + terminal_a.login(dest_ip_address="192.168.0.11") # login to computer_b assert terminal_a.is_connected is True terminal_a.operating_state = ServiceOperatingState.STOPPED - payload: SSHPacket = SSHPacket(payload="Test_Payload", - transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST, - connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_DATA) + payload: SSHPacket = SSHPacket( + payload="Test_Payload", + transport_message=SSHTransportMessage.SSH_MSG_SERVICE_REQUEST, + connection_message=SSHConnectionMessage.SSH_MSG_CHANNEL_DATA, + ) assert not terminal_a.send(payload=payload, dest_ip_address="192.168.0.11")