refactor: split generic source registry mechanics

tests/wf_api/test_source_registry.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from pydantic import Field
+
+from wf_api.source_registry import (
+    AtomicJsonRegistryStore,
+    SourceRegistryBaseModel,
+    validate_source_registry_id,
+    validate_unique_source_ids,
+)
+
+
+class FakeEntry(SourceRegistryBaseModel):
+    id: str
+
+
+class FakeRegistry(SourceRegistryBaseModel):
+    version: int = 1
+    entries: list[FakeEntry] = Field(default_factory=list)
+
+
+def test_validate_source_registry_id_accepts_normal_ids() -> None:
+    assert validate_source_registry_id("github.work") == "github.work"
+    assert validate_source_registry_id("my_source") == "my_source"
+    assert validate_source_registry_id("test-v1") == "test-v1"
+
+
+def test_validate_source_registry_id_rejects_unsafe_ids() -> None:
+    with pytest.raises(ValueError, match="source id must start"):
+        validate_source_registry_id("../bad")
+    with pytest.raises(ValueError, match="source id must start"):
+        validate_source_registry_id("has space")
+    with pytest.raises(ValueError, match="source id must start"):
+        validate_source_registry_id("")
+
+
+def test_validate_unique_source_ids_accepts_unique_ids() -> None:
+    entries = [FakeEntry(id="a"), FakeEntry(id="b")]
+    validate_unique_source_ids(entries)
+
+
+def test_validate_unique_source_ids_rejects_duplicate_ids() -> None:
+    entries = [FakeEntry(id="a"), FakeEntry(id="a")]
+    with pytest.raises(ValueError, match="duplicate source id 'a'"):
+        validate_unique_source_ids(entries)
+
+
+def test_validate_unique_source_ids_rejects_non_string_id() -> None:
+    entries: list[object] = [FakeEntry(id="a"), object()]
+    with pytest.raises(
+        ValueError, match="source registry entries must expose string id"
+    ):
+        validate_unique_source_ids(entries)
+
+
+def test_atomic_json_registry_store_loads_empty_when_missing(tmp_path: Path) -> None:
+    store = AtomicJsonRegistryStore(
+        tmp_path,
+        filename="registry.json",
+        registry_type=FakeRegistry,
+        empty_factory=FakeRegistry,
+        corrupt_label="test registry",
+    )
+
+    registry = store.load_registry()
+
+    assert registry.version == 1
+    assert registry.entries == []
+    assert store.path == tmp_path / "registry.json"
+
+
+def test_atomic_json_registry_store_round_trips(tmp_path: Path) -> None:
+    store = AtomicJsonRegistryStore(
+        tmp_path,
+        filename="registry.json",
+        registry_type=FakeRegistry,
+        empty_factory=FakeRegistry,
+        corrupt_label="test registry",
+    )
+    registry = FakeRegistry(entries=[FakeEntry(id="test.entry")])
+
+    store.save_registry(registry)
+    loaded = store.load_registry()
+
+    assert len(loaded.entries) == 1
+    assert loaded.entries[0].id == "test.entry"
+
+
+def test_atomic_json_registry_store_rejects_corrupted_json(tmp_path: Path) -> None:
+    store = AtomicJsonRegistryStore(
+        tmp_path,
+        filename="registry.json",
+        registry_type=FakeRegistry,
+        empty_factory=FakeRegistry,
+        corrupt_label="test registry",
+    )
+    store.path.write_text("not json{{{", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="corrupted"):
+        store.load_registry()
+
+
+def test_source_registry_base_model_rejects_extra_fields() -> None:
+    with pytest.raises(ValueError, match="Extra inputs are not permitted"):
+        SourceRegistryBaseModel.model_validate({"unknown_field": "value"})

tests/wf_mcp/test_source_registry.py

@@ -10,6 +10,7 @@ from wf_mcp.source_registry import (
     McpSourceRegistryEntry,
     SourceRegistryFile,
     StdioSourceTransport,
+    registry_entry_to_connection_config,
 )
 
 
@@ -52,33 +53,11 @@ def test_source_registry_accepts_http_transport() -> None:
     assert str(entry.transport.url) == "https://example.test/mcp"
 
 
-def test_source_registry_rejects_duplicate_ids() -> None:
-    with pytest.raises(ValueError, match="duplicate source id 'github.work'"):
-        SourceRegistryFile(sources=[_entry("github.work"), _entry("github.work")])
-
-
 def test_source_registry_rejects_reserved_ids() -> None:
     with pytest.raises(ValueError, match="reserved"):
         _entry("wf.admin")
 
 
-def test_source_registry_rejects_unsafe_ids() -> None:
-    with pytest.raises(ValueError, match="connection id"):
-        _entry("../bad")
-
-
-def test_file_source_registry_store_loads_empty_registry_when_missing(
-    tmp_path: Path,
-) -> None:
-    store = FileSourceRegistryStore(tmp_path)
-
-    registry = store.load_registry()
-
-    assert registry.version == 1
-    assert registry.sources == []
-    assert store.path == tmp_path / "source_registry.json"
-
-
 def test_file_source_registry_store_round_trips_registry(tmp_path: Path) -> None:
     store = FileSourceRegistryStore(tmp_path)
     registry = SourceRegistryFile(sources=[_entry("github.work")])
@@ -102,9 +81,38 @@ def test_file_source_registry_store_validates_loaded_registry(tmp_path: Path) ->
         store.load_registry()
 
 
-def test_file_source_registry_store_rejects_corrupted_json(tmp_path: Path) -> None:
-    store = FileSourceRegistryStore(tmp_path)
-    store.path.write_text("not json{{{", encoding="utf-8")
+def test_registry_entry_to_connection_config_preserves_identity() -> None:
+    entry = _entry()
+    config = registry_entry_to_connection_config(entry)
 
-    with pytest.raises(ValueError, match="corrupted"):
-        store.load_registry()
+    assert config.id == "github.work"
+    assert config.server == "github"
+    assert config.account == "work"
+    assert config.enabled is True
+
+
+def test_registry_entry_to_connection_config_preserves_transport_metadata() -> None:
+    entry = _entry()
+    entry.auth_ref = "github.work.auth"
+    config = registry_entry_to_connection_config(entry)
+
+    assert config.metadata["auth_ref"] == "github.work.auth"
+    assert config.metadata["profile"] is None
+    assert config.metadata["transport"]["kind"] == "stdio"
+    assert config.metadata["transport"]["command"] == "npx"
+    assert config.metadata["source_registry"] is True
+
+
+def test_registry_entry_to_connection_config_preserves_user_metadata() -> None:
+    entry = _entry()
+    config = registry_entry_to_connection_config(entry)
+
+    assert config.metadata["purpose"] == "tests"
+
+
+def test_registry_entry_to_connection_config_disabled_entry() -> None:
+    entry = _entry()
+    entry.enabled = False
+    config = registry_entry_to_connection_config(entry)
+
+    assert config.enabled is False