"""Host worker tests: python3 -m unittest discover -s deploy -p 'test_*.py'.""" import importlib.util import json import subprocess import tempfile import unittest import shutil import sys from pathlib import Path from unittest.mock import patch spec = importlib.util.spec_from_file_location("server_agent", Path(__file__).with_name("server-agent.py")) agent_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(agent_module) Agent, ManagementError = agent_module.Agent, agent_module.ManagementError class AgentTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.deploy, self.mailbox, self.private = [self.root / n for n in ("deploy", "mailbox", "private")] for directory in (self.deploy, self.mailbox, self.private): directory.mkdir() self.schema = [ {"key": "LOG_LEVEL", "group": "api", "type": "string", "default_value": "info", "secret": False, "readonly": False}, {"key": "GITEA_TOKEN", "group": "api", "type": "string", "default_value": "", "secret": True, "readonly": False}, {"key": "TRAIN_HOUR", "group": "recommender", "type": "integer", "default_value": "3", "secret": False, "readonly": False}, {"key": "PIPU_DATA", "group": "deployment", "type": "string", "default_value": "/srv/pipu", "secret": False, "readonly": True}, ] (self.deploy / "config.schema.json").write_text(json.dumps(self.schema), encoding="utf-8") (self.deploy / "config.local").write_text("", encoding="utf-8") (self.deploy / ".env").write_text("LOG_LEVEL=warn\nGITEA_TOKEN=top-secret\n", encoding="utf-8") self.agent = Agent(self.deploy, self.mailbox, self.private) self.running_commit = "0" * 40 self.agent.installed_commit = lambda body: body.get("_running_commit", self.running_commit) self.agent.save("sources.json", [{"id": "origin", "name": "origin", "url": "https://example.org/o/pipu.git"}]) def test_unchanged_configuration_needs_no_restart_even_without_saved_baseline(self): self.assertFalse(self.agent.config()["restart_required"]) with patch.object(self.agent, "command", return_value=""): result = self.agent.save_config({"revision": self.agent.revision(), "values": {"LOG_LEVEL": "debug"}}) self.assertFalse(result["restart_required"], "LOG_LEVEL is overridden by .env") def test_returning_to_applied_values_removes_pending_restart(self): self.assertFalse(self.agent.config()["restart_required"]) with patch.object(self.agent, "command", return_value=""): changed = self.agent.save_config({"revision": self.agent.revision(), "values": {"TRAIN_HOUR": "6"}}) self.assertTrue(changed["restart_required"]) restored = self.agent.save_config({"revision": changed["revision"], "values": {"TRAIN_HOUR": None}}) self.assertFalse(restored["restart_required"]) def test_explicit_default_value_does_not_require_a_restart(self): self.assertFalse(self.agent.config()["restart_required"]) with patch.object(self.agent, "command", return_value=""): result = self.agent.save_config({"revision": self.agent.revision(), "values": {"TRAIN_HOUR": "3"}}) self.assertFalse(result["restart_required"]) def test_update_check_and_start_both_reject_installed_main_commit(self): refs = {"refs/heads/main": self.running_commit} body = {"source_id": "origin", "channel": "main"} with patch.object(self.agent, "refs", return_value=refs), patch.object(self.agent, "start_job") as start: self.assertEqual("current", self.agent.handle("updates.get", body)["update_state"]) with self.assertRaises(ManagementError) as error: self.agent.handle("updates.start", body) self.assertEqual("management.already_installed", error.exception.code) start.assert_not_called() def test_check_distinguishes_missing_selection_and_changed_target(self): with patch.object(self.agent, "refs", return_value={"refs/heads/main": "a" * 40}), patch.object(self.agent, "start_job") as start: self.assertEqual("select_version", self.agent.handle("updates.get", {"channel": "version"})["update_state"]) with self.assertRaises(ManagementError) as error: self.agent.handle("updates.start", {"channel": "main", "expected_commit": "b" * 40}) self.assertEqual("management.target_changed", error.exception.code) start.assert_not_called() def test_unknown_installed_commit_and_older_release_cannot_be_installed(self): with patch.object(self.agent, "refs", return_value={"refs/tags/v1.8.0": "a" * 40}), patch.object(self.agent, "start_job") as start: for body, state in [({"_running_commit": ""}, "unknown"), ({"_running_version": "1.9.0-1-gbc55528"}, "ahead")]: self.assertEqual(state, self.agent.handle("updates.get", body)["update_state"]) with self.assertRaises(ManagementError): self.agent.handle("updates.start", body) start.assert_not_called() def test_installed_identity_uses_running_build_and_resolves_legacy_describe(self): with patch.object(self.agent, "command", return_value="a" * 40) as command: self.assertEqual("b" * 40, Agent.installed_commit(self.agent, {"_running_commit": "b" * 40})) command.assert_not_called() self.assertEqual("a" * 40, Agent.installed_commit(self.agent, {"_running_version": "1.9.0-1-gbc55528"})) self.assertEqual("bc55528^{commit}", command.call_args.args[0][-1]) self.assertEqual("a" * 40, Agent.installed_commit(self.agent, {"_running_version": "main-bc55528"})) self.assertEqual("", Agent.installed_commit(self.agent, {"_running_version": "0.0.0-dev"})) def test_env_priority_and_secrets_never_returned(self): (self.deploy / "config.local").write_text("LOG_LEVEL=debug\nTRAIN_HOUR=5\n", encoding="utf-8") config = self.agent.config() entries = {v["key"]: v for v in config["entries"]} self.assertEqual(("warn", "env", "debug"), tuple(entries["LOG_LEVEL"][k] for k in ("effective", "source", "local"))) self.assertEqual(("5", "local"), tuple(entries["TRAIN_HOUR"][k] for k in ("effective", "source"))) self.assertNotIn("top-secret", json.dumps(config)) def test_save_roundtrip_reset_and_conflict(self): with patch.object(self.agent, "command", return_value=""): result = self.agent.save_config({"revision": self.agent.revision(), "values": {"TRAIN_HOUR": "7", "GITEA_TOKEN": "a$'\\b#c"}}) self.assertEqual("a$'\\b#c", agent_module.read_env(self.deploy / "config.local")["GITEA_TOKEN"]) self.assertTrue(result["restart_required"]) self.assertNotIn("a$", json.dumps(result)) self.assertNotIn("a$", (self.private / "audit.jsonl").read_text()) with self.assertRaises(ManagementError) as raised: self.agent.save_config({"revision": "stale", "values": {"TRAIN_HOUR": "8"}}) self.assertEqual(409, raised.exception.status) self.agent.save_config({"revision": result["revision"], "values": {"TRAIN_HOUR": None}}) self.assertNotIn("TRAIN_HOUR", agent_module.read_env(self.deploy / "config.local")) def test_invalid_values_and_compose_failure_preserve_original(self): original = (self.deploy / "config.local").read_bytes() for values in ({"TRAIN_HOUR": "24"}, {"TRAIN_HOUR": "bad"}, {"UNKNOWN": "x"}, {"PIPU_DATA": "/tmp"}, {"LOG_LEVEL": "x\nOTHER=y"}): with self.assertRaises(ManagementError): self.agent.save_config({"revision": self.agent.revision(), "values": values}) self.assertEqual(original, (self.deploy / "config.local").read_bytes()) with patch.object(self.agent, "command", side_effect=subprocess.CalledProcessError(1, "docker")): with self.assertRaises(ManagementError): self.agent.save_config({"revision": self.agent.revision(), "values": {"TRAIN_HOUR": "6"}}) self.assertEqual(original, (self.deploy / "config.local").read_bytes()) def test_source_validation_rejects_shell_local_and_credentials(self): for url in ("/tmp/repo", "file:///tmp/repo", "ext::sh -c evil", "--upload-pack=evil", "https://user:secret@example.org/o/r", "https://secret@example.org/o/r", "https://example.org/o/r\narg", "ssh://root@example.org/o/r"): with self.subTest(url=url), self.assertRaises(ManagementError): agent_module.validate_url(url) for url in ("https://example.org/o/r", "http://10.0.0.1:3000/o/r", "git@example.org:o/r.git", "ssh://git@example.org:2222/o/r.git"): agent_module.validate_url(url) def test_latest_is_semver_stable_and_pins_peeled_commit(self): refs = {"refs/tags/v1.9.0": "a" * 40, "refs/tags/v1.10.0": "b" * 40, "refs/tags/v1.10.0^{}": "c" * 40, "refs/tags/v2.0.0-rc.1": "d" * 40} with patch.object(self.agent, "refs", return_value=refs), patch.object(self.agent, "start_job", return_value={}) as start: self.agent.handle("updates.start", {"source_id": "origin", "channel": "latest"}) self.assertEqual("v1.10.0", start.call_args.args[1]) self.assertEqual("c" * 40, start.call_args.args[3]["PIPU_UPDATE_COMMIT"]) self.assertEqual("refs/tags/v1.10.0", start.call_args.args[3]["PIPU_UPDATE_REF"]) with self.assertRaises(ManagementError): self.agent.handle("updates.start", {"source_id": "origin", "channel": "version", "version": "--evil"}) def test_main_and_default_source(self): with patch.object(self.agent, "refs", return_value={"refs/heads/main": "a" * 40}), patch.object(self.agent, "start_job", return_value={}) as start: self.agent.handle("updates.start", {"source_id": "origin", "channel": "main"}) self.assertEqual("main", start.call_args.args[1]) result = self.agent.handle("updates.get", {"source_id": None}) self.assertEqual("origin", result["sources"][0]["id"]) def test_running_job_blocks_mutations_and_survives_restart_as_failure(self): self.agent.job = {"state": "running"} for operation, body in (("updates.start", {"channel": "main"}), ("config.apply", {}), ("sources.save", {}), ("config.save", {})): with self.subTest(operation=operation), self.assertRaises(ManagementError) as raised: self.agent.handle(operation, body) self.assertEqual(409, raised.exception.status) self.agent.save("job.json", {"state": "running", "id": "old"}) restored = Agent(self.deploy, self.mailbox, self.private) self.assertEqual("failed", restored.job["state"]) def test_completed_job_redacts_secrets_and_marks_applied_revision(self): job = {"id": "test", "kind": "config", "state": "running", "started_at": "", "finished_at": "", "target": "config.local", "log": ""} self.agent.job = job revision = self.agent.revision() self.agent.run_job(job, [sys.executable, "-c", "print('top-secret')"], {}, revision) self.assertEqual("succeeded", job["state"]) self.assertNotIn("top-secret", job["log"]) self.assertIn("[redacted]", job["log"]) self.assertFalse(self.agent.config()["restart_required"]) self.agent.run_job(job, [sys.executable, "-c", "raise SystemExit(2)"], {}, revision) self.assertEqual("failed", job["state"]) def test_mailbox_discards_stale_request_and_never_executes_it(self): path = self.mailbox / "00000000-0000-0000-0000-000000000001.request.json" path.write_text(json.dumps({"operation": "config.apply", "created_at": 1}), encoding="utf-8") with patch.object(self.agent, "handle") as handle: self.agent.tick() handle.assert_not_called() response = json.loads(path.with_name(path.name.replace("request", "response")).read_text(encoding="utf-8")) self.assertEqual("management.request_expired", response["code"]) class ComposeTests(unittest.TestCase): @unittest.skipUnless(shutil.which('docker'), 'Docker Compose CLI unavailable') def test_real_compose_priorities_and_literal_secret_roundtrip(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / '.env').write_text('POSTGRES_PASSWORD=test-db\nINTERNAL_TOKEN=test-internal\nMEILI_MASTER_KEY=test-meili\nLOG_LEVEL=warn\n', encoding='utf-8') secret = "literal$VALUE'with\\slash#hash" (root / 'config.local').write_text("LOG_LEVEL='debug'\nTRAIN_HOUR='7'\nGITEA_TOKEN='" + secret.replace("'", "\\'") + "'\n", encoding='utf-8') compose = Path(__file__).with_name('compose.yaml').resolve() result = subprocess.run(['docker', 'compose', '--project-directory', str(root), '-f', str(compose), '--env-file', str(root / 'config.local'), '--env-file', str(root / '.env'), 'config', '--format', 'json'], capture_output=True, text=True, timeout=30) self.assertEqual(0, result.returncode, 'Docker Compose configuration failed: ' + result.stderr) services = json.loads(result.stdout)['services'] self.assertEqual('warn', services['api']['environment']['LOG_LEVEL']) self.assertEqual('7', services['recommender']['environment']['TRAIN_HOUR']) self.assertEqual('7', services['analyzer']['environment']['TRAIN_HOUR']) # Canonical Compose output escapes dollars for reuse as YAML. self.assertEqual(secret.replace('$', '$$'), services['api']['environment']['GITEA_TOKEN']) self.assertNotIn('GITEA_TOKEN', services['recommender']['environment']) if __name__ == "__main__": unittest.main()