test: cover python source loader edge cases

tests/fixtures/python_source_ops.py

@@ -24,3 +24,10 @@ def upper(payload: EchoInput) -> EchoOutput:
 
 
 registry = [echo, upper]
+
+
+def callable_registry():
+    return {"echo": echo, "upper": upper}
+
+
+duplicate_registry = [echo, echo]

tests/wf_sources_python/test_loader.py

@@ -21,6 +21,39 @@ def test_load_python_source_from_sequence_registry() -> None:
     assert source.permissions.safe_for_workflow is True
 
 
+def test_load_python_source_from_callable_registry() -> None:
+    source = load_python_source(
+        source_id="local.ops",
+        module="tests.fixtures.python_source_ops",
+        registry="callable_registry",
+    )
+
+    assert set(source.capabilities.node_specs) == {
+        "local.ops.echo",
+        "local.ops.upper",
+    }
+
+
+def test_load_python_source_propagates_enabled_flag() -> None:
+    source = load_python_source(
+        source_id="local.ops",
+        module="tests.fixtures.python_source_ops",
+        registry="registry",
+        enabled=False,
+    )
+
+    assert source.enabled is False
+
+
+def test_load_python_source_rejects_duplicate_qualified_names() -> None:
+    with pytest.raises(ValueError, match="duplicate NodeSpec names"):
+        load_python_source(
+            source_id="local.ops",
+            module="tests.fixtures.python_source_ops",
+            registry="duplicate_registry",
+        )
+
+
 def test_load_python_source_rejects_missing_registry() -> None:
     with pytest.raises(ValueError, match="missing registry object"):
         load_python_source(