add unit tests

Author: mlikasam-askui

tests/unit/tools/askui/test_agent_os_server.py

@@ -0,0 +1,137 @@
+import re
+
+import pytest
+
+from askui.tools.askui.agent_os_server import (
+    AgentOsServer,
+    LocalAgentOsServer,
+    RemoteAgentOsServer,
+    _generate_session_guid,
+    _replace_port,
+)
+
+
+class TestSessionGuid:
+    def test_generated_guid_is_brace_wrapped_uuid(self) -> None:
+        guid = _generate_session_guid()
+        assert re.fullmatch(
+            r"\{[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\}",
+            guid,
+        )
+
+    def test_each_generated_guid_is_unique(self) -> None:
+        assert _generate_session_guid() != _generate_session_guid()
+
+
+class TestReplacePort:
+    def test_replaces_port_on_bare_authority(self) -> None:
+        assert _replace_port("example.com:1234", 23000) == "example.com:23000"
+
+    def test_replaces_port_on_url_with_scheme(self) -> None:
+        assert _replace_port("http://example.com:1234", 23000) == "example.com:23000"
+
+    def test_falls_back_to_localhost_when_host_missing(self) -> None:
+        # A bare ":1234" has no hostname, so the helper falls back to "localhost".
+        assert _replace_port(":1234", 23000) == "localhost:23000"
+
+
+class TestAgentOsServer:
+    def test_session_guid_unique_per_instance(self) -> None:
+        a = RemoteAgentOsServer(address="1.2.3.4:23000", description="a")
+        b = RemoteAgentOsServer(address="5.6.7.8:23000", description="b")
+        assert a.session_guid != b.session_guid
+
+    def test_computer_id_defaults_to_session_guid(self) -> None:
+        s = RemoteAgentOsServer(address="1.2.3.4:23000", description="a")
+        assert s.computer_id == s.session_guid
+
+    def test_explicit_computer_id_is_preserved(self) -> None:
+        s = RemoteAgentOsServer(
+            address="1.2.3.4:23000", description="a", computer_id="laptop"
+        )
+        assert s.computer_id == "laptop"
+        assert s.session_guid != "laptop"
+
+    def test_display_defaults_to_one_and_is_settable(self) -> None:
+        s = RemoteAgentOsServer(address="1.2.3.4:23000", description="a")
+        assert s.display == 1
+        s.display = 3
+        assert s.display == 3
+
+    def test_explicit_display_is_preserved(self) -> None:
+        s = RemoteAgentOsServer(address="1.2.3.4:23000", description="a", display=2)
+        assert s.display == 2
+
+    def test_repr_contains_identity_fields(self) -> None:
+        s = RemoteAgentOsServer(
+            address="1.2.3.4:23000",
+            description="my rig",
+            display=2,
+            computer_id="rig",
+        )
+        r = repr(s)
+        assert "RemoteAgentOsServer" in r
+        assert "computer_id='rig'" in r
+        assert "description='my rig'" in r
+        assert "display=2" in r
+
+    def test_base_class_is_not_local(self) -> None:
+        s = RemoteAgentOsServer(address="1.2.3.4:23000", description="a")
+        assert s.is_local is False
+
+    def test_start_and_stop_are_no_ops_on_remote(self) -> None:
+        s = RemoteAgentOsServer(address="1.2.3.4:23000", description="a")
+        s.start()
+        s.stop()
+
+
+class TestLocalAgentOsServer:
+    def test_is_local(self) -> None:
+        s = LocalAgentOsServer(discover_service=False)
+        assert s.is_local is True
+
+    def test_default_description(self) -> None:
+        s = LocalAgentOsServer(discover_service=False)
+        assert s.description == "Local Agent OS server"
+
+    def test_default_address(self) -> None:
+        s = LocalAgentOsServer(discover_service=False)
+        assert s.address == "localhost:23000"
+
+    def test_is_service_default_false(self) -> None:
+        s = LocalAgentOsServer(discover_service=False)
+        assert s.is_service is False
+
+    def test_explicit_computer_id(self) -> None:
+        s = LocalAgentOsServer(discover_service=False, computer_id="my-laptop")
+        assert s.computer_id == "my-laptop"
+
+    def test_parse_port_rejects_bad_address(self) -> None:
+        s = LocalAgentOsServer(discover_service=False, address="no-port-here")
+        with pytest.raises(ValueError, match="Could not parse port"):
+            s._parse_port()  # noqa: SLF001 - intentional unit test against helper
+
+    def test_parse_port_extracts_port(self) -> None:
+        s = LocalAgentOsServer(discover_service=False, address="localhost:24567")
+        assert s._parse_port() == 24567  # noqa: SLF001
+
+
+class TestSubclassesPassThroughDisplayAndId:
+    @pytest.mark.parametrize(
+        "factory",
+        [
+            lambda: LocalAgentOsServer(
+                discover_service=False, display=4, computer_id="local"
+            ),
+            lambda: RemoteAgentOsServer(
+                address="1.2.3.4:23000",
+                description="r",
+                display=4,
+                computer_id="remote",
+            ),
+        ],
+    )
+    def test_display_and_computer_id_round_trip(self, factory) -> None:  # noqa: ANN001
+        s: AgentOsServer = factory()
+        assert s.display == 4
+        assert s.computer_id in {"local", "remote"}

tests/unit/tools/askui/test_agent_os_server_manager.py

@@ -0,0 +1,187 @@
+import pytest
+
+from askui.tools.askui.agent_os_server import (
+    LocalAgentOsServer,
+    RemoteAgentOsServer,
+)
+from askui.tools.askui.agent_os_server_manager import AgentOsServerManager
+
+
+def _make_remote(
+    address: str = "1.2.3.4:23000",
+    description: str = "remote",
+    computer_id: str | None = None,
+) -> RemoteAgentOsServer:
+    return RemoteAgentOsServer(
+        address=address, description=description, computer_id=computer_id
+    )
+
+
+def _make_local(computer_id: str | None = None) -> LocalAgentOsServer:
+    return LocalAgentOsServer(discover_service=False, computer_id=computer_id)
+
+
+class TestConstruction:
+    def test_empty_constructor_yields_empty_manager(self) -> None:
+        m = AgentOsServerManager()
+        assert m.list() == []
+        assert m.active is None
+        assert len(m) == 0
+
+    def test_constructor_registers_initial_servers_in_order(self) -> None:
+        a = _make_remote(address="1.1.1.1:23000", computer_id="a")
+        b = _make_remote(address="2.2.2.2:23000", computer_id="b")
+        m = AgentOsServerManager(agent_os_servers=[a, b])
+        assert m.list() == [a, b]
+        # First registered becomes active.
+        assert m.active is a
+
+    def test_first_added_becomes_active(self) -> None:
+        m = AgentOsServerManager()
+        a = _make_remote(computer_id="a")
+        m.add(a)
+        assert m.active is a
+
+
+class TestAddConstraints:
+    def test_rejects_second_local_server(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_local(computer_id="first"))
+        with pytest.raises(ValueError, match="second local Agent OS server"):
+            m.add(_make_local(computer_id="second"))
+
+    def test_rejects_duplicate_computer_id(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(address="1.1.1.1:23000", computer_id="rig"))
+        with pytest.raises(ValueError, match="computer_id='rig'"):
+            m.add(_make_remote(address="2.2.2.2:23000", computer_id="rig"))
+
+    def test_rejects_duplicate_remote_address(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(address="1.1.1.1:23000", computer_id="a"))
+        with pytest.raises(
+            ValueError, match="remote Agent OS server with address '1.1.1.1:23000'"
+        ):
+            m.add(_make_remote(address="1.1.1.1:23000", computer_id="b"))
+
+    def test_allows_local_plus_remote_with_same_address(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_local(computer_id="local"))
+        # Local target's default address is 'localhost:23000' but the local/remote
+        # address-uniqueness rule only applies between remote servers.
+        m.add(
+            _make_remote(
+                address="localhost:23000", description="remote", computer_id="remote"
+            )
+        )
+        assert len(m) == 2
+
+
+class TestAddRemote:
+    def test_constructs_and_registers(self) -> None:
+        m = AgentOsServerManager()
+        server = m.add_remote(address="1.2.3.4:23000", description="r")
+        assert isinstance(server, RemoteAgentOsServer)
+        assert server.address == "1.2.3.4:23000"
+        assert server.description == "r"
+        assert m.list() == [server]
+
+
+class TestGetAndSwitch:
+    def test_get_returns_server_by_computer_id(self) -> None:
+        m = AgentOsServerManager()
+        a = _make_remote(address="1.1.1.1:23000", computer_id="a")
+        m.add(a)
+        assert m.get("a") is a
+
+    def test_get_raises_keyerror_with_registered_ids(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(address="1.1.1.1:23000", computer_id="a"))
+        with pytest.raises(KeyError) as exc_info:
+            m.get("missing")
+        message = str(exc_info.value)
+        assert "missing" in message
+        assert "'a'" in message  # registered id surfaced
+
+    def test_switch_changes_active(self) -> None:
+        m = AgentOsServerManager()
+        a = _make_remote(address="1.1.1.1:23000", computer_id="a")
+        b = _make_remote(address="2.2.2.2:23000", computer_id="b")
+        m.add(a)
+        m.add(b)
+        assert m.active is a
+        m.switch("b")
+        assert m.active is b
+
+    def test_switch_unknown_id_raises_keyerror(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(computer_id="a"))
+        with pytest.raises(KeyError, match="missing"):
+            m.switch("missing")
+
+
+class TestGetBySessionGuid:
+    def test_returns_server_by_session_guid(self) -> None:
+        m = AgentOsServerManager()
+        a = _make_remote(address="1.1.1.1:23000", computer_id="a")
+        m.add(a)
+        assert m.get_by_session_guid(a.session_guid) is a
+
+    def test_returns_none_for_unknown_session_guid(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(computer_id="a"))
+        assert m.get_by_session_guid("no-such-guid") is None
+
+
+class TestRemove:
+    def test_remove_drops_server(self) -> None:
+        m = AgentOsServerManager()
+        a = _make_remote(address="1.1.1.1:23000", computer_id="a")
+        b = _make_remote(address="2.2.2.2:23000", computer_id="b")
+        m.add(a)
+        m.add(b)
+        m.remove("a")
+        assert m.list() == [b]
+
+    def test_remove_active_falls_back_to_first_remaining(self) -> None:
+        m = AgentOsServerManager()
+        a = _make_remote(address="1.1.1.1:23000", computer_id="a")
+        b = _make_remote(address="2.2.2.2:23000", computer_id="b")
+        m.add(a)
+        m.add(b)
+        assert m.active is a
+        m.remove("a")
+        assert m.active is b
+
+    def test_remove_last_clears_active(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(computer_id="a"))
+        m.remove("a")
+        assert m.active is None
+        assert len(m) == 0
+
+    def test_remove_inactive_keeps_active_unchanged(self) -> None:
+        m = AgentOsServerManager()
+        a = _make_remote(address="1.1.1.1:23000", computer_id="a")
+        b = _make_remote(address="2.2.2.2:23000", computer_id="b")
+        m.add(a)
+        m.add(b)
+        m.remove("b")
+        assert m.active is a
+
+    def test_remove_unknown_raises_keyerror(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(computer_id="a"))
+        with pytest.raises(KeyError):
+            m.remove("missing")
+
+
+class TestReset:
+    def test_reset_clears_all(self) -> None:
+        m = AgentOsServerManager()
+        m.add(_make_remote(computer_id="a"))
+        m.add(_make_remote(address="2.2.2.2:23000", computer_id="b"))
+        m.reset()
+        assert m.list() == []
+        assert m.active is None
+        assert len(m) == 0