fix: allow platform source self-bindings

This commit is contained in:
lda
2026-06-15 01:00:55 +07:00 Verified
parent 340d923c9d
commit 93d3f06c0f
2 changed files with 36 additions and 2 deletions
+8 -2
View File
@@ -24,7 +24,11 @@ def validate_deployment_dependencies(
for logical_source, concrete_source in bindings.items(): for logical_source, concrete_source in bindings.items():
source = sources_by_id.get(logical_source) source = sources_by_id.get(logical_source)
if source is not None and source.platform: if (
source is not None
and source.platform
and concrete_source != logical_source
):
diagnostics.append( diagnostics.append(
DependencyDiagnostic( DependencyDiagnostic(
severity=DiagnosticSeverity.ERROR, severity=DiagnosticSeverity.ERROR,
@@ -47,7 +51,9 @@ def validate_deployment_dependencies(
if platform_source is not None and platform_source.platform: if platform_source is not None and platform_source.platform:
# Platform sources have fixed ids matching required.logical_source, so # Platform sources have fixed ids matching required.logical_source, so
# sources_by_id can resolve them directly and bindings.get(...) is # sources_by_id can resolve them directly and bindings.get(...) is
# intentionally bypassed when choosing bound_source_id. # intentionally bypassed when choosing bound_source_id. A legacy
# explicit self-binding such as wf.std -> wf.std is accepted for
# compatibility, but it has no effect.
bound_source_id = required.logical_source bound_source_id = required.logical_source
else: else:
bound_source_id = bindings.get(required.logical_source) bound_source_id = bindings.get(required.logical_source)
+28
View File
@@ -234,6 +234,34 @@ def test_platform_source_requirement_does_not_need_binding() -> None:
assert diagnostics == [] assert diagnostics == []
def test_platform_source_accepts_legacy_self_binding() -> None:
artifact = artifact_with(
required_capability(logical_source="wf.std", capability_name="replace")
)
deployment = WorkflowDeployment(
id="demo.default",
artifact_id=artifact.id,
artifact_version=artifact.version,
bindings={"wf.std": "wf.std"},
)
diagnostics = validate_deployment_dependencies(
artifact=artifact,
deployment=deployment,
sources=[
AvailableSource(
id="wf.std",
platform=True,
capabilities={
"replace": AvailableCapability(name="replace", kind="node_spec")
},
)
],
)
assert diagnostics == []
def test_platform_source_rejects_explicit_deployment_binding() -> None: def test_platform_source_rejects_explicit_deployment_binding() -> None:
artifact = artifact_with( artifact = artifact_with(
required_capability(logical_source="wf.std", capability_name="replace") required_capability(logical_source="wf.std", capability_name="replace")