From f4f19aec5462e80d830b54c900cc180e576b1993 Mon Sep 17 00:00:00 2001 From: Iftiquar Ali Date: Thu, 26 Mar 2026 18:34:35 +0530 Subject: [PATCH] add unit tests for agent_spec, source_registry, and service_registry Cover the three previously untested core modules with 94 new tests: - test_agent_spec (41 tests): frontmatter parsing, metadata validation, file discovery, agent selection, in-memory loading, slugification, and error paths for malformed specs. - test_source_registry (24 tests): default source registration, dependency resolution (transitive deps like sso->links), secret/config-field aggregation, custom registration, and error handling. - test_service_registry (29 tests): default service registration across infrastructure/application/integration categories, ServiceDefinition properties (volume/image/container names), dependency resolution, secret aggregation, category filtering, port configuration, and custom registration. --- tests/unit/test_agent_spec.py | 412 ++++++++++++++++++++++++++++ tests/unit/test_service_registry.py | 276 +++++++++++++++++++ tests/unit/test_source_registry.py | 179 ++++++++++++ 3 files changed, 867 insertions(+) create mode 100644 tests/unit/test_agent_spec.py create mode 100644 tests/unit/test_service_registry.py create mode 100644 tests/unit/test_source_registry.py diff --git a/tests/unit/test_agent_spec.py b/tests/unit/test_agent_spec.py new file mode 100644 index 000000000..fbf60e508 --- /dev/null +++ b/tests/unit/test_agent_spec.py @@ -0,0 +1,412 @@ +""" +Unit tests for agent spec loading, parsing, and selection. + +Tests cover: +- AgentSpec dataclass +- Frontmatter parsing (valid, empty, malformed) +- Metadata extraction (name, tools validation) +- File discovery (list_agent_files) +- Agent selection (select_agent_spec) +- In-memory loading (load_agent_spec_from_text) +- Filename slugification (slugify_agent_name) +""" + +import importlib.util +import sys +from pathlib import Path + +import pytest + +# Load agent_spec directly from its file path to avoid the heavy transitive +# dependencies pulled in by src.archi.pipelines.__init__ (LangChain, etc.). +_spec = importlib.util.spec_from_file_location( + "agent_spec", + str( + Path(__file__).resolve().parents[2] + / "src" + / "archi" + / "pipelines" + / "agents" + / "agent_spec.py" + ), +) +_mod = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = _mod +_spec.loader.exec_module(_mod) + +AgentSpec = _mod.AgentSpec +AgentSpecError = _mod.AgentSpecError +list_agent_files = _mod.list_agent_files +load_agent_spec = _mod.load_agent_spec +load_agent_spec_from_text = _mod.load_agent_spec_from_text +select_agent_spec = _mod.select_agent_spec +slugify_agent_name = _mod.slugify_agent_name + + +VALID_AGENT_MD = """\ +--- +name: Test Agent +tools: + - search_local_files + - search_vectorstore_hybrid +--- + +You are a helpful test assistant. +""" + +MINIMAL_AGENT_MD = """\ +--- +name: Minimal +tools: + - search_local_files +--- + +Answer concisely. +""" + +MCP_AGENT_MD = """\ +--- +name: MCP Research Agent +tools: + - search_vectorstore_hybrid + - fetch_catalog_document + - mcp +--- + +You are a research-focused assistant. +Use vectorstore tools for internal docs first. +Use MCP tools for external checks when internal evidence is insufficient. +""" + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def agents_dir(tmp_path): + """Create a temporary agents directory with sample spec files.""" + d = tmp_path / "agents" + d.mkdir() + (d / "01_default.md").write_text(VALID_AGENT_MD) + (d / "02_minimal.md").write_text(MINIMAL_AGENT_MD) + (d / "03_mcp.md").write_text(MCP_AGENT_MD) + return d + + +@pytest.fixture +def single_agent_dir(tmp_path): + """Directory with exactly one agent file.""" + d = tmp_path / "agents" + d.mkdir() + (d / "solo.md").write_text(VALID_AGENT_MD) + return d + + +@pytest.fixture +def empty_agents_dir(tmp_path): + """Directory that exists but contains no .md files.""" + d = tmp_path / "agents" + d.mkdir() + (d / "readme.txt").write_text("not an agent spec") + return d + + +# --------------------------------------------------------------------------- +# AgentSpec dataclass +# --------------------------------------------------------------------------- + + +class TestAgentSpecDataclass: + + def test_fields_are_accessible(self): + spec = AgentSpec( + name="Demo", + tools=["search_local_files"], + prompt="You are helpful.", + source_path=Path("/tmp/demo.md"), + ) + assert spec.name == "Demo" + assert spec.tools == ["search_local_files"] + assert spec.prompt == "You are helpful." + assert spec.source_path == Path("/tmp/demo.md") + + def test_frozen_prevents_mutation(self): + spec = AgentSpec( + name="Demo", + tools=["search_local_files"], + prompt="Prompt.", + source_path=Path("/tmp/demo.md"), + ) + with pytest.raises(AttributeError): + spec.name = "Changed" + + +# --------------------------------------------------------------------------- +# load_agent_spec +# --------------------------------------------------------------------------- + + +class TestLoadAgentSpec: + + def test_loads_valid_spec(self, agents_dir): + spec = load_agent_spec(agents_dir / "01_default.md") + assert spec.name == "Test Agent" + assert "search_local_files" in spec.tools + assert "search_vectorstore_hybrid" in spec.tools + assert "helpful test assistant" in spec.prompt + + def test_loads_mcp_spec(self, agents_dir): + spec = load_agent_spec(agents_dir / "03_mcp.md") + assert spec.name == "MCP Research Agent" + assert "mcp" in spec.tools + assert len(spec.tools) == 3 + + def test_source_path_matches_input(self, agents_dir): + path = agents_dir / "01_default.md" + spec = load_agent_spec(path) + assert spec.source_path == path + + def test_prompt_is_stripped(self, tmp_path): + content = "---\nname: Padded\ntools:\n - search_local_files\n---\n\n Prompt with whitespace. \n\n" + f = tmp_path / "padded.md" + f.write_text(content) + spec = load_agent_spec(f) + assert spec.prompt == "Prompt with whitespace." + + def test_tools_are_stripped(self, tmp_path): + content = ( + "---\nname: Spaces\ntools:\n - ' search_local_files '\n---\n\nPrompt body." + ) + f = tmp_path / "spaces.md" + f.write_text(content) + spec = load_agent_spec(f) + assert spec.tools == ["search_local_files"] + + +# --------------------------------------------------------------------------- +# load_agent_spec_from_text +# --------------------------------------------------------------------------- + + +class TestLoadAgentSpecFromText: + + def test_loads_from_string(self): + spec = load_agent_spec_from_text(VALID_AGENT_MD) + assert spec.name == "Test Agent" + assert spec.tools == ["search_local_files", "search_vectorstore_hybrid"] + assert "helpful test assistant" in spec.prompt + + def test_source_path_is_memory_sentinel(self): + spec = load_agent_spec_from_text(VALID_AGENT_MD) + assert spec.source_path == Path("") + + def test_round_trips_with_file_load(self, agents_dir): + path = agents_dir / "01_default.md" + from_file = load_agent_spec(path) + from_text = load_agent_spec_from_text(path.read_text()) + assert from_file.name == from_text.name + assert from_file.tools == from_text.tools + assert from_file.prompt == from_text.prompt + + +# --------------------------------------------------------------------------- +# Frontmatter parsing edge cases +# --------------------------------------------------------------------------- + + +class TestFrontmatterParsing: + + def test_empty_file_raises(self, tmp_path): + f = tmp_path / "empty.md" + f.write_text("") + with pytest.raises(AgentSpecError, match="empty"): + load_agent_spec(f) + + def test_missing_opening_fence_raises(self, tmp_path): + f = tmp_path / "no_fence.md" + f.write_text("name: Test\ntools:\n - x\n---\nPrompt.") + with pytest.raises(AgentSpecError, match="missing YAML frontmatter"): + load_agent_spec(f) + + def test_missing_closing_fence_raises(self, tmp_path): + f = tmp_path / "no_close.md" + f.write_text("---\nname: Test\ntools:\n - x\nPrompt without closing fence.") + with pytest.raises(AgentSpecError, match="missing closing"): + load_agent_spec(f) + + def test_empty_prompt_body_raises(self, tmp_path): + f = tmp_path / "no_prompt.md" + f.write_text("---\nname: Test\ntools:\n - x\n---\n") + with pytest.raises(AgentSpecError, match="prompt body is empty"): + load_agent_spec(f) + + def test_invalid_yaml_raises(self, tmp_path): + f = tmp_path / "bad_yaml.md" + f.write_text("---\n: [invalid yaml\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="invalid YAML"): + load_agent_spec(f) + + def test_leading_blank_lines_are_skipped(self, tmp_path): + content = "\n\n\n---\nname: Blank\ntools:\n - search_local_files\n---\n\nBody." + f = tmp_path / "blanks.md" + f.write_text(content) + spec = load_agent_spec(f) + assert spec.name == "Blank" + + def test_frontmatter_not_a_dict_raises(self, tmp_path): + f = tmp_path / "list_fm.md" + f.write_text("---\n- item1\n- item2\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="must be a mapping"): + load_agent_spec(f) + + +# --------------------------------------------------------------------------- +# Metadata validation +# --------------------------------------------------------------------------- + + +class TestMetadataExtraction: + + def test_missing_name_raises(self, tmp_path): + f = tmp_path / "no_name.md" + f.write_text("---\ntools:\n - search_local_files\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="name"): + load_agent_spec(f) + + def test_non_string_name_raises(self, tmp_path): + f = tmp_path / "int_name.md" + f.write_text("---\nname: 123\ntools:\n - search_local_files\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="name"): + load_agent_spec(f) + + def test_missing_tools_raises(self, tmp_path): + f = tmp_path / "no_tools.md" + f.write_text("---\nname: NoTools\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="tools"): + load_agent_spec(f) + + def test_empty_tools_list_raises(self, tmp_path): + f = tmp_path / "empty_tools.md" + f.write_text("---\nname: EmptyTools\ntools: []\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="tools"): + load_agent_spec(f) + + def test_non_list_tools_raises(self, tmp_path): + f = tmp_path / "str_tools.md" + f.write_text("---\nname: StrTools\ntools: search_local_files\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="tools"): + load_agent_spec(f) + + def test_tools_with_non_string_entry_raises(self, tmp_path): + f = tmp_path / "mixed_tools.md" + f.write_text( + "---\nname: Mixed\ntools:\n - search_local_files\n - 42\n---\n\nPrompt." + ) + with pytest.raises(AgentSpecError, match="tools"): + load_agent_spec(f) + + def test_tool_with_only_whitespace_raises(self, tmp_path): + f = tmp_path / "ws_tool.md" + f.write_text("---\nname: WSTool\ntools:\n - ' '\n---\n\nPrompt.") + with pytest.raises(AgentSpecError, match="tools"): + load_agent_spec(f) + + +# --------------------------------------------------------------------------- +# list_agent_files +# --------------------------------------------------------------------------- + + +class TestListAgentFiles: + + def test_returns_sorted_md_files(self, agents_dir): + files = list_agent_files(agents_dir) + assert len(files) == 3 + names = [f.name for f in files] + assert names == sorted(names) + + def test_ignores_non_md_files(self, agents_dir): + (agents_dir / "notes.txt").write_text("ignore me") + (agents_dir / "data.json").write_text("{}") + files = list_agent_files(agents_dir) + assert all(f.suffix == ".md" for f in files) + + def test_nonexistent_dir_raises(self, tmp_path): + with pytest.raises(AgentSpecError, match="not found"): + list_agent_files(tmp_path / "nonexistent") + + def test_file_path_instead_of_dir_raises(self, tmp_path): + f = tmp_path / "file.md" + f.write_text("content") + with pytest.raises(AgentSpecError, match="not a directory"): + list_agent_files(f) + + def test_empty_dir_returns_empty_list(self, empty_agents_dir): + files = list_agent_files(empty_agents_dir) + assert files == [] + + +# --------------------------------------------------------------------------- +# select_agent_spec +# --------------------------------------------------------------------------- + + +class TestSelectAgentSpec: + + def test_selects_by_name(self, agents_dir): + spec = select_agent_spec(agents_dir, agent_name="Minimal") + assert spec.name == "Minimal" + + def test_selects_first_when_no_name_given(self, agents_dir): + spec = select_agent_spec(agents_dir) + assert spec.name == "Test Agent" # 01_default.md is first lexicographically + + def test_unknown_name_raises(self, agents_dir): + with pytest.raises(AgentSpecError, match="not found"): + select_agent_spec(agents_dir, agent_name="Does Not Exist") + + def test_empty_dir_raises(self, empty_agents_dir): + with pytest.raises(AgentSpecError, match="No agent markdown files"): + select_agent_spec(empty_agents_dir) + + def test_single_agent_returns_it(self, single_agent_dir): + spec = select_agent_spec(single_agent_dir) + assert spec.name == "Test Agent" + + +# --------------------------------------------------------------------------- +# slugify_agent_name +# --------------------------------------------------------------------------- + + +class TestSlugifyAgentName: + + def test_simple_name(self): + assert slugify_agent_name("CMS CompOps") == "cms-compops.md" + + def test_special_characters(self): + result = slugify_agent_name("Agent: V2 (beta)") + assert result.endswith(".md") + assert " " not in result + assert ":" not in result + assert "(" not in result + + def test_leading_trailing_whitespace(self): + result = slugify_agent_name(" My Agent ") + assert result == "my-agent.md" + + def test_empty_string_returns_default(self): + assert slugify_agent_name("") == "agent.md" + + def test_only_special_chars_returns_default(self): + assert slugify_agent_name("!!!@@@") == "agent.md" + + def test_numeric_name(self): + result = slugify_agent_name("Agent 42") + assert result == "agent-42.md" + + def test_consecutive_specials_collapse(self): + result = slugify_agent_name("a---b___c") + assert result == "a-b-c.md" diff --git a/tests/unit/test_service_registry.py b/tests/unit/test_service_registry.py new file mode 100644 index 000000000..48f32bf48 --- /dev/null +++ b/tests/unit/test_service_registry.py @@ -0,0 +1,276 @@ +""" +Unit tests for the service registry. + +Tests cover: +- Default service registration (infrastructure, application, integration) +- Service definition properties (volume names, image names, container names) +- Dependency resolution +- Secret aggregation +- Category filtering +- Custom service registration +""" + +import pytest + +from src.cli.service_registry import ServiceDefinition, ServiceRegistry + + +@pytest.fixture +def registry(): + """Fresh registry instance with default services.""" + return ServiceRegistry() + + +# --------------------------------------------------------------------------- +# Default services +# --------------------------------------------------------------------------- + + +class TestDefaultServices: + + def test_infrastructure_services_registered(self, registry): + infra = registry.get_infrastructure_services() + assert "data-manager" in infra + assert "postgres" in infra + + def test_application_services_registered(self, registry): + apps = registry.get_application_services() + assert "chatbot" in apps + assert "grafana" in apps + assert "grader" in apps + + def test_integration_services_registered(self, registry): + integrations = registry.get_integration_services() + assert "piazza" in integrations + assert "mattermost" in integrations + assert "redmine-mailer" in integrations + + def test_all_services_returns_complete_set(self, registry): + all_svc = registry.get_all_services() + expected = { + "data-manager", + "postgres", + "chatbot", + "grafana", + "grader", + "piazza", + "mattermost", + "redmine-mailer", + "benchmarking", + } + assert expected == set(all_svc.keys()) + + +# --------------------------------------------------------------------------- +# ServiceDefinition properties +# --------------------------------------------------------------------------- + + +class TestServiceDefinition: + + def test_get_volume_name_with_pattern(self): + svc = ServiceDefinition( + name="postgres", + description="DB", + category="infrastructure", + requires_volume=True, + volume_name_pattern="archi-pg-{name}", + ) + assert svc.get_volume_name("mybot") == "archi-pg-mybot" + + def test_get_volume_name_default_pattern(self): + svc = ServiceDefinition( + name="chatbot", + description="Chat", + category="application", + requires_volume=True, + ) + assert svc.get_volume_name("demo") == "archi-demo" + + def test_get_volume_name_none_when_not_required(self): + svc = ServiceDefinition( + name="mattermost", + description="Mattermost", + category="integration", + requires_volume=False, + ) + assert svc.get_volume_name("demo") is None + + def test_get_image_name(self): + svc = ServiceDefinition( + name="chatbot", + description="Chat", + category="application", + ) + assert svc.get_image_name("prod") == "chatbot-prod" + + def test_get_image_name_none_when_no_image(self): + svc = ServiceDefinition( + name="external", + description="External", + category="integration", + requires_image=False, + ) + assert svc.get_image_name("prod") is None + + def test_get_container_name(self): + svc = ServiceDefinition( + name="chatbot", + description="Chat", + category="application", + ) + assert svc.get_container_name("dev") == "chatbot-dev" + + +# --------------------------------------------------------------------------- +# Dependency resolution +# --------------------------------------------------------------------------- + + +class TestDependencyResolution: + + def test_chatbot_pulls_in_postgres(self, registry): + resolved = registry.resolve_dependencies(["chatbot"]) + assert "postgres" in resolved + + def test_infrastructure_always_included(self, registry): + resolved = registry.resolve_dependencies(["chatbot"]) + assert "data-manager" in resolved + assert "postgres" in resolved + + def test_no_duplicates(self, registry): + resolved = registry.resolve_dependencies(["chatbot", "grafana", "grader"]) + assert len(resolved) == len(set(resolved)) + + def test_empty_input_still_includes_auto_enable(self, registry): + resolved = registry.resolve_dependencies([]) + assert "data-manager" in resolved + assert "postgres" in resolved + + def test_unknown_service_skipped(self, registry): + resolved = registry.resolve_dependencies(["nonexistent"]) + infra = registry.get_infrastructure_services() + for name in infra: + assert name in resolved + + def test_grafana_pulls_in_postgres(self, registry): + resolved = registry.resolve_dependencies(["grafana"]) + assert "postgres" in resolved + + +# --------------------------------------------------------------------------- +# Secret aggregation +# --------------------------------------------------------------------------- + + +class TestRequiredSecrets: + + def test_chatbot_no_required_secrets(self, registry): + secrets = registry.get_required_secrets(["chatbot"]) + assert secrets == set() + + def test_grafana_requires_grafana_password(self, registry): + secrets = registry.get_required_secrets(["grafana"]) + assert "GRAFANA_PG_PASSWORD" in secrets + + def test_piazza_secrets(self, registry): + secrets = registry.get_required_secrets(["piazza"]) + assert "PIAZZA_EMAIL" in secrets + assert "PIAZZA_PASSWORD" in secrets + assert "SLACK_WEBHOOK" in secrets + + def test_multiple_services_union(self, registry): + secrets = registry.get_required_secrets(["grafana", "piazza"]) + assert "GRAFANA_PG_PASSWORD" in secrets + assert "PIAZZA_EMAIL" in secrets + + def test_unknown_service_ignored(self, registry): + secrets = registry.get_required_secrets(["nonexistent"]) + assert secrets == set() + + +# --------------------------------------------------------------------------- +# Category filtering +# --------------------------------------------------------------------------- + + +class TestCategoryFiltering: + + def test_get_services_by_category(self, registry): + infra = registry.get_services_by_category("infrastructure") + assert all(s.category == "infrastructure" for s in infra.values()) + + def test_unknown_category_returns_empty(self, registry): + result = registry.get_services_by_category("nonexistent") + assert result == {} + + +# --------------------------------------------------------------------------- +# Custom registration +# --------------------------------------------------------------------------- + + +class TestCustomRegistration: + + def test_register_new_service(self, registry): + registry.register( + ServiceDefinition( + name="discord", + description="Discord bot integration", + category="integration", + required_secrets=["DISCORD_TOKEN"], + ) + ) + svc = registry.get_service("discord") + assert svc.description == "Discord bot integration" + assert "DISCORD_TOKEN" in svc.required_secrets + + def test_overwrite_existing_service(self, registry): + registry.register( + ServiceDefinition( + name="chatbot", + description="Updated chatbot", + category="application", + ) + ) + assert registry.get_service("chatbot").description == "Updated chatbot" + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- + + +class TestErrorHandling: + + def test_get_unknown_service_raises(self, registry): + with pytest.raises(ValueError, match="Unknown service"): + registry.get_service("nonexistent_service") + + +# --------------------------------------------------------------------------- +# Port configuration +# --------------------------------------------------------------------------- + + +class TestPortConfiguration: + + def test_chatbot_default_ports(self, registry): + svc = registry.get_service("chatbot") + assert svc.default_host_port == 7861 + assert svc.default_container_port == 7861 + + def test_data_manager_default_ports(self, registry): + svc = registry.get_service("data-manager") + assert svc.default_host_port == 7871 + assert svc.default_container_port == 7871 + + def test_grafana_default_ports(self, registry): + svc = registry.get_service("grafana") + assert svc.default_host_port == 3000 + assert svc.default_container_port == 3000 + + def test_postgres_has_no_default_ports(self, registry): + svc = registry.get_service("postgres") + assert svc.default_host_port is None + assert svc.default_container_port is None diff --git a/tests/unit/test_source_registry.py b/tests/unit/test_source_registry.py new file mode 100644 index 000000000..f31111fcf --- /dev/null +++ b/tests/unit/test_source_registry.py @@ -0,0 +1,179 @@ +""" +Unit tests for the data-source registry. + +Tests cover: +- Default source registration (links, sso, git, jira, redmine) +- Dependency resolution +- Secret and config-field aggregation +- Custom source registration +- Error handling for unknown sources +""" + +import pytest + +from src.cli.source_registry import SourceDefinition, SourceRegistry + + +@pytest.fixture +def registry(): + """Fresh registry instance with default sources.""" + return SourceRegistry() + + +# --------------------------------------------------------------------------- +# Default sources +# --------------------------------------------------------------------------- + + +class TestDefaultSources: + + def test_links_registered(self, registry): + defn = registry.get("links") + assert defn.name == "links" + assert defn.required_secrets == [] + + def test_sso_registered(self, registry): + defn = registry.get("sso") + assert "SSO_USERNAME" in defn.required_secrets + assert "SSO_PASSWORD" in defn.required_secrets + + def test_git_registered(self, registry): + defn = registry.get("git") + assert "GIT_USERNAME" in defn.required_secrets + assert "GIT_TOKEN" in defn.required_secrets + + def test_jira_registered(self, registry): + defn = registry.get("jira") + assert "JIRA_PAT" in defn.required_secrets + + def test_redmine_registered(self, registry): + defn = registry.get("redmine") + assert "REDMINE_USER" in defn.required_secrets + assert "REDMINE_PW" in defn.required_secrets + + def test_names_returns_sorted_list(self, registry): + names = registry.names() + assert names == sorted(names) + assert set(names) == {"git", "jira", "links", "redmine", "sso"} + + +# --------------------------------------------------------------------------- +# Dependency resolution +# --------------------------------------------------------------------------- + + +class TestDependencyResolution: + + def test_sso_pulls_in_links(self, registry): + resolved = registry.resolve_dependencies(["sso"]) + assert "links" in resolved + assert resolved.index("links") < resolved.index("sso") + + def test_git_pulls_in_links(self, registry): + resolved = registry.resolve_dependencies(["git"]) + assert "links" in resolved + + def test_links_has_no_extra_deps(self, registry): + resolved = registry.resolve_dependencies(["links"]) + assert resolved == ["links"] + + def test_jira_standalone(self, registry): + resolved = registry.resolve_dependencies(["jira"]) + assert resolved == ["jira"] + + def test_multiple_sources_deduplicated(self, registry): + resolved = registry.resolve_dependencies(["sso", "git", "links"]) + assert resolved.count("links") == 1 + + def test_unknown_source_skipped(self, registry): + resolved = registry.resolve_dependencies(["nonexistent"]) + assert resolved == [] + + def test_empty_list_resolved(self, registry): + assert registry.resolve_dependencies([]) == [] + + +# --------------------------------------------------------------------------- +# Secret aggregation +# --------------------------------------------------------------------------- + + +class TestRequiredSecrets: + + def test_single_source_secrets(self, registry): + secrets = registry.required_secrets(["jira"]) + assert "JIRA_PAT" in secrets + + def test_transitive_secrets_included(self, registry): + secrets = registry.required_secrets(["sso"]) + assert "SSO_USERNAME" in secrets + assert "SSO_PASSWORD" in secrets + + def test_no_duplicates(self, registry): + secrets = registry.required_secrets(["sso", "git"]) + assert len(secrets) == len(set(secrets)) + + def test_links_has_no_required_secrets(self, registry): + assert registry.required_secrets(["links"]) == [] + + +# --------------------------------------------------------------------------- +# Config field aggregation +# --------------------------------------------------------------------------- + + +class TestRequiredConfigFields: + + def test_links_config_fields(self, registry): + fields = registry.required_config_fields(["links"]) + assert "data_manager.sources.links.input_lists" in fields + + def test_jira_config_fields(self, registry): + fields = registry.required_config_fields(["jira"]) + assert "data_manager.sources.jira.url" in fields + assert "data_manager.sources.jira.projects" in fields + + def test_no_duplicates(self, registry): + fields = registry.required_config_fields(["links", "sso"]) + assert len(fields) == len(set(fields)) + + +# --------------------------------------------------------------------------- +# Custom registration +# --------------------------------------------------------------------------- + + +class TestCustomRegistration: + + def test_register_new_source(self, registry): + registry.register( + SourceDefinition( + name="confluence", + description="Confluence wiki scraper", + required_secrets=["CONFLUENCE_TOKEN"], + ) + ) + defn = registry.get("confluence") + assert defn.description == "Confluence wiki scraper" + assert "confluence" in registry.names() + + def test_overwrite_existing_source(self, registry): + registry.register( + SourceDefinition( + name="links", + description="Updated links source", + ) + ) + assert registry.get("links").description == "Updated links source" + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- + + +class TestErrorHandling: + + def test_get_unknown_source_raises(self, registry): + with pytest.raises(KeyError, match="Unknown source"): + registry.get("nonexistent_source")