Run canonical SDK compatibility through a pinned CI executor tag

    - apt-get update -qq && apt-get install -y --no-install-recommends git

    - pip install uv==0.11.26

  script:

    - python ci/sdk_consumer.py

    - |

      if [ "${CI_COMMIT_TAG:-}" = "sdk-consumer-runner-v1" ] || [ -n "${SDK_EXECUTOR_SHA:-}" ]; then

        python ci/sdk_consumer_executor.py

      else

        python ci/sdk_consumer.py

      fi

  artifacts:

    when: always

    paths:

its own checkout against `EXPECTED_SRM_SHA` before installing anything. A changed

branch head requires a new run.

## Canonical check execution

Automatic compatibility checks start on the protected, versioned

`sdk-consumer-runner-v1` tag, whose exact source is pinned by SDK. Developers

may initiate pipelines on this one tag; main/develop protections stay unchanged.

The executor verifies the server pipeline and protected tag, checks the selected

main/develop head, fetches that exact commit and checks it out before running the

existing compatibility recipe. A changed head, wrong executor or wrong source

fails before installing the SDK. No personal credential is introduced.

The qualification retains the actual tested `source_sha` and records the

executor's `ref`/`source_sha` separately. This initial executor supports only

canonical compatibility tests (`SDK_BUILD_IMAGE=0`). Explicit MR adoption and

optional selected-consumer image pipelines keep their existing direct source

execution and image receipts. Selecting a protected branch for an optional

image experiment still requires that branch's existing execution permission.

This tag does not qualify Developer access to that additional experiment path.

## Compatibility without adoption

SDK target-branch checks and branch experiments set

"""Run canonical compatibility checks from a versioned CI execution tag."""

import json

import os

import re

import subprocess

from pathlib import Path

from urllib.request import urlopen

# Load the reviewed runner before replacing its checkout with the tested source.

import sdk_consumer

ROOT = Path(__file__).resolve().parents[1]

API = "https://labs.etsi.org/rep/api/v4/"

EXECUTOR_REF = "sdk-consumer-runner-v1"

SHA = re.compile(r"[a-f0-9]{40}")

def read(path):

    # These project metadata endpoints are public; do not forward a credential.

    with urlopen(API + path, timeout=30) as response:

        return json.load(response)

def validate(environment, head, get=read):

    require = sdk_consumer.require

    require(

        environment.get("CI_PROJECT_ID") == "377"

        and environment.get("SDK_PROJECT_ID") == "396"

        and environment.get("CI_PIPELINE_SOURCE") == "pipeline"

        and environment.get("SDK_CONSUMER_MODE") == "compatibility"

        and environment.get("SDK_BUILD_IMAGE") == "0",

        "The execution tag accepts only SDK canonical compatibility checks",

    )

    branch = environment.get("EXPECTED_SRM_BRANCH", "")

    source = environment.get("EXPECTED_SRM_SHA", "")

    executor = environment.get("SDK_EXECUTOR_SHA", "")

    pipeline_id = environment.get("CI_PIPELINE_ID", "")

    require(branch in {"main", "develop"}, "Use an actual canonical SRM branch")

    require(

        SHA.fullmatch(source) and SHA.fullmatch(executor) and pipeline_id.isdecimal(),

        "Exact tested-source and executor identities are required",

    )

    require(

        environment.get("CI_COMMIT_TAG") == EXECUTOR_REF

        and environment.get("CI_COMMIT_SHA") == head == executor,

        "Consumer executor checkout differs from its pinned tag",

    )

    pipeline = get(f"projects/377/pipelines/{pipeline_id}")

    tag = get(f"projects/377/repository/tags/{EXECUTOR_REF}")

    require(

        pipeline.get("sha") == executor

        and pipeline.get("ref") == EXECUTOR_REF

        and pipeline.get("source") == "pipeline"

        and tag.get("commit", {}).get("id") == executor

        and tag.get("protected") is True,

        "Server pipeline and protected executor tag must match the pin",

    )

    check_target(branch, source, get)

    return {"ref": EXECUTOR_REF, "source_sha": executor}

def check_target(branch, source, get=read):

    current = get("projects/377/repository/branches/" + branch)

    sdk_consumer.require(

        current.get("commit", {}).get("id") == source,

        "Canonical SRM head changed during selection; start a new run",

    )

def execute(root=ROOT, environment=None, get=read, call=subprocess.run, qualify=None):

    environment = dict(os.environ if environment is None else environment)

    root = Path(root).resolve()

    def git(*args):

        return call(

            ["git", *args], cwd=root, check=True, capture_output=True, text=True

        ).stdout.strip()

    executor = validate(environment, git("rev-parse", "HEAD"), get)

    sdk_consumer.require(

        not git("status", "--porcelain", "--untracked-files=no"),

        "Executor checkout has modified tracked files",

    )

    source = environment["EXPECTED_SRM_SHA"]

    git("fetch", "--quiet", "--no-tags", "--depth=1", "origin", source)

    git("checkout", "--quiet", "--detach", source)

    sdk_consumer.require(git("rev-parse", "HEAD") == source, "Wrong tested SRM checkout")

    check_target(environment["EXPECTED_SRM_BRANCH"], source, get)

    # Reuse the existing locked-then-SDK-wheel recipe on the actual target tree.

    out = root / ".cache/sdk-consumer"

    proof = (qualify or sdk_consumer.qualify)(root=root, out=out)

    sdk_consumer.require(

        proof.get("source_sha") == source

        and proof.get("status") == "passed"

        and proof.get("qualification_mode") == "compatibility",

        "Consumer result did not qualify the selected canonical source",

    )

    proof.update(executor=executor, target_branch=environment["EXPECTED_SRM_BRANCH"])

    sdk_consumer.write(out / "qualification.json", proof)

    return proof

if __name__ == "__main__":

    execute()

"""The execution ref cannot replace or relabel the actual canonical consumer."""

import json

import subprocess

import sys

from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "ci"))

import sdk_consumer_executor as executor

RUNNER = "a" * 40

TARGET = "b" * 40

def environment(**extra):

    return {

        "CI_PROJECT_ID": "377",

        "SDK_PROJECT_ID": "396",

        "CI_PIPELINE_SOURCE": "pipeline",

        "CI_PIPELINE_ID": "10",

        "CI_COMMIT_TAG": executor.EXECUTOR_REF,

        "CI_COMMIT_SHA": RUNNER,

        "SDK_CONSUMER_MODE": "compatibility",

        "SDK_BUILD_IMAGE": "0",

        "EXPECTED_SRM_BRANCH": "develop",

        "EXPECTED_SRM_SHA": TARGET,

        "SDK_EXECUTOR_SHA": RUNNER,

        **extra,

    }

def metadata(runner=RUNNER, target=TARGET):

    return {

        "projects/377/pipelines/10": {

            "sha": runner,

            "ref": executor.EXECUTOR_REF,

            "source": "pipeline",

        },

        f"projects/377/repository/tags/{executor.EXECUTOR_REF}": {

            "commit": {"id": runner},

            "protected": True,

        },

        "projects/377/repository/branches/develop": {"commit": {"id": target}},

    }

@pytest.mark.parametrize(

    "field,value",

    [

        ("EXPECTED_SRM_BRANCH", "feature/green"),

        ("EXPECTED_SRM_SHA", "../source"),

        ("SDK_EXECUTOR_SHA", "c" * 40),

        ("SDK_CONSUMER_MODE", "adoption"),

        ("SDK_BUILD_IMAGE", "1"),

        ("CI_PROJECT_ID", "999"),

        ("SDK_PROJECT_ID", "999"),

        ("CI_PIPELINE_SOURCE", "web"),

        ("CI_COMMIT_TAG", "another-tag"),

    ],

)

def test_invalid_request_fails_before_any_server_or_git_operation(field, value):

    with pytest.raises(ValueError):

        executor.validate(

            environment(**{field: value}),

            RUNNER,

            get=lambda _: pytest.fail("No metadata access for an invalid request"),

        )

@pytest.mark.parametrize("fault", ["tag-source", "unprotected", "pipeline-ref", "target"])

def test_server_metadata_mismatch_cannot_qualify(fault):

    values = metadata()

    tag = values[f"projects/377/repository/tags/{executor.EXECUTOR_REF}"]

    if fault == "tag-source":

        tag["commit"]["id"] = "c" * 40

    if fault == "unprotected":

        tag["protected"] = False

    if fault == "pipeline-ref":

        values["projects/377/pipelines/10"]["ref"] = "main"

    if fault == "target":

        values["projects/377/repository/branches/develop"]["commit"]["id"] = "c" * 40

    with pytest.raises(ValueError):

        executor.validate(environment(), RUNNER, get=values.__getitem__)

@pytest.mark.parametrize("fault", ["dirty", "checkout", "head-moved"])

def test_changed_source_fails_before_consumer_installation(tmp_path, fault):

    calls, revisions, reads = [], iter([RUNNER, TARGET if fault != "checkout" else RUNNER]), []

    values = metadata()

    def call(args, **kwargs):

        calls.append(args)

        value = next(revisions) if args[1:3] == ["rev-parse", "HEAD"] else ""

        if fault == "dirty" and args[1] == "status":

            value = " M pyproject.toml"

        return subprocess.CompletedProcess(args, 0, stdout=value)

    def get(path):

        if path.endswith("branches/develop"):

            reads.append(path)

            if fault == "head-moved" and len(reads) > 1:

                return {"commit": {"id": "c" * 40}}

        return values[path]

    with pytest.raises(ValueError):

        executor.execute(

            root=tmp_path,

            environment=environment(),

            get=get,

            call=call,

            qualify=lambda **_: pytest.fail("No installation for a changed target"),

        )

    assert not (tmp_path / ".cache/sdk-consumer/qualification.json").exists()

    if fault == "dirty":

        assert not any(c[1] == "fetch" for c in calls)

def test_local_git_checkout_qualifies_target_and_records_executor_separately(tmp_path):

    origin, checkout = tmp_path / "origin", tmp_path / "checkout"

    def git(directory, *args):

        return subprocess.check_output(["git", "-C", str(directory), *args], text=True).strip()

    origin.mkdir()

    git(origin, "init", "--quiet", "--initial-branch=develop")

    git(origin, "config", "user.name", "Local Test")

    git(origin, "config", "user.email", "test@example.invalid")

    (origin / "source.txt").write_text("actual target")

    git(origin, "add", "source.txt")

    git(origin, "commit", "--quiet", "-m", "target")

    target = git(origin, "rev-parse", "HEAD")

    (origin / "source.txt").write_text("executor tree")

    git(origin, "commit", "--quiet", "-am", "runner")

    runner = git(origin, "rev-parse", "HEAD")

    git(origin, "tag", executor.EXECUTOR_REF)

    git(origin, "update-ref", "refs/heads/develop", target)

    subprocess.run(

        ["git", "clone", "--quiet", "--no-checkout", str(origin), str(checkout)], check=True

    )

    git(checkout, "checkout", "--quiet", executor.EXECUTOR_REF)

    calls = []

    def qualify(root, out):

        calls.append(git(root, "rev-parse", "HEAD"))

        assert (root / "source.txt").read_text() == "actual target"

        out.mkdir(parents=True)

        return {"source_sha": calls[-1], "status": "passed", "qualification_mode": "compatibility"}

    proof = executor.execute(

        root=checkout,

        environment=environment(

            CI_COMMIT_SHA=runner, SDK_EXECUTOR_SHA=runner, EXPECTED_SRM_SHA=target

        ),

        get=metadata(runner, target).__getitem__,

        qualify=qualify,

    )

    assert calls == [target] and target != runner

    assert proof["source_sha"] == target

    assert proof["executor"] == {"ref": executor.EXECUTOR_REF, "source_sha": runner}

    assert json.loads((checkout / ".cache/sdk-consumer/qualification.json").read_text()) == proof