test: qualify current release artifacts and report observed features

      allow_failure: false

      variables:

        OOP_LAB_PROFILE: ci/lab/profile.release-2.0.json

        OOP_LAB_REFRESH_CANDIDATE: "1"

    - if: '$CI_COMMIT_BRANCH == "test/platform-lab"'

      when: manual

      allow_failure: false

    - if: '$RUN_PLATFORM_LAB == "1" && $CI_COMMIT_REF_PROTECTED == "true"'

    - when: never

  before_script:

    - apk add --no-cache python3

  script:

    - sh ci/run_platform_lab.sh "${OOP_LAB_PROFILE:-ci/lab/profile.health.json}" .cache/platform-lab

  artifacts:

      - .cache/platform-lab/result.json

      - .cache/platform-lab/junit.xml

      - .cache/platform-lab/chart.json

      - .cache/platform-lab/candidate.json

      - .cache/platform-lab/feature-matrix.json

      - .cache/platform-lab/feature-matrix.md

      - .cache/platform-lab/wrapper-junit.xml

      - .cache/platform-lab/pods.txt

      - .cache/platform-lab/events.txt

"""Resolve current artifacts without silently substituting an older candidate."""

import base64

from copy import deepcopy

import hashlib

import io

import json

from pathlib import Path

import sys

from urllib.parse import unquote, urlparse, parse_qs

import zipfile

import pytest

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

sys.path.insert(0, str(ROOT / "ci"))

from resolve_candidate import resolve_candidate

def wheel(version, contents=b"runtime = 1\n"):

    stream = io.BytesIO()

    with zipfile.ZipFile(stream, "w") as archive:

        archive.writestr("openop_tfsdk/__init__.py", contents)

        archive.writestr("openop_tfsdk-x.dist-info/METADATA",

                         f"Name: openop-tfsdk\nVersion: {version}\n")

    return stream.getvalue()

class Registry:

    def __init__(self, fault=None):

        self.fault = fault

        self.names = {1: "oeg", 2: "fm", 3: "srm", 4: "sdk", 5: "helm"}

        self.heads = {name: str(i) * 40 for i, name in self.names.items()}

        self.paths = dict(zip(self.names.values(), (

            "oop/code/open-exposure-gateway", "oop/code/federation-manager",

            "oop/code/service-resource-manager", "oop/code/tf-sdk", "oop/deploy/helm")))

        self.locked = wheel("2.0.0rc4")

        self.latest = wheel("2.0.0rc5", b"runtime = 2\n" if fault == "different-sdk" else b"runtime = 1\n")

        self.reads = {}

    def __call__(self, url):

        path = unquote(urlparse(url).path.removeprefix("/rep/api/v4/"))

        query = parse_qs(urlparse(url).query)

        self.reads[path] = self.reads.get(path, 0) + 1

        if url.endswith("locked.whl"):

            return self.locked

        if "/artifacts/.cache/ci/dist/" in path:

            return b"corrupt" if self.fault == "checksum" else self.latest

        if "/artifacts/.cache/ci/manifest.json" in path:

            return json.dumps({"source_sha": "a" * 40 if self.fault == "wrong-source" else self.heads["sdk"],

                "artifacts": [{"name": "sdk.whl", "sha256": hashlib.sha256(self.latest).hexdigest()}]}).encode()

        if path.startswith("projects/oop/"):

            project = path.removeprefix("projects/")

            identity = next(i for i, name in self.names.items() if self.paths[name] == project)

            return json.dumps({"id": identity}).encode()

        identity = int(path.split("/")[1]);name = self.names[identity]

        if path.endswith("/repository/branches/test/current-consumer"):

            result = {"commit": {"id": "6" * 40}}

        elif path.endswith("/repository/branches/develop"):

            head = "a" * 40 if self.fault == "moving-head" and self.reads[path] > 1 else self.heads[name]

            result = {"commit": {"id": head}}

        elif path.endswith("/pipelines"):

            assert query["sha"] == [self.heads[name]]

            result = [] if self.fault == "missing-pipeline" else [

                {"id": 100 + identity, "sha": self.heads[name], "status": "success", "web_url": url}]

        elif path.endswith("/jobs"):

            result = [{"id": 200 + identity, "name": "build-package" if name == "sdk" else "build-and-push",

                       "status": "success"}]

        elif path.endswith("/bridges"):

            result = [{"name": "srm-consumer", "status": "success", "downstream_pipeline": {

                "id": 300, "sha": "6" * 40, "ref": "test/current-consumer", "status": "success", "web_url": url}}]

        elif path.endswith("/registry/repositories"):

            result = [{"id": identity, "path": self.paths[name],

                       "location": "labs.etsi.org:5050/" + self.paths[name]}]

        elif "/registry/repositories/" in path:

            result = {"digest": "sha256:" + str(identity) * 64}

        elif "/repository/files/" in path:

            assert query["ref"] == [self.heads["srm"]]

            body = ('[[package]]\nname = "openop-tfsdk"\nversion = "2.0.0rc4"\n'

                    'wheels = [{url = "https://labs.etsi.org/locked.whl", hash = "sha256:'

                    + hashlib.sha256(self.locked).hexdigest() + '"}]\n')

            result = {"content": base64.b64encode(body.encode()).decode()}

        elif "/repository/tags/" in path:

            result = {"commit": {"id": "a" * 40}}

        else:

            raise AssertionError(url)

        return json.dumps(result).encode()

def test_latest_image_refs_and_actual_locked_sdk_are_resolved_together(tmp_path):

    original = {"images": {}, "checkpoint": "terminated", "acceptance": {"sdk": {}}}

    unchanged = deepcopy(original)

    registry = Registry()

    profile, evidence, sdk_wheel = resolve_candidate(original, tmp_path, fetch=registry)

    assert original == unchanged

    assert profile["chart_revision"] == registry.heads["helm"]

    assert profile["images"]["srm"].endswith("@sha256:" + "3" * 64)

    assert profile["acceptance"]["sdk"]["version"] == "2.0.0rc4"

    assert evidence["sdk"]["latest_build"]["version"] == "2.0.0rc5"

    assert evidence["sdk"]["runtime_matches_current_source"] is True

    assert sdk_wheel.read_bytes() == registry.locked

@pytest.mark.parametrize("fault", ["missing-pipeline", "wrong-source", "checksum", "different-sdk", "moving-head"])

def test_stale_unverified_or_changed_candidates_are_rejected(tmp_path, fault):

    with pytest.raises((ValueError, RuntimeError)):

        resolve_candidate({"acceptance": {"sdk": {}}}, tmp_path, fetch=Registry(fault))

"""A feature matrix must never promote absent or broken proof to a pass."""

import json

from pathlib import Path

import sys

import xml.etree.ElementTree as ET

import pytest

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

from feature_report import write_report, finalize_report

from checkpoints import PHASES

def evidence(path, *, fault=None):

    (path / "health").mkdir()

    root = ET.Element("testsuite")

    for key, _, nodes in PHASES:

        for node in nodes:

            name = node.split("::", 1)[1]

            if fault == "missing" and key == "endpoints":

                continue

            case = ET.SubElement(root, "testcase", name=name)

            if key == "endpoints" and fault in ("failure", "skipped"):

                ET.SubElement(case, fault)

            if key == "endpoints" and fault == "duplicate":

                ET.SubElement(root, "testcase", name=name)

    if fault != "no-report":

        ET.ElementTree(root).write(path / "health/junit.xml")

    (path / "result.json").write_text(json.dumps({"status": "passed" if not fault else "failed",

        "cleanup": "failed" if fault == "cleanup" else "passed"}))

def test_matrix_reports_actual_features_and_keeps_unselected_scopes_unverified(tmp_path):

    evidence(tmp_path)

    report = write_report(tmp_path, {"checkpoint": "terminated"})

    assert report["lab_transaction_passed"] is True

    assert report["candidate_run_passed"] is False  # Outer cleanup is not verified yet.

    (tmp_path / "wrapper-junit.xml").write_text('<testsuite><testcase name="cleanup"/></testsuite>')

    assert finalize_report(tmp_path, 0) is True

    assert sum(x["status"] == "passed" for x in report["features"]) == 10

    assert all(x["status"] == "not_run" for x in report["features"] if x["feature"] in ("qod", "helm", "device-ti"))

    assert "no limits" in next(x['scope'] for x in report['features'] if x['feature'] == 'resources')

@pytest.mark.parametrize("fault", ["missing", "duplicate", "failure", "skipped", "no-report", "cleanup"])

def test_incomplete_or_failed_run_is_not_release_qualification(tmp_path, fault):

    evidence(tmp_path, fault=fault)

    report = write_report(tmp_path, {"checkpoint": "terminated"})

    assert report["candidate_run_passed"] is False

    endpoint = next(x for x in report["features"] if x["feature"] == "endpoints")

    if fault not in ("cleanup",):

        assert endpoint["status"] != "passed"

@pytest.mark.parametrize("exit_code,failed", [(1, False), (0, True)])

def test_outer_wrapper_failure_cannot_leave_a_green_matrix(tmp_path, exit_code, failed):

    evidence(tmp_path)

    write_report(tmp_path, {"checkpoint": "terminated"})

    root = ET.Element("testsuite")

    case = ET.SubElement(root, "testcase", name="cleanup")

    if failed:

        ET.SubElement(case, "failure")

    ET.ElementTree(root).write(tmp_path / "wrapper-junit.xml")

    assert finalize_report(tmp_path, exit_code) is False

    if a[1].endswith("/acceptance/."):

        sys.exit(int(os.environ["FAULT"] == "receipt-copy"))

    target.parent.mkdir(parents=True, exist_ok=True)

    if target.name == "feature-matrix.json":

        target.write_text(json.dumps({"lab_transaction_passed": True, "features": [], "job_url": ""}))

    else:

        target.write_text("{}" if target.suffix == ".json" else "<testsuite/>")

elif a[0] not in ("info", "stop", "cp"):

    sys.exit(2)

"""Describe only observed acceptance proofs, with their scope and source inputs."""

import argparse

import json

import os

from pathlib import Path

import sys

import xml.etree.ElementTree as ET

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

sys.path.insert(0, str(ROOT / "ansible/tests/r2_e2e"))

from checkpoints import PHASES, selected_phases  # noqa: E402

LIMITS = {

    "health": "Public OEG/FM HTTP health; other API behavior has separate checks.",

    "auth": "One operator; issuer, missing/invalid token and App Provider role checks.",

    "onboarding": "Supported application manifest read/list/delete; no complete optional-field round trip.",

    "zones": "Configured lab workload zones and public identities; no dynamic backend health claim.",

    "workload": "One application, one zone and one ready Deployment/Pod/NodePort Service.",

    "resources": "Container CPU and decimal-MB memory requests; no limits, storage or accelerators.",

    "instance": "Expected app/zone identity and ready state for the selected instance.",

    "endpoints": "One observed external HTTP NodePort; no Ingress, LoadBalancer or multi-interface claim.",

    "reachable": "Controller-to-application HTTP with Pod identity; no WAN or data-plane policy claim.",

    "terminated": "Public instance and owned Deployment/Pod/Service removal.",

}

SEPARATE = {

    "helm": "HELM application lifecycle needs its own selected packaging scenario.",

    "qod": "Real provider QoD lifecycle, profiles and notification configuration need a NEF/device lab.",

    "location": "Real provider location retrieval needs a known-device lab.",

    "device-ti": "Device TI and forwarding need the selected NEF/device/application path.",

    "multi-cluster": "Control-path isolation needs separate two-cluster execution evidence.",

    "federation": "Federation requires multiple operator platforms and agreed partners.",

    "updates": "Scaling, updates and recovery require their own lifecycle assertions.",

}

def write_report(output, profile):

    output = Path(output)

    result = json.loads((output / "result.json").read_text())

    report = output / "health/junit.xml"

    cases = ET.parse(report).findall(".//testcase") if report.is_file() else []

    checkpoint = profile.get("checkpoint", "health")

    selected = {key for key, _, _ in selected_phases(checkpoint)}

    phase_status = {row["id"]: row["status"] for row in result.get("steps", [])}

    rows = []

    for key, name, nodes in PHASES:

        expected = {node.split("::", 1)[1] for node in nodes}

        observed = [case for case in cases if case.get("name") in expected]

        status = "not_run"

        if key in selected and observed:

            complete = len(observed) == len(expected) and {x.get("name") for x in observed} == expected

            bad = any(case.find(tag) is not None for case in observed for tag in ("failure", "error", "skipped"))

            status = "passed" if complete and not bad else "failed"

        if key in selected and phase_status.get(key) == "failed":

            status = "failed"

        rows.append({"feature": key, "behavior": name, "status": status,

                     "scope": LIMITS[key], "tests": sorted(expected),

                     "observed_tests": len(observed)})

    rows += [{"feature": key, "behavior": key, "status": "not_run", "scope": scope,

              "tests": [], "observed_tests": 0} for key, scope in SEPARATE.items()]

    candidate = output / "candidate.json"

    matrix = {"schema_version": 1, "checkpoint": checkpoint,

        "lab_transaction_passed": (result.get("status") == "passed" and result.get("cleanup") == "passed"

            and all(row["status"] == "passed" for row in rows if row["feature"] in selected)),

        "candidate_run_passed": False, "wrapper_verified": False,

        "lab_cleanup": result.get("cleanup", "not_run"),

        "job_url": os.environ.get("CI_JOB_URL"), "pipeline_url": os.environ.get("CI_PIPELINE_URL"),

        "ansible_source": os.environ.get("OOP_LAB_SOURCE_SHA"),

        "chart_revision": profile.get("chart_revision"), "images": profile.get("images", {}),

        "artifact_selection": json.loads(candidate.read_text()) if candidate.is_file() else None,

        "features": rows}

    save_report(output, matrix)

    return matrix

def save_report(output, matrix):

    (output / "feature-matrix.json").write_text(json.dumps(matrix, indent=2) + "\n")

    lines = ["# Observed release acceptance", "",

        f"Candidate transaction passed: **{matrix['candidate_run_passed']}**. "

        "An observed phase pass does not override a failed transaction or cleanup.", "",

        "| Feature | Observed result | Verified scope / limit |", "|---|---|---|"]

    lines += [f"| {row['behavior']} | {row['status']} | {row['scope']} |" for row in matrix["features"]]

    if matrix["job_url"]:

        lines += ["", f"[Native job]({matrix['job_url']})"]

    lines += ["", "Exact source, image, chart and SDK identities are in feature-matrix.json and candidate.json.",

              "Provider integration and real-provider acceptance remain distinct evidence levels.", ""]

    (output / "feature-matrix.md").write_text("\n".join(lines))

def finalize_report(output, exit_code):

    output = Path(output)

    matrix = json.loads((output / "feature-matrix.json").read_text())

    wrapper = ET.parse(output / "wrapper-junit.xml")

    cases = wrapper.findall(".//testcase")

    cleanup_passed = len(cases) == 1 and not any(

        case.find(tag) is not None for case in cases for tag in ("failure", "error", "skipped"))

    matrix.update(wrapper_verified=True, wrapper_exit_code=exit_code, wrapper_cleanup_passed=cleanup_passed,

                  candidate_run_passed=exit_code == 0 and cleanup_passed and matrix["lab_transaction_passed"])

    save_report(output, matrix)

    return matrix["candidate_run_passed"]

if __name__ == "__main__":

    parser = argparse.ArgumentParser(description=__doc__)

    parser.add_argument("report_dir", type=Path)

    parser.add_argument("--wrapper-exit-code", type=int, required=True)

    args = parser.parse_args()

    sys.exit(0 if finalize_report(args.report_dir, args.wrapper_exit_code) else 1)