"""Tests for the worker registry and configuration schema (#798, epic #797).""" import json import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from webui.worker_registry import ( ALLOWED_ROLES, SCHEMA_VERSION, RegistryValidationError, WorkerRegistry, default_registry_path, find_provider, find_worker, history_dir, list_revisions, load_registry, registry_to_dict, registry_to_document, rollback_to_revision, save_registry, validate_payload, worker_to_dict, workers_for_provider, ) _EXPECTED_PROVIDER_IDS = ("claude", "grok", "codex", "agy", "kimi-k") def _provider(provider_id: str = "claude", **overrides) -> dict: payload = { "id": provider_id, "display_name": "Claude", "vendor": "Anthropic", "executable": "claude", "available": True, "models": ["claude-opus-4-8"], "notes": "", } payload.update(overrides) return payload def _worker(worker_id: str = "claude-author", **overrides) -> dict: payload = { "id": worker_id, "display_name": "Claude author", "provider": "claude", "model": "claude-opus-4-8", "project": "gitea-tools", "role": "author", "namespace": "gitea-author", "profile": "prgs-author", "workflow": "skills/llm-project-workflow/workflows/work-issue.md", "schedule": {"kind": "cron", "expression": "0 * * * *"}, "timeout_seconds": 3600, "enabled": True, "scheduler": {"kind": "launchd", "label": "cc.prgs.claude.author"}, "notes": "", } payload.update(overrides) return payload def _document(providers=None, workers=None, **overrides) -> dict: payload = { "version": SCHEMA_VERSION, "revision": 1, "updated_at": "2026-07-22T00:00:00Z", "providers": providers if providers is not None else [_provider()], "workers": workers if workers is not None else [_worker()], } payload.update(overrides) return payload class _TempRegistryCase(unittest.TestCase): """Base case giving each test an isolated registry file.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.addCleanup(self._tmp.cleanup) self.path = Path(self._tmp.name) / "workers.registry.json" def write(self, document: dict) -> Path: self.path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8") return self.path def parse(self, document: dict) -> WorkerRegistry: return validate_payload(document, source_path=self.path) class TestPackagedRegistry(unittest.TestCase): """AC: the declarative registry is the source of truth and ships with the app.""" def test_default_path_points_at_packaged_data(self): path = default_registry_path() self.assertEqual(path.name, "workers.registry.json") self.assertEqual(path.parent.name, "data") def test_packaged_registry_loads_and_validates(self): registry = load_registry() self.assertEqual(registry.version, SCHEMA_VERSION) self.assertGreaterEqual(registry.revision, 1) def test_packaged_registry_declares_all_five_providers(self): registry = load_registry() self.assertEqual( tuple(provider.id for provider in registry.providers), _EXPECTED_PROVIDER_IDS, ) def test_packaged_registry_carries_no_credentials(self): raw = default_registry_path().read_text(encoding="utf-8").lower() for marker in ("token", "password", "secret", "api_key", "credential"): self.assertNotIn(marker, raw) class TestSeparateEntities(_TempRegistryCase): """AC: providers and configured workers are separate entities.""" def test_provider_may_exist_with_no_workers(self): registry = self.parse( _document(providers=[_provider("grok", display_name="Grok")], workers=[]) ) self.assertEqual(len(registry.providers), 1) self.assertEqual(registry.workers, ()) self.assertEqual(workers_for_provider(registry, "grok"), ()) def test_many_workers_may_share_one_provider(self): registry = self.parse( _document( workers=[ _worker("claude-author"), _worker( "claude-reviewer", role="reviewer", namespace="gitea-reviewer", profile="prgs-reviewer", scheduler={"kind": "launchd", "label": "cc.prgs.claude.reviewer"}, ), ] ) ) self.assertEqual(len(workers_for_provider(registry, "claude")), 2) self.assertEqual(len(registry.providers), 1) def test_worker_referencing_unknown_provider_is_refused(self): with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=[_worker(provider="mystery")])) self.assertIn("unknown provider", str(ctx.exception)) def test_lookup_helpers(self): registry = self.parse(_document()) self.assertIsNotNone(find_worker(registry, "claude-author")) self.assertIsNone(find_worker(registry, "absent")) self.assertIsNotNone(find_provider(registry, "claude")) self.assertIsNone(find_provider(registry, "absent")) class TestRecordedFields(_TempRegistryCase): """AC: records provider, model, project, role, namespace/profile, workflow, schedule, timeout, enabled state, and scheduler metadata.""" def test_every_required_field_is_recorded(self): registry = self.parse(_document()) worker = registry.workers[0] self.assertEqual(worker.provider, "claude") self.assertEqual(worker.model, "claude-opus-4-8") self.assertEqual(worker.project, "gitea-tools") self.assertEqual(worker.role, "author") self.assertEqual(worker.namespace, "gitea-author") self.assertEqual(worker.profile, "prgs-author") self.assertEqual(worker.workflow, "skills/llm-project-workflow/workflows/work-issue.md") self.assertEqual(worker.schedule.kind, "cron") self.assertEqual(worker.schedule.expression, "0 * * * *") self.assertEqual(worker.timeout_seconds, 3600) self.assertTrue(worker.enabled) self.assertEqual(worker.scheduler.kind, "launchd") self.assertEqual(worker.scheduler.label, "cc.prgs.claude.author") def test_each_required_field_is_individually_required(self): for field in ( "provider", "model", "project", "role", "namespace", "profile", "workflow", "schedule", "timeout_seconds", "enabled", "scheduler", "id", "display_name", ): with self.subTest(field=field): worker = _worker() worker.pop(field) with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[worker])) def test_all_sanctioned_roles_are_accepted(self): for role in ALLOWED_ROLES: with self.subTest(role=role): registry = self.parse(_document(workers=[_worker(role=role)])) self.assertEqual(registry.workers[0].role, role) def test_unsanctioned_role_is_refused(self): with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=[_worker(role="admin")])) self.assertIn("role must be one of", str(ctx.exception)) def test_worker_dict_round_trips_every_field(self): registry = self.parse(_document()) encoded = worker_to_dict(registry.workers[0]) self.assertEqual(encoded, _worker()) json.dumps(encoded) # must stay JSON-safe for the #799 API class TestSchemaValidation(_TempRegistryCase): """AC: supports schema validation — and fails closed.""" def test_unsupported_version_is_refused(self): with self.assertRaises(RegistryValidationError): self.parse(_document(version=2)) def test_root_must_be_an_object(self): with self.assertRaises(RegistryValidationError): validate_payload([], source_path=self.path) def test_providers_must_be_non_empty(self): with self.assertRaises(RegistryValidationError): self.parse(_document(providers=[])) def test_unknown_top_level_field_is_refused(self): with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(fleet=[])) self.assertIn("unknown fields", str(ctx.exception)) def test_unknown_worker_field_is_refused_not_ignored(self): # A typo'd field must not be silently dropped: "timeout_second" would # otherwise read as "no timeout declared". worker = _worker() worker["timeout_second"] = 30 with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=[worker])) self.assertIn("timeout_second", str(ctx.exception)) def test_credentials_are_refused_anywhere_in_the_document(self): for label, mutate in ( ("provider.api_token", lambda doc: doc["providers"][0].__setitem__("api_token", "x")), ("worker.password", lambda doc: doc["workers"][0].__setitem__("password", "x")), ("root.secret", lambda doc: doc.__setitem__("secret", "x")), ): with self.subTest(field=label): document = _document() mutate(document) with self.assertRaises(ValueError) as ctx: self.parse(document) self.assertIn("credential", str(ctx.exception).lower()) def test_duplicate_worker_id_is_refused(self): workers = [_worker("dup"), _worker("dup", scheduler={"kind": "manual"})] with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=workers)) self.assertIn("duplicate worker id", str(ctx.exception)) def test_duplicate_provider_id_is_refused(self): with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(providers=[_provider("claude"), _provider("claude")], workers=[])) self.assertIn("duplicate provider id", str(ctx.exception)) def test_duplicate_launchagent_label_is_refused(self): # Two workers sharing a label would silently overwrite each other's agent. workers = [ _worker("a", scheduler={"kind": "launchd", "label": "cc.prgs.same"}), _worker("b", scheduler={"kind": "launchd", "label": "cc.prgs.same"}), ] with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=workers)) self.assertIn("duplicate scheduler label", str(ctx.exception)) def test_manual_scheduler_needs_no_label_and_many_may_coexist(self): workers = [ _worker("a", scheduler={"kind": "manual"}), _worker("b", scheduler={"kind": "manual"}), ] registry = self.parse(_document(workers=workers)) self.assertEqual([w.scheduler.label for w in registry.workers], [None, None]) def test_launchd_scheduler_requires_a_label(self): with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=[_worker(scheduler={"kind": "launchd"})])) self.assertIn("label is required", str(ctx.exception)) def test_unknown_scheduler_kind_is_refused(self): with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[_worker(scheduler={"kind": "systemd", "label": "x"})])) def test_timeout_must_be_a_positive_bounded_integer(self): for bad in (0, -1, "3600", 1.5, True, 86_401): with self.subTest(timeout=bad): with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[_worker(timeout_seconds=bad)])) def test_enabled_must_be_a_real_boolean(self): for bad in ("true", 1, None): with self.subTest(enabled=bad): with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[_worker(enabled=bad)])) def test_identifier_shape_is_enforced(self): for bad in ("Claude Author", "-leading", "UPPER", ""): with self.subTest(worker_id=bad): with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[_worker(bad)])) class TestScheduleValidation(_TempRegistryCase): """Schedules are declarations; next-run computation belongs to #803.""" def test_interval_schedule_requires_positive_seconds(self): registry = self.parse( _document(workers=[_worker(schedule={"kind": "interval", "seconds": 900})]) ) self.assertEqual(registry.workers[0].schedule.seconds, 900) with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[_worker(schedule={"kind": "interval"})])) with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[_worker(schedule={"kind": "interval", "seconds": 0})])) def test_cron_schedule_requires_five_fields(self): with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=[_worker(schedule={"kind": "cron", "expression": "0 *"})])) self.assertIn("five crontab fields", str(ctx.exception)) def test_manual_schedule_needs_no_timing(self): registry = self.parse(_document(workers=[_worker(schedule={"kind": "manual"})])) schedule = registry.workers[0].schedule self.assertEqual(schedule.kind, "manual") self.assertIsNone(schedule.seconds) self.assertIsNone(schedule.expression) def test_fields_from_the_wrong_kind_are_refused(self): with self.assertRaises(RegistryValidationError) as ctx: self.parse(_document(workers=[_worker(schedule={"kind": "manual", "seconds": 60})])) self.assertIn("not valid for kind", str(ctx.exception)) def test_unknown_schedule_kind_is_refused(self): with self.assertRaises(RegistryValidationError): self.parse(_document(workers=[_worker(schedule={"kind": "hourly"})])) class TestAtomicPersistence(_TempRegistryCase): """AC: atomic persistence.""" def test_save_then_load_round_trips(self): registry = self.parse(_document()) save_registry(registry, self.path) reloaded = load_registry(self.path) self.assertEqual( [worker_to_dict(w) for w in reloaded.workers], [worker_to_dict(w) for w in registry.workers], ) def test_save_leaves_no_temp_files_behind(self): registry = self.parse(_document()) save_registry(registry, self.path) save_registry(registry, self.path) leftovers = [p.name for p in self.path.parent.iterdir() if p.name.startswith(".")] self.assertEqual(leftovers, []) def test_save_refuses_to_persist_an_invalid_document(self): registry = self.parse(_document()) broken = WorkerRegistry( version=registry.version, revision=registry.revision, updated_at=registry.updated_at, providers=registry.providers, # A worker whose provider is not declared in the registry. workers=tuple( type(worker)(**{**worker.__dict__, "provider": "vanished"}) for worker in registry.workers ), source_path=self.path, ) with self.assertRaises(RegistryValidationError): save_registry(broken, self.path) self.assertFalse(self.path.exists(), "invalid save must not create the file") def test_document_shape_excludes_local_paths_but_api_shape_includes_it(self): registry = self.parse(_document()) self.assertNotIn("source_path", registry_to_document(registry)) self.assertEqual(registry_to_dict(registry)["source_path"], str(self.path)) class TestVersioningAndRollback(_TempRegistryCase): """AC: versioning and rollback.""" def _seed(self) -> WorkerRegistry: self.write(_document()) return load_registry(self.path) def test_revision_increments_on_each_save(self): registry = self._seed() self.assertEqual(registry.revision, 1) second = save_registry(registry, self.path) self.assertEqual(second.revision, 2) third = save_registry(second, self.path) self.assertEqual(third.revision, 3) def test_updated_at_is_refreshed_and_utc(self): registry = self._seed() saved = save_registry(registry, self.path) self.assertRegex(saved.updated_at, r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") def test_superseded_revisions_are_retained(self): registry = self._seed() second = save_registry(registry, self.path) save_registry(second, self.path) self.assertEqual(list_revisions(self.path), (1, 2)) self.assertTrue(history_dir(self.path).is_dir()) def test_rollback_restores_prior_content_as_a_new_revision(self): self.write(_document(workers=[_worker("original")])) registry = load_registry(self.path) changed = WorkerRegistry( version=registry.version, revision=registry.revision, updated_at=registry.updated_at, providers=registry.providers, workers=(), # operator deletes every worker source_path=self.path, ) save_registry(changed, self.path) self.assertEqual(load_registry(self.path).workers, ()) restored = rollback_to_revision(1, self.path) self.assertEqual([w.id for w in restored.workers], ["original"]) # Append-only: the rollback publishes a new head rather than rewinding. self.assertGreater(restored.revision, 2) self.assertEqual([w.id for w in load_registry(self.path).workers], ["original"]) def test_rollback_to_unknown_revision_fails_closed(self): self._seed() with self.assertRaises(RegistryValidationError) as ctx: rollback_to_revision(99, self.path) self.assertIn("not retained", str(ctx.exception)) def test_revision_must_be_a_positive_integer(self): for bad in (0, -1, "1", None): with self.subTest(revision=bad): with self.assertRaises(RegistryValidationError): self.parse(_document(revision=bad)) def test_history_is_empty_before_any_save(self): self.write(_document()) self.assertEqual(list_revisions(self.path), ()) if __name__ == "__main__": unittest.main()