feat: implement AI Workspace service manager with lifecycle control for local services

scripts/aiw/test_services.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import socket
+import sys
+import tempfile
+import unittest
+import warnings
+from pathlib import Path
+from contextlib import redirect_stdout
+from unittest.mock import patch
+
+
+SERVICES_PATH = Path(__file__).with_name("services.py")
+SPEC = importlib.util.spec_from_file_location("aiw_services", SERVICES_PATH)
+services = importlib.util.module_from_spec(SPEC)
+assert SPEC.loader is not None
+sys.modules[SPEC.name] = services
+SPEC.loader.exec_module(services)
+
+
+def sample_manifest() -> dict:
+    return {
+        "services": {
+            "alpha": {
+                "enabled": True,
+                "kind": "process",
+                "command": ["python3", "-c", "import time; time.sleep(60)"],
+                "groups": ["core"],
+            },
+            "beta": {
+                "enabled": True,
+                "kind": "process",
+                "command": ["python3", "-c", "import time; time.sleep(60)"],
+                "groups": ["capture"],
+                "depends_on": ["alpha"],
+            },
+            "disabled": {
+                "enabled": False,
+                "kind": "process",
+                "command": ["python3", "-c", "import time; time.sleep(60)"],
+                "groups": ["core"],
+            },
+        }
+    }
+
+
+class ServiceManagerTests(unittest.TestCase):
+    def test_select_services_excludes_disabled_by_default(self) -> None:
+        selected = services.select_services(sample_manifest(), names=[], group=None)
+
+        self.assertEqual([item.name for item in selected], ["alpha", "beta"])
+
+    def test_select_services_can_include_disabled_for_status(self) -> None:
+        selected = services.select_services(sample_manifest(), names=[], group=None, include_disabled=True)
+
+        self.assertEqual([item.name for item in selected], ["alpha", "beta", "disabled"])
+
+    def test_select_services_filters_by_group(self) -> None:
+        selected = services.select_services(sample_manifest(), names=[], group="capture")
+
+        self.assertEqual([item.name for item in selected], ["beta"])
+
+    def test_health_ok_tcp_reports_open_port(self) -> None:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
+            server.bind(("127.0.0.1", 0))
+            server.listen(1)
+            port = server.getsockname()[1]
+
+            ok, detail = services.health_ok({"health": {"type": "tcp", "host": "127.0.0.1", "port": port}})
+
+        self.assertTrue(ok)
+        self.assertIn(f"127.0.0.1:{port}", detail)
+
+    def test_read_pid_ignores_invalid_pid_file(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            pid_dir = Path(tmp) / "pids"
+            target = pid_dir / "fidelity"
+            target.mkdir(parents=True)
+            (target / "alpha.pid").write_text("not-a-pid\n", encoding="utf-8")
+
+            with patch.object(services, "PID_DIR", pid_dir):
+                self.assertIsNone(services.read_pid("fidelity", "alpha"))
+
+    def test_start_and_stop_managed_process(self) -> None:
+        manifest = {
+            "services": {
+                "sleeper": {
+                    "enabled": True,
+                    "kind": "process",
+                    "command": [sys.executable, "-c", "import time; time.sleep(60)"],
+                    "restart": "never",
+                }
+            }
+        }
+        ref = services.ServiceRef("sleeper", manifest["services"]["sleeper"])
+
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            with patch.object(services, "PID_DIR", root / "pids"), \
+                patch.object(services, "LOG_DIR", root / "logs"), \
+                patch.object(services, "STATE_DIR", root / "state"):
+                started: set[str] = set()
+                services.ensure_runtime()
+                with warnings.catch_warnings():
+                    warnings.simplefilter("ignore", ResourceWarning)
+                    with redirect_stdout(io.StringIO()):
+                        services.start_service("test", ref, manifest, started)
+
+                pid = services.read_pid("test", "sleeper")
+                self.assertTrue(services.is_running(pid))
+                self.assertIn("sleeper", started)
+
+                with redirect_stdout(io.StringIO()):
+                    services.stop_service("test", ref)
+                self.assertFalse(services.is_running(pid))
+
+
+if __name__ == "__main__":
+    unittest.main()