fix(release): validate published schemas and reuse accepted processing runtime

Author: fabricebrito

.github/workflows/build.yaml

@@ -6,7 +6,7 @@ on:
   workflow_dispatch:
     inputs:
       publish:
-        description: Publish the accepted stable 2.0.0 from main
+        description: Publish the accepted version from main
         type: boolean
         default: false
 permissions:
@@ -26,7 +26,8 @@ jobs:
           sudo apt-get update
           sudo apt-get install -y podman graphviz
           uv venv --python 3.12 .venv
-          uv pip sync --python .venv/bin/python --require-hashes requirements-notebooks.txt
+          uv pip sync --python .venv/bin/python --require-hashes requirements/development.txt
+          uv pip install --python .venv/bin/python --no-deps --no-build-isolation -e . -e runner
           uv pip check --python .venv/bin/python
       - name: Validate release contracts and wrapper API
         run: |
@@ -42,8 +43,13 @@ jobs:
           python scripts/prepare-oci.py --source-dir cwl-workflow --wrapped-dir acceptance-results/wrapped --output acceptance-results/annotations --revision "$GITHUB_SHA" --version "$(jq -er .version codemeta.json)"
       - name: Test the processing runner
         run: |
-          cd runner
-          ../.venv/bin/python -m nose2 -v
+          .venv/bin/python -m pytest -q
+      - name: Build the exact candidate processing runtime
+        run: |
+          mkdir -p acceptance-results/runtime
+          docker build -t ghcr.io/eoap/application-package-patterns/runner:acceptance runner
+          docker save -o acceptance-results/runtime/runner.tar ghcr.io/eoap/application-package-patterns/runner:acceptance
+          podman load -i acceptance-results/runtime/runner.tar
       - name: Start authenticated disposable S3
         run: |
           printf '%s' '{"identities":[{"name":"test","credentials":[{"accessKey":"test","secretKey":"test"}],"actions":["Admin","Read","Write","List","Tagging"]}]}' > /tmp/s3.json
@@ -54,7 +60,7 @@ jobs:
       - name: Execute all sixteen real-data wrapped cases and verify S3
         run: |
           source .venv/bin/activate
-          python scripts/run-acceptance.py --runner cwltool --output "$GITHUB_WORKSPACE/acceptance-results/workflows"
+          python scripts/run-acceptance.py --processing-image ghcr.io/eoap/application-package-patterns/runner:acceptance --runner cwltool --output "$GITHUB_WORKSPACE/acceptance-results/workflows"
       - name: Build documentation strictly
         run: .venv/bin/mkdocs build --strict --site-dir acceptance-results/site
       - uses: actions/upload-artifact@v4
@@ -87,12 +93,15 @@ jobs:
           if gh release view "$version" >/dev/null 2>&1; then echo 'Version already published'; exit 1; fi
           mkdir release-assets
           cp cwl-workflow/*.cwl cwl-workflow/*-inputs.json requirements-notebooks.txt requirements-oci.txt mastering-staging.lock.json release-assets/
-          cp -r templates release-assets/
+          cp -r templates requirements release-assets/
           mkdir release-assets/wrapped
           cp accepted/wrapped/*.cwl release-assets/wrapped/
           cp -r accepted/annotations release-assets/
           uv venv --python 3.12 .oci-venv
           uv pip sync --python .oci-venv/bin/python --require-hashes requirements-oci.txt
+          uv venv --python 3.12 .venv
+          uv pip sync --python .venv/bin/python --require-hashes requirements/development.txt
+          uv pip install --python .venv/bin/python --no-deps --no-build-isolation -e . -e runner
           cd release-assets
           find . -type f -print0 | sort -z | xargs -0 sha256sum > SHA256SUMS
       - name: Refuse to replace any existing CWL artifact
@@ -110,6 +119,22 @@ jobs:
               echo "Artifact already exists: $name-wrapped:$VERSION"; exit 1
             fi
           done
+      - name: Publish the exact accepted processing archive
+        run: |
+          image="ghcr.io/eoap/application-package-patterns/runner:$VERSION"
+          printf '%s' "$GH_TOKEN" | docker login ghcr.io --username "$GITHUB_ACTOR" --password-stdin
+          printf '%s' "$GH_TOKEN" | oras login ghcr.io --username "$GITHUB_ACTOR" --password-stdin
+          if oras resolve "$image" >/dev/null 2>&1; then echo "Image tag already published"; exit 1; fi
+          docker load -i accepted/runtime/runner.tar
+          docker tag ghcr.io/eoap/application-package-patterns/runner:acceptance "$image"
+          docker push "$image"
+          digest=$(oras resolve "$image")
+          [[ "$digest" =~ ^sha256:[a-f0-9]{64}$ ]]
+          reference="ghcr.io/eoap/application-package-patterns/runner@$digest"
+          .venv/bin/python -m patterns.release --image "$reference" --directory release-assets
+          printf '%s\n' "$reference" > release-assets/processing-image.txt
+          .venv/bin/python scripts/cwl-docs.py --source release-assets --validate --output accepted/published-cwl-docs
+          .oci-venv/bin/python scripts/prepare-oci.py --source-dir release-assets --wrapped-dir release-assets/wrapped --output release-assets/annotations --revision "$GITHUB_SHA" --version "$VERSION"
       - name: Publish and verify standalone and wrapped CWL on GHCR
         run: |
           printf '%s' "$GH_TOKEN" | oras login ghcr.io --username "$GITHUB_ACTOR" --password-stdin

cwl-workflow/test-custom-types.cwl

@@ -48,11 +48,6 @@ $graph:
         - ndvi
         - ndwi
         - none
-      - type: record
-        name: https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml#DateTime
-        fields:
-        - name: value
-          type: string
   inputs:
   - id: bbox
     type: https://raw.githubusercontent.com/eoap/schemas/main/ogc.yaml#BBox
@@ -146,11 +141,6 @@ $graph:
         - ndvi
         - ndwi
         - none
-      - type: record
-        name: https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml#DateTime
-        fields:
-        - name: value
-          type: string
   baseCommand:
   - echo
   arguments:

docs/helpers.py

@@ -174,6 +174,12 @@ def restore_imports(value):
                 result = []
                 imported = set()
                 for schema in value["types"]:
+                    existing_import = schema.get("$import") if isinstance(schema, dict) else None
+                    if existing_import in approved:
+                        if existing_import not in imported:
+                            result.append(schema)
+                            imported.add(existing_import)
+                        continue
                     name = schema.get("name", "") if isinstance(schema, dict) else ""
                     url = next((url for url in approved if name.startswith(url + "#")), None)
                     if url:

patterns/graph.py

@@ -16,7 +16,7 @@
 
 SCHEMA_ALIASES = {
     "https://raw.githubusercontent.com/eoap/schemas/main/ogc.yaml#": "#_shared_ogc_",
-    "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml#URI": "#URI",
+    "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml#": "#_shared_string_format_",
     "https://raw.githubusercontent.com/eoap/schemas/main/geojson.yaml#": "#_shared_geojson_",
 }
 
@@ -61,32 +61,34 @@ def load_document(path: Path) -> Process | list[Process]:
     if not isinstance(scoped, dict):
         raise ValueError("CWL graph metadata must be a mapping")
     result = load_cwl_from_yaml(scoped, uri=path.resolve().as_uri())
-    _restore_uri_ports(result if isinstance(result, list) else [result])
+    _restore_shared_ports(result if isinstance(result, list) else [result])
     return result
 
 
-def _restore_uri_ports(processes: list[Process]) -> None:
-    """Restore the published URI identifier required by the wrapping SDK."""
-    uri_type = "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml#URI"
+def _published_type(reference: str) -> str:
+    """Expose the original published identifier after temporary SDK scoping."""
+    fragment = "#" + reference.rsplit("#", 1)[-1] if "#" in reference else reference
+    for namespace, alias in SCHEMA_ALIASES.items():
+        if fragment.startswith(alias):
+            return namespace + fragment[len(alias) :]
+    return reference
+
+
+def _restore_shared_ports(processes: list[Process]) -> None:
+    """Restore published type identifiers for wrapping and lesson displays."""
     for process in processes:
         for port in [*process.inputs, *process.outputs]:
-            if isinstance(port.type_, str) and port.type_.endswith("#URI"):
-                port.type_ = uri_type
+            if isinstance(port.type_, str):
+                port.type_ = _published_type(port.type_)
             elif isinstance(port.type_, list):
-                port.type_ = [
-                    uri_type if isinstance(t, str) and t.endswith("#URI") else t for t in port.type_
-                ]
-            elif (
-                isinstance(
-                    port.type_,
-                    (
-                        InputArraySchema,
-                        OutputArraySchema,
-                        CommandInputArraySchema,
-                        CommandOutputArraySchema,
-                    ),
-                )
-                and isinstance(port.type_.items, str)
-                and port.type_.items.endswith("#URI")
-            ):
-                port.type_.items = uri_type
+                port.type_ = [_published_type(t) if isinstance(t, str) else t for t in port.type_]
+            elif isinstance(
+                port.type_,
+                (
+                    InputArraySchema,
+                    OutputArraySchema,
+                    CommandInputArraySchema,
+                    CommandOutputArraySchema,
+                ),
+            ) and isinstance(port.type_.items, str):
+                port.type_.items = _published_type(port.type_.items)

patterns/release.py

@@ -0,0 +1,52 @@
+"""Select a processing runtime for acceptance without rewriting shared staging."""
+
+from pathlib import Path
+
+import yaml
+
+RUNNER_IMAGE = "ghcr.io/eoap/application-package-patterns/runner"
+
+
+def _replace(value: object, image: str) -> int:
+    count = 0
+    if isinstance(value, dict):
+        pull = value.get("dockerPull")
+        if isinstance(pull, str) and pull.startswith((RUNNER_IMAGE + ":", RUNNER_IMAGE + "@")):
+            value["dockerPull"] = image
+            count += 1
+        for child in value.values():
+            count += _replace(child, image)
+    elif isinstance(value, list):
+        for child in value:
+            count += _replace(child, image)
+    return count
+
+
+def select_processing_image(path: Path, image: str) -> int:
+    """Override this course's runner only; retain Mastering images and imports."""
+    if not image.strip():
+        raise ValueError("Processing image must not be empty")
+    document = yaml.safe_load(path.read_text())
+    count = _replace(document, image)
+    if not count:
+        raise ValueError("Workflow does not reference this course's processing image")
+    path.write_text(yaml.safe_dump(document, sort_keys=False))
+    return count
+
+
+def main() -> None:
+    """Select the accepted runtime in release assets without rebuilding it."""
+    import argparse
+
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--image", required=True)
+    parser.add_argument("--directory", type=Path, required=True)
+    args = parser.parse_args()
+    for path in args.directory.rglob("*.cwl"):
+        if RUNNER_IMAGE in path.read_text():
+            select_processing_image(path, args.image)
+            print(f"Selected accepted processing runtime in {path}")
+
+
+if __name__ == "__main__":
+    main()

runner/tests/test_runtime_selection.py

@@ -0,0 +1,56 @@
+"""Candidate testing must never substitute the shared staging runtime."""
+
+import pytest
+import yaml
+
+from patterns.release import RUNNER_IMAGE, select_processing_image
+
+
+def test_only_owned_runner_is_changed(tmp_path):
+    path = tmp_path / "workflow.cwl"
+    path.write_text(
+        yaml.safe_dump(
+            {
+                "$graph": [
+                    {
+                        "requirements": {
+                            "DockerRequirement": {"dockerPull": RUNNER_IMAGE + "@sha256:old"}
+                        }
+                    },
+                    {
+                        "requirements": [
+                            {
+                                "class": "DockerRequirement",
+                                "dockerPull": "ghcr.io/eoap/mastering-app-package/stage-out:2.0.1",
+                            }
+                        ]
+                    },
+                    {
+                        "types": [
+                            {
+                                "$import": "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml"
+                            }
+                        ]
+                    },
+                ]
+            }
+        )
+    )
+    assert select_processing_image(path, "runner:accepted") == 1
+    result = yaml.safe_load(path.read_text())["$graph"]
+    assert result[0]["requirements"]["DockerRequirement"]["dockerPull"] == "runner:accepted"
+    assert (
+        result[1]["requirements"][0]["dockerPull"]
+        == "ghcr.io/eoap/mastering-app-package/stage-out:2.0.1"
+    )
+    assert (
+        result[2]["types"][0]["$import"]
+        == "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml"
+    )
+
+
+def test_missing_runner_is_an_error(tmp_path):
+    path = tmp_path / "workflow.cwl"
+    path.write_text("class: Workflow\n")
+    with pytest.raises(ValueError, match="processing image"):
+        select_processing_image(path, "runner:accepted")

runner/tests/test_schema_imports.py

@@ -42,3 +42,22 @@ def visit(value):
         for value in records
     )
     assert not any(str(value.get("$import", "")).startswith("eoap-types") for value in records)
+
+
+def test_custom_type_ports_retain_published_identifiers():
+    from patterns.graph import load_document
+
+    root = Path(__file__).resolve().parents[2]
+    processes = load_document(root / "cwl-workflow/test-custom-types.cwl")
+    assert isinstance(processes, list)
+    workflow = next(process for process in processes if process.id.endswith("test-custom-types"))
+    types = {port.id.rsplit("/", 1)[-1]: port.type_ for port in workflow.inputs}
+    assert (
+        types["start_time"]
+        == "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml#DateTime"
+    )
+    assert (
+        types["product_uri"]
+        == "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml#URI"
+    )
+    assert types["bbox"] == "https://raw.githubusercontent.com/eoap/schemas/main/ogc.yaml#BBox"

scripts/cwl-docs.py

@@ -2,6 +2,25 @@
 import argparse
 from pathlib import Path
 import yaml
+from urllib.parse import urlsplit
+from schema_salad.fetcher import DefaultFetcher
+from cwltool.context import LoadingContext
+
+APPROVED_SCHEMAS = (
+    "https://raw.githubusercontent.com/eoap/schemas/main/ogc.yaml",
+    "https://raw.githubusercontent.com/eoap/schemas/main/geojson.yaml",
+    "https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml",
+)
+
+
+class PublishedSchemaFetcher(DefaultFetcher):
+    """Fetch the approved published contracts through the schema SDK."""
+
+    def fetch_text(self, url: str, content_types: list[str] | None = None) -> str:
+        if urlsplit(url).scheme in {"http", "https"} and url.split("#")[0] not in APPROVED_SCHEMAS:
+            raise RuntimeError("Only approved published schemas may be fetched")
+        return super().fetch_text(url, content_types)
+
 
 
 def render(path):
@@ -14,7 +33,7 @@ def check(value):
         if isinstance(value, dict):
             assert '$include' not in value, f'{path}: external include'
             if '$import' in value:
-                assert value['$import'] in ('https://raw.githubusercontent.com/eoap/schemas/main/ogc.yaml', 'https://raw.githubusercontent.com/eoap/schemas/main/geojson.yaml'), f'{path}: unapproved schema import'
+                assert value['$import'] in ('https://raw.githubusercontent.com/eoap/schemas/main/ogc.yaml', 'https://raw.githubusercontent.com/eoap/schemas/main/geojson.yaml', 'https://raw.githubusercontent.com/eoap/schemas/main/string_format.yaml'), f'{path}: unapproved schema import'
             for child in value.values(): check(child)
         elif isinstance(value, list):
             for child in value: check(child)
@@ -53,29 +72,28 @@ def check(value):
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--validate', action='store_true', help='Also validate isolated CWL, allowing the approved OGC schema import')
     parser.add_argument('--output', type=Path, default=Path('/tmp/stac-cwl-docs'))
+    parser.add_argument("--source", type=Path, default=Path(__file__).resolve().parents[1])
     args = parser.parse_args()
-    root = Path(__file__).resolve().parents[1]
+    root = args.source
     for path in sorted(root.rglob('*.cwl')):
         if path.name.startswith('.') or not path.stat().st_size or any(part.startswith('.') for part in path.relative_to(root).parts): continue
         markdown = render(path)
         if args.validate:
-            import requests
             import shutil
             import tempfile
+            from cwltool.context import LoadingContext
             from cwltool.main import main
-            original_request = requests.sessions.Session.request
-            def blocked_request(session, method, url, **kwargs):
-                if url.split('#')[0] in ('https://raw.githubusercontent.com/eoap/schemas/main/ogc.yaml', 'https://raw.githubusercontent.com/eoap/schemas/main/geojson.yaml'):
-                    return original_request(session, method, url, **kwargs)
-                raise RuntimeError('Only the approved OGC schema may be fetched')
-            requests.sessions.Session.request = blocked_request
-            try:
-                with tempfile.TemporaryDirectory() as isolated:
-                    target_cwl = Path(isolated) / path.name
-                    shutil.copy(path, target_cwl)
-                    assert main(['--validate', str(target_cwl)]) == 0, path
-            finally:
-                requests.sessions.Session.request = original_request
+            document = yaml.safe_load(path.read_text())
+            graph = document.get("$graph")
+            entrypoints = [process["id"].rsplit("#", 1)[-1] for process in graph] if graph else [None]
+            with tempfile.TemporaryDirectory() as isolated:
+                target_cwl = Path(isolated) / path.name
+                shutil.copy(path, target_cwl)
+                for entrypoint in entrypoints:
+                    context = LoadingContext()
+                    context.fetcher_constructor = PublishedSchemaFetcher
+                    target = str(target_cwl) + ("#" + entrypoint if entrypoint else "")
+                    assert main(["--validate", target], loadingContext=context) == 0, (path, entrypoint)
         target = args.output / path.relative_to(root).with_suffix('.md')
         target.parent.mkdir(parents=True, exist_ok=True)
         target.write_text(markdown)