diff --git a/tests/test_inventory_limits.py b/tests/test_inventory_limits.py new file mode 100644 index 0000000..b9781ee --- /dev/null +++ b/tests/test_inventory_limits.py @@ -0,0 +1,360 @@ +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 + + monkeypatch.setattr(main, "Path", lambda _p: SimpleNamespace(exists=lambda: docker)) + 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 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 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 + + class Denied: + @property + def info(self): + raise main.psutil.AccessDenied() + + processes = [ + 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): + 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 + 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()) + assert [row["pid"] for row in response.json()["results"]] == expected_pids + + +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 + } + ) + ) + + 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