From 329e2b7b162a46ece1c560f5a6d88dc1d57666f5 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Sun, 1 Nov 2026 22:18:02 +0200 Subject: [PATCH 1/2] Add comprehensive tests for SystemGate inventory limits, address scopes, and package cache - Test address scope mapping (_safe_addr) for loopback, tailscale, private-lan, public-or-host - Test wildcard address scope depends on container detection - Test empty address handling (no IP or port) - Test services endpoint grouping, deduplication, sorting, and limit clamping - Test services endpoint gracefully handles AccessDenied errors - Test processes endpoint sorting by CPU/memory, skipping AccessDenied, and limit clamping - Test packages endpoint caching behavior with proper expiration timing All tests use mocking to avoid dependencies on real system state and cover edge cases that were not previously tested by the existing test suite. --- tests/test_inventory_limits.py | 345 +++++++++++++++++++++++++++++++++ 1 file changed, 345 insertions(+) create mode 100644 tests/test_inventory_limits.py diff --git a/tests/test_inventory_limits.py b/tests/test_inventory_limits.py new file mode 100644 index 0000000..3ee0240 --- /dev/null +++ b/tests/test_inventory_limits.py @@ -0,0 +1,345 @@ +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from fastapi.testclient import TestClient + + +def make_client(tmp_path, monkeypatch): + monkeypatch.setenv("SYSTEMGATE_ADMIN_KEY", "system-test-key-1234") + monkeypatch.setenv("SYSTEMGATE_DATA_DIR", str(tmp_path / "data")) + monkeypatch.setenv("SYSTEMGATE_BACKUP_ROOT", str(tmp_path / "backups")) + from systemgate.main import app + return TestClient(app) + + +def headers(): + return {"X-SystemGate-Key": "system-test-key-1234"} + + +@pytest.mark.parametrize("ip,expected_scope", [ + ("127.0.0.1", "loopback"), + ("127.9.9.9", "loopback"), + ("::1", "loopback"), + ("100.64.0.5", "tailscale"), + ("10.1.2.3", "private-lan"), + ("192.168.1.9", "private-lan"), + ("172.16.0.1", "private-lan"), + ("172.31.255.1", "private-lan"), + ("172.15.0.1", "public-or-host"), + ("172.32.0.1", "public-or-host"), + ("8.8.8.8", "public-or-host"), + ("fe80::1", "public-or-host"), +]) +def test_address_scopes(ip, expected_scope): + import systemgate.main as main + addr = SimpleNamespace(ip=ip, port=8080) + result = main._safe_addr(addr) + assert result == f"{expected_scope}:8080" + + +@pytest.mark.parametrize("ip,docker,expected", [ + ("0.0.0.0", True, "container-internal:53"), + ("0.0.0.0", False, "all-interfaces:53"), + ("::", True, "container-internal:53"), + ("::", False, "all-interfaces:53"), +]) +def test_wildcard_scope_depends_on_a_container(ip, docker, expected, monkeypatch): + import systemgate.main as main + + # Mock the Path class constructor + def mock_path_constructor(path_str): + if path_str == "/.dockerenv": + return SimpleNamespace(exists=lambda: docker) + return main.Path(path_str) + + monkeypatch.setattr(main, "Path", mock_path_constructor) + addr = SimpleNamespace(ip=ip, port=53) + result = main._safe_addr(addr) + assert result == expected + + +def test_an_address_without_ip_or_port_is_empty(): + import systemgate.main as main + + # Test None address + assert main._safe_addr(None) == "" + + # Test empty IP + addr1 = SimpleNamespace(ip="", port=8080) + assert main._safe_addr(addr1) == "" + + # Test port 0 + addr2 = SimpleNamespace(ip="127.0.0.1", port=0) + assert main._safe_addr(addr2) == "" + + # Test None port + addr3 = SimpleNamespace(ip="127.0.0.1", port=None) + assert main._safe_addr(addr3) == "" + + +def test_services_group_dedupe_and_sort(tmp_path, monkeypatch): + import systemgate.main as main + + # Mock net_connections + connections = [ + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=9000), pid=1), + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=81), pid=2), + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=81), pid=3), + Mock(status="LISTEN", laddr=SimpleNamespace(ip="10.0.0.2", port=82), pid=3), + Mock(status="ESTABLISHED", laddr=SimpleNamespace(ip="127.0.0.1", port=7), pid=2), + Mock(status="LISTEN", laddr=None, pid=2), + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=0), pid=2), + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=90), pid=5), + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=91), pid=None), + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=92), pid=4), + ] + monkeypatch.setattr(main.psutil, "net_connections", lambda kind: connections) + + # Mock Process + def mock_process(pid): + proc = Mock() + if pid == 5: + raise main.psutil.NoSuchProcess(5) + name_map = {1: "zeta", 2: "alpha", 3: "alpha", 4: None} + proc.name.return_value = name_map.get(pid, "unknown") + return proc + + monkeypatch.setattr(main.psutil, "Process", mock_process) + + with make_client(tmp_path, monkeypatch) as client: + response = client.get("/services", headers=headers()) + data = response.json() + + assert data["metadata_only"] is True + results = data["results"] + assert len(results) == 3 + + # Check first service (alpha) + alpha = results[0] + assert alpha["id"] == "service-001" + assert alpha["name"] == "alpha" + assert alpha["status"] == "listening" + assert alpha["kind"] == "process-listener" + assert set(alpha["listeners"]) == {"loopback:81", "private-lan:82"} + + # Check second service (unknown) + unknown = results[1] + assert unknown["id"] == "service-002" + assert unknown["name"] == "unknown" + assert unknown["status"] == "listening" + assert unknown["kind"] == "process-listener" + assert set(unknown["listeners"]) == {"loopback:90", "loopback:91", "loopback:92"} + + # Check third service (zeta) + zeta = results[2] + assert zeta["id"] == "service-003" + assert zeta["name"] == "zeta" + assert zeta["status"] == "listening" + assert zeta["kind"] == "process-listener" + assert zeta["listeners"] == ["loopback:9000"] + + +def test_services_limit_is_clamped(tmp_path, monkeypatch): + import systemgate.main as main + + # Create 250 connections + connections = [] + for i in range(1, 251): + connections.append(Mock( + status="LISTEN", + laddr=SimpleNamespace(ip="127.0.0.1", port=8000 + i), + pid=i + )) + + monkeypatch.setattr(main.psutil, "net_connections", lambda kind: connections) + + def mock_process(pid): + proc = Mock() + proc.name.return_value = f"p{pid:03d}" + return proc + + monkeypatch.setattr(main.psutil, "Process", mock_process) + + with make_client(tmp_path, monkeypatch) as client: + # Test limit values + test_cases = [ + (0, 1), + (-5, 1), + (1, 1), + (5, 5), + (200, 200), + (1000, 200), + (None, 80), # default + ] + + for limit_param, expected_count in test_cases: + if limit_param is None: + response = client.get("/services", headers=headers()) + else: + response = client.get(f"/services?limit={limit_param}", headers=headers()) + + data = response.json() + assert len(data["results"]) == expected_count + + # Test specific limit=3 case + response = client.get("/services?limit=3", headers=headers()) + data = response.json() + assert len(data["results"]) == 3 + assert data["results"][0]["name"] == "p001" + assert data["results"][1]["name"] == "p002" + assert data["results"][2]["name"] == "p003" + assert data["results"][0]["id"] == "service-001" + assert data["results"][1]["id"] == "service-002" + assert data["results"][2]["id"] == "service-003" + + +def test_services_survive_a_denied_connection_listing(tmp_path, monkeypatch): + import systemgate.main as main + + monkeypatch.setattr(main.psutil, "net_connections", lambda kind: (_ for _ in ()).throw(main.psutil.AccessDenied())) + + with make_client(tmp_path, monkeypatch) as client: + response = client.get("/services", headers=headers()) + data = response.json() + + assert response.status_code == 200 + assert data["results"] == [] + assert data["metadata_only"] is True + + +def test_processes_sort_skip_and_clamp(tmp_path, monkeypatch): + import systemgate.main as main + + # Create mock process objects that behave like psutil process objects + class MockProcess: + def __init__(self, info_dict=None, raises_access_denied=False): + if raises_access_denied: + self.info = None # This will cause AccessDenied when accessed + else: + self.info = info_dict or {} + + @property + def info(self): + if hasattr(self, '_raises_access_denied') and self._raises_access_denied: + raise main.psutil.AccessDenied() + return self._info + + @info.setter + def info(self, value): + if value is None: + self._raises_access_denied = True + self._info = {} + else: + self._raises_access_denied = False + self._info = value + + processes = [ + MockProcess({"pid": 1, "name": "proc1", "username": "user", "cpu_percent": 1.0, "memory_percent": 5.0}), + MockProcess({"pid": 2, "name": "proc2", "username": "user", "cpu_percent": 9.0, "memory_percent": 1.0}), + MockProcess(raises_access_denied=True), # This will raise AccessDenied + MockProcess({"pid": 3, "name": "proc3", "username": "user", "cpu_percent": 1.0, "memory_percent": 7.0}), + MockProcess({"pid": 4, "name": "proc4", "username": "user", "cpu_percent": None, "memory_percent": None}), + ] + + def mock_process_iter(fields): + return processes + + monkeypatch.setattr(main.psutil, "process_iter", mock_process_iter) + + with make_client(tmp_path, monkeypatch) as client: + # Test default ordering and limits + response = client.get("/processes", headers=headers()) + data = response.json() + results = data["results"] + + # Check ordering: should be by CPU then memory, descending + assert results[0]["pid"] == 2 # CPU 9.0, mem 1.0 + assert results[1]["pid"] == 3 # CPU 1.0, mem 7.0 + assert results[2]["pid"] == 1 # CPU 1.0, mem 5.0 + assert results[3]["pid"] == 4 # CPU None, mem None + + # Test limit clamping + test_cases = [ + (0, 1), + (2, 2), + (1000, 4), + ] + + for limit_param, expected_count in test_cases: + response = client.get(f"/processes?limit={limit_param}", headers=headers()) + data = response.json() + assert len(data["results"]) == expected_count + + +def test_processes_sort_skip_and_clamp_with_150_processes(tmp_path, monkeypatch): + import systemgate.main as main + + # Create 150 processes with varying CPU/memory + processes = [] + for i in range(150): + processes.append(Mock(info={ + "pid": i + 1, + "name": f"proc{i+1}", + "username": "user", + "cpu_percent": float(i % 10), # CPU 0-9 + "memory_percent": float(i % 5), # Memory 0-4 + })) + + monkeypatch.setattr(main.psutil, "process_iter", lambda fields: processes) + + with make_client(tmp_path, monkeypatch) as client: + response = client.get("/processes?limit=1000", headers=headers()) + data = response.json() + + # Should be clamped to 100 + assert len(data["results"]) == 100 + + +def test_packages_are_cached_for_an_hour(tmp_path, monkeypatch): + import systemgate.main as main + + # Track command calls + commands_run = [] + + def mock_run_summary(command): + commands_run.append(command[0]) # Track first element of command + return {"ok": True, "output": "[]"} + + monkeypatch.setattr(main, "_run_summary", mock_run_summary) + + # Mock time + clock = [1000.0] + + def mock_time(): + return clock[0] + + monkeypatch.setattr(main.time, "time", mock_time) + + with make_client(tmp_path, monkeypatch) as client: + # First call - should run commands + response = client.get("/packages", headers=headers()) + data = response.json() + + assert len(commands_run) == 3 + assert commands_run == ["python", "npm", "sh"] + assert data["generated_at"] == 1000.0 + + # Second call at 3599 seconds - should use cache + clock[0] = 1000.0 + 3599 + response = client.get("/packages", headers=headers()) + data = response.json() + + assert len(commands_run) == 3 # No new commands + assert data["generated_at"] == 1000.0 # Same timestamp + + # Third call at 3600 seconds - should recompute + clock[0] = 1000.0 + 3600 + response = client.get("/packages", headers=headers()) + data = response.json() + + assert len(commands_run) == 6 # 3 new commands + assert commands_run == ["python", "npm", "sh", "python", "npm", "sh"] + assert data["generated_at"] == 4600.0 From 8cde1d1ca84de7018cb1f6fb3351ce13f372cb59 Mon Sep 17 00:00:00 2001 From: alexeybe1kin <210597588+alexeybe1kin@users.noreply.github.com> Date: Sun, 1 Nov 2026 22:34:22 +0200 Subject: [PATCH 2/2] Tighten inventory tests: ordered listeners, pid order per limit, ruff format Listener lists were compared as sets although the order is part of the behaviour; the limit cases only counted rows; the docker patch could recurse into itself; the denied-process stand-in was convoluted; the file was not ruff-formatted. Co-Authored-By: Claude Opus 5.5 --- tests/test_inventory_limits.py | 169 ++++++++++++++++++--------------- 1 file changed, 92 insertions(+), 77 deletions(-) diff --git a/tests/test_inventory_limits.py b/tests/test_inventory_limits.py index 3ee0240..b9781ee 100644 --- a/tests/test_inventory_limits.py +++ b/tests/test_inventory_limits.py @@ -12,6 +12,7 @@ def make_client(tmp_path, monkeypatch): monkeypatch.setenv("SYSTEMGATE_DATA_DIR", str(tmp_path / "data")) monkeypatch.setenv("SYSTEMGATE_BACKUP_ROOT", str(tmp_path / "backups")) from systemgate.main import app + return TestClient(app) @@ -19,43 +20,44 @@ def headers(): return {"X-SystemGate-Key": "system-test-key-1234"} -@pytest.mark.parametrize("ip,expected_scope", [ - ("127.0.0.1", "loopback"), - ("127.9.9.9", "loopback"), - ("::1", "loopback"), - ("100.64.0.5", "tailscale"), - ("10.1.2.3", "private-lan"), - ("192.168.1.9", "private-lan"), - ("172.16.0.1", "private-lan"), - ("172.31.255.1", "private-lan"), - ("172.15.0.1", "public-or-host"), - ("172.32.0.1", "public-or-host"), - ("8.8.8.8", "public-or-host"), - ("fe80::1", "public-or-host"), -]) +@pytest.mark.parametrize( + "ip,expected_scope", + [ + ("127.0.0.1", "loopback"), + ("127.9.9.9", "loopback"), + ("::1", "loopback"), + ("100.64.0.5", "tailscale"), + ("10.1.2.3", "private-lan"), + ("192.168.1.9", "private-lan"), + ("172.16.0.1", "private-lan"), + ("172.31.255.1", "private-lan"), + ("172.15.0.1", "public-or-host"), + ("172.32.0.1", "public-or-host"), + ("8.8.8.8", "public-or-host"), + ("fe80::1", "public-or-host"), + ], +) def test_address_scopes(ip, expected_scope): import systemgate.main as main + addr = SimpleNamespace(ip=ip, port=8080) result = main._safe_addr(addr) assert result == f"{expected_scope}:8080" -@pytest.mark.parametrize("ip,docker,expected", [ - ("0.0.0.0", True, "container-internal:53"), - ("0.0.0.0", False, "all-interfaces:53"), - ("::", True, "container-internal:53"), - ("::", False, "all-interfaces:53"), -]) +@pytest.mark.parametrize( + "ip,docker,expected", + [ + ("0.0.0.0", True, "container-internal:53"), + ("0.0.0.0", False, "all-interfaces:53"), + ("::", True, "container-internal:53"), + ("::", False, "all-interfaces:53"), + ], +) def test_wildcard_scope_depends_on_a_container(ip, docker, expected, monkeypatch): import systemgate.main as main - # Mock the Path class constructor - def mock_path_constructor(path_str): - if path_str == "/.dockerenv": - return SimpleNamespace(exists=lambda: docker) - return main.Path(path_str) - - monkeypatch.setattr(main, "Path", mock_path_constructor) + monkeypatch.setattr(main, "Path", lambda _p: SimpleNamespace(exists=lambda: docker)) addr = SimpleNamespace(ip=ip, port=53) result = main._safe_addr(addr) assert result == expected @@ -123,7 +125,7 @@ def mock_process(pid): assert alpha["name"] == "alpha" assert alpha["status"] == "listening" assert alpha["kind"] == "process-listener" - assert set(alpha["listeners"]) == {"loopback:81", "private-lan:82"} + assert alpha["listeners"] == ["loopback:81", "private-lan:82"] # Check second service (unknown) unknown = results[1] @@ -131,7 +133,7 @@ def mock_process(pid): assert unknown["name"] == "unknown" assert unknown["status"] == "listening" assert unknown["kind"] == "process-listener" - assert set(unknown["listeners"]) == {"loopback:90", "loopback:91", "loopback:92"} + assert unknown["listeners"] == ["loopback:90", "loopback:91", "loopback:92"] # Check third service (zeta) zeta = results[2] @@ -148,11 +150,9 @@ def test_services_limit_is_clamped(tmp_path, monkeypatch): # Create 250 connections connections = [] for i in range(1, 251): - connections.append(Mock( - status="LISTEN", - laddr=SimpleNamespace(ip="127.0.0.1", port=8000 + i), - pid=i - )) + connections.append( + Mock(status="LISTEN", laddr=SimpleNamespace(ip="127.0.0.1", port=8000 + i), pid=i) + ) monkeypatch.setattr(main.psutil, "net_connections", lambda kind: connections) @@ -199,7 +199,11 @@ def mock_process(pid): def test_services_survive_a_denied_connection_listing(tmp_path, monkeypatch): import systemgate.main as main - monkeypatch.setattr(main.psutil, "net_connections", lambda kind: (_ for _ in ()).throw(main.psutil.AccessDenied())) + monkeypatch.setattr( + main.psutil, + "net_connections", + lambda kind: (_ for _ in ()).throw(main.psutil.AccessDenied()), + ) with make_client(tmp_path, monkeypatch) as client: response = client.get("/services", headers=headers()) @@ -213,35 +217,49 @@ def test_services_survive_a_denied_connection_listing(tmp_path, monkeypatch): def test_processes_sort_skip_and_clamp(tmp_path, monkeypatch): import systemgate.main as main - # Create mock process objects that behave like psutil process objects - class MockProcess: - def __init__(self, info_dict=None, raises_access_denied=False): - if raises_access_denied: - self.info = None # This will cause AccessDenied when accessed - else: - self.info = info_dict or {} - + class Denied: @property def info(self): - if hasattr(self, '_raises_access_denied') and self._raises_access_denied: - raise main.psutil.AccessDenied() - return self._info - - @info.setter - def info(self, value): - if value is None: - self._raises_access_denied = True - self._info = {} - else: - self._raises_access_denied = False - self._info = value + raise main.psutil.AccessDenied() processes = [ - MockProcess({"pid": 1, "name": "proc1", "username": "user", "cpu_percent": 1.0, "memory_percent": 5.0}), - MockProcess({"pid": 2, "name": "proc2", "username": "user", "cpu_percent": 9.0, "memory_percent": 1.0}), - MockProcess(raises_access_denied=True), # This will raise AccessDenied - MockProcess({"pid": 3, "name": "proc3", "username": "user", "cpu_percent": 1.0, "memory_percent": 7.0}), - MockProcess({"pid": 4, "name": "proc4", "username": "user", "cpu_percent": None, "memory_percent": None}), + Mock( + info={ + "pid": 1, + "name": "proc1", + "username": "user", + "cpu_percent": 1.0, + "memory_percent": 5.0, + } + ), + Mock( + info={ + "pid": 2, + "name": "proc2", + "username": "user", + "cpu_percent": 9.0, + "memory_percent": 1.0, + } + ), + Denied(), + Mock( + info={ + "pid": 3, + "name": "proc3", + "username": "user", + "cpu_percent": 1.0, + "memory_percent": 7.0, + } + ), + Mock( + info={ + "pid": 4, + "name": "proc4", + "username": "user", + "cpu_percent": None, + "memory_percent": None, + } + ), ] def mock_process_iter(fields): @@ -262,31 +280,28 @@ def mock_process_iter(fields): assert results[3]["pid"] == 4 # CPU None, mem None # Test limit clamping - test_cases = [ - (0, 1), - (2, 2), - (1000, 4), - ] - - for limit_param, expected_count in test_cases: + for limit_param, expected_pids in [(0, [2]), (2, [2, 3]), (1000, [2, 3, 1, 4])]: response = client.get(f"/processes?limit={limit_param}", headers=headers()) - data = response.json() - assert len(data["results"]) == expected_count + assert [row["pid"] for row in response.json()["results"]] == expected_pids -def test_processes_sort_skip_and_clamp_with_150_processes(tmp_path, monkeypatch): +def test_processes_limit_is_capped_at_100(tmp_path, monkeypatch): import systemgate.main as main # Create 150 processes with varying CPU/memory processes = [] for i in range(150): - processes.append(Mock(info={ - "pid": i + 1, - "name": f"proc{i+1}", - "username": "user", - "cpu_percent": float(i % 10), # CPU 0-9 - "memory_percent": float(i % 5), # Memory 0-4 - })) + processes.append( + Mock( + info={ + "pid": i + 1, + "name": f"proc{i + 1}", + "username": "user", + "cpu_percent": float(i % 10), # CPU 0-9 + "memory_percent": float(i % 5), # Memory 0-4 + } + ) + ) monkeypatch.setattr(main.psutil, "process_iter", lambda fields: processes)