test: run actual EURECOM network compatibility in the platform lab

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

    - when: never

# Actual provider compatibility, extending the same container application lab.

platform-lab-network:

  extends: platform-lab

  variables:

    OOP_LAB_NETWORK_PROVIDER: EURECOM

  rules:

    - if: '$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH || $CI_COMMIT_BRANCH == "test/platform-application-checkpoints" || $CI_MERGE_REQUEST_SOURCE_BRANCH_NAME == "test/platform-application-checkpoints"'

      variables:

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

        OOP_LAB_REFRESH_CANDIDATE: "1"

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

    - when: never

# Explicit live mode leaves all default component tests and deployment jobs alone.

sdk-live-acceptance:

  stage: test

)

EURECOM_QOS_PROFILES = ("qos-e", "qos2", "qos3", "qos4")

NETWORK_PHASES = (

    ("qod", "OEG QoD profiles, provider readback, refusals and cleanup", (

        *("test_deploy_to_ti.py::test_qod_profile_lifecycle["+profile+"]" for profile in EURECOM_QOS_PROFILES),

        "test_deploy_to_ti.py::test_qod_unsupported_profile_is_rejected",

        "test_deploy_to_ti.py::test_qod_backend_refusal_retains_diagnostics",

    )),

    ("location", "Known-device location and missing-device rejection", (

        "test_deploy_to_ti.py::test_known_device_returns_geographic_location",

        "test_deploy_to_ti.py::test_location_without_device_is_rejected",

    )),

    ("device-ti", "Device Traffic Influence uses the deployed application endpoint", (

        "test_deploy_to_ti.py::test_device_ti_uses_the_deployed_application_endpoint",

    )),

)

def selected_phases(checkpoint):

    if checkpoint == "network":

        return (*PHASES[:-1], *NETWORK_PHASES, PHASES[-1])

    if checkpoint == "helm-terminated":

        return tuple(row for row in PHASES if row[0] != "resources")

    names = [key for key, _, _ in PHASES]

}

CHECKPOINTS["helm-terminated"] = {node.split("::", 1)[1] for node in selected_nodes("helm-terminated")}

CHECKPOINTS["network"] = {node.split("::", 1)[1] for node in selected_nodes("network")}

{

  "dedicated_lab": true,

  "operator": "sdk_lab",

  "kube_context": "oop-sdk-lab",

  "platform_namespace": "oop-e2e-platform",

  "timeout_seconds": 180,

  "srm": {

    "deployment": "srmcontroller",

    "container": "srmcontroller",

    "image_digest": "sha256:REPLACE_WITH_REVIEWED_SRM_IMAGE_DIGEST"

  },

  "oeg": {

    "deployment": "oegcontroller",

    "container": "oegcontroller",

    "image_digest": "sha256:REPLACE_WITH_REVIEWED_OEG_IMAGE_DIGEST"

  },

  "sdk": {

    "revision": "REPLACE_WITH_REVIEWED_SDK_COMMIT",

    "version": "REPLACE_WITH_CANDIDATE_VERSION",

    "sha256": "REPLACE_WITH_INSTALLED_CONTENTS_FINGERPRINT"

  },

  "qod": {

    "single_writer_lab": true,

    "control_path_ref": "nef-oai",

    "base_url": "http://nef.example.test:8080",

    "subscriptions_url": "http://nef.example.test:8080/3gpp-as-session-with-qos/v1/af/subscriptions",

    "notification_url": "http://192.0.2.1:8080/nef-notifications",

    "nef_token_env": null,

    "profiles": ["qos-e", "qos2", "qos3", "qos4"],

    "unsupported_profile": "e2e-unsupported-profile",

    "backend_refusal_profile": null,

    "session_request": {

      "device": {"ipv4Address": {"publicAddress": "REPLACE_WITH_REAL_UE_IPV4", "publicPort": 12345}},

      "applicationServer": {"ipv4Address": "REPLACE_WITH_REAL_SERVER_IPV4"},

      "duration": 120

    }

  }

}

"""Read actual individual provider resources when its collection API is absent."""

import json

import subprocess

from urllib.parse import urlparse

def read_subscriptions(qod, http):

    observer = qod.get('observer')

    if not observer:

        response = http('GET', qod['subscriptions_url'])

        assert response.status_code == 200, 'Provider collection observation failed'

        return response.json()

    # Only the isolated lab's transparent ingress writes this metadata file.

    result = subprocess.run(['docker','exec',observer['container'],'cat','/var/log/nginx/observed.log'],

                            capture_output=True,text=True,check=True,timeout=20)

    prefix = urlparse(qod['subscriptions_url']).path

    locations = set()

    for line in result.stdout.splitlines():

        row = json.loads(line)

        if row['method']=='POST' and row['path']==prefix and row['status']==201:

            path = urlparse(row['location']).path

            assert path.startswith(prefix+'/') and path!=prefix+'/', 'Successful provider create omitted its Location'

            locations.add(path)

    subscriptions = []

    for path in sorted(locations):

        response = http('GET',qod['base_url']+path)

        assert response.status_code in (200,404), 'Provider individual observation failed'

        if response.status_code == 200:

            subscriptions.append(response.json())

    return subscriptions

"""Opt-in OEG QoD proof against an explicitly selected, real NEF lab."""

from copy import deepcopy

from datetime import datetime, timezone

from ipaddress import IPv4Address, IPv4Network

import json

import re

from urllib.parse import urlparse

from uuid import uuid4

from sdk_scenario import poll, preflight_components

def _url(value):

    if not isinstance(value, str):

        raise ValueError("Supply an explicit NEF URL")

    parsed = urlparse(value)

    if (parsed.scheme not in ("http", "https") or not parsed.hostname

            or parsed.username or parsed.password or parsed.query or parsed.fragment):

        raise ValueError("Use an explicit NEF URL without credentials or query parameters")

    return parsed

def validate_profile(profile):

    qod = profile["qod"]

    if qod.get("single_writer_lab") is not True:

        raise ValueError("QoD acceptance requires an isolated single-writer lab")

    for key in ("base_url", "subscriptions_url", "notification_url"):

        _url(qod[key])

    if not qod["subscriptions_url"].startswith(qod["base_url"].rstrip("/") + "/"):

        raise ValueError("Observe subscriptions on the configured NEF base URL")

    if not isinstance(qod.get("control_path_ref"), str) or not qod["control_path_ref"].strip():

        raise ValueError("Declare the OAI control_path_ref")

    profiles = qod["profiles"]

    if not isinstance(profiles, list) or not profiles or len(set(profiles)) != len(profiles):

        raise ValueError("Declare a nonempty, unique list of expected QoD profiles")

    values = [*profiles, qod["unsupported_profile"]]

    if qod.get("backend_refusal_profile") is not None:

        values.append(qod["backend_refusal_profile"])

    for value in values:

        if not isinstance(value, str) or not re.fullmatch(r"[a-zA-Z0-9_.-]{3,256}", value):

            raise ValueError("Profiles must be valid opaque QoD identifiers")

    if qod["unsupported_profile"] in profiles or qod.get("backend_refusal_profile") in profiles:

        raise ValueError("Refusal probes must differ from expected supported profiles")

    if qod.get("backend_refusal_profile") == qod["unsupported_profile"]:

        raise ValueError("Use distinct SDK and backend refusal profiles")

    body = qod["session_request"]

    if set(body) != {"device", "applicationServer", "duration"}:

        raise ValueError("Declare device/applicationServer/duration; the control path supplies notification URL")

    IPv4Address(body["device"]["ipv4Address"]["publicAddress"])

    IPv4Address(body["applicationServer"]["ipv4Address"])

    if type(body["duration"]) is not int or not 1 <= body["duration"] <= 3600:

        raise ValueError("Use a lab session duration between1 and3600 seconds")

    token_env = qod.get("nef_token_env")

    if token_env is not None and not re.fullmatch(r"[A-Z_][A-Z0-9_]*", token_env):

        raise ValueError("nef_token_env must name an environment variable")

BOOTSTRAP_PROBE = """

import json, sys, yaml

from srm.config import get_settings

expected = json.loads(sys.argv[1])

with open(get_settings().bootstrap_topology_path) as source:

    bootstrap = yaml.safe_load(source)

entries = [x for x in bootstrap['control_paths'] if x['ref'] == expected['control_path_ref']]

assert len(entries) == 1, 'Expected one configured QoD control path'

entry = entries[0]

assert entry['adapter_type'] == 'oai', 'QoD lab must select the OAI adapter'

for key in ('base_url', 'notification_url'):

    assert entry['infrastructure'][key] == expected[key], 'QoD backend configuration mismatch'

print(json.dumps({'bootstrap_matches': True}))

"""

def preflight(operator, profile):

    evidence = preflight_components(operator, profile)

    qod = profile["qod"]

    expected = {key: qod[key] for key in ("control_path_ref", "base_url", "notification_url")}

    observed = json.loads(operator.kubectl(

        "exec", evidence["components"]["srm"]["pod"], "--container", profile["srm"]["container"],

        "--", "python", "-c", BOOTSTRAP_PROBE, json.dumps(expected),

    ))

    assert observed == {"bootstrap_matches": True}

    return {**evidence, **observed}

def subscription_map(body):

    assert isinstance(body, list), "NEF collection must return a subscription list"

    result = {}

    for entry in body:

        assert isinstance(entry, dict) and isinstance(entry.get("self"), str), "Missing NEF subscription identity"

        identity = entry["self"].rstrip("/").rsplit("/", 1)[-1]

        assert identity and identity not in result, "Duplicate NEF subscription identity"

        result[identity] = entry

    return result

def verify_flow(entry, device_ip, server_ip):

    """Require the one OAI IPv4 host-to-host flow requested by this scenario."""

    flows = entry.get("flowInfo")

    assert isinstance(flows, list) and len(flows) == 1, "Expected one NEF QoD flow"

    descriptions = flows[0].get("flowDescriptions")

    assert isinstance(descriptions, list) and len(descriptions) == 1, "Expected one NEF flow description"

    match = re.fullmatch(

        r"permit out ip from ([0-9.]+)(?:/32)? to ([0-9.]+)(?:/32)?", descriptions[0]

    )

    assert match, "NEF flow must select exactly the requested IPv4 hosts"

    assert IPv4Address(match[1]) == IPv4Address(device_ip), "NEF flow device differs from request"

    assert IPv4Address(match[2]) == IPv4Address(server_ip), "NEF flow server differs from request"

def target_identity(body):

    """Ignore absent optional fields and normalize an IPv4 host's /32 spelling."""

    def compact(value):

        return {k: compact(v) for k, v in value.items() if v is not None} if isinstance(value, dict) else value

    result = {key: compact(body[key]) for key in ("device", "applicationServer")}

    address = IPv4Network(result["applicationServer"]["ipv4Address"])

    assert address.prefixlen == 32, "Public QoD server must select one host"

    result["applicationServer"]["ipv4Address"] = str(address.network_address)

    return result

def has_backend_bad_request(raw):

    """Extract only fixed diagnostic fields; never retain raw component logs."""

    for line in raw.splitlines():

        line = re.sub(r"\x1b\[[0-9;]*m", "", line)

        try:

            event = json.loads(line)

        except (ValueError, TypeError):

            if ("backend_call_failed" in line

                    and re.search(r"\bmethod[=:]\s*['\"]?create_qod_session\b", line)

                    and re.search(r"\bbackend_status[=:]\s*400\b", line)

                    and re.search(r"\breason[=:]\s*['\"]?bad_request\b", line)):

                return True

        else:

            if isinstance(event, dict) and all(event.get(k) == v for k, v in {

                "event": "backend_call_failed", "method": "create_qod_session",

                "backend_status": 400, "reason": "bad_request",

            }.items()):

                return True

    return False

class QodScenario:

    base = "/quality-on-demand/v1/sessions"

    def __init__(self, operator, profile, token, http, read_nef, evidence):

        self.operator, self.profile, self.token = operator, profile, token

        self.http, self.read_nef, self.evidence = http, read_nef, evidence

        self.session_id, self.before = None, None

        self.attempted = False

    def call(self, method, suffix="", **kwargs):

        return self.http(method, self.operator.oeg_url + self.base + suffix, self.token, **kwargs)

    def create(self, profile_name):

        self.before = subscription_map(self.read_nef())

        assert not self.before, "Use an empty, dedicated NEF subscription collection"

        self.started = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

        self.evidence.update(profile=profile_name, x_correlator=str(uuid4()))

        body = {**deepcopy(self.profile["qod"]["session_request"]), "qosProfile": profile_name}

        self.attempted = True

        response = self.call("POST", json=body, headers={"x-correlator": self.evidence["x_correlator"]})

        if response.status_code in (400, 401, 403, 404, 405, 422):

            self.attempted = False  # Explicit rejection is distinct from a lost reply.

        assert response.status_code == 201, f"QoD creation returned HTTP{response.status_code}"

        self.session_id = response.json().get("sessionId")

        assert self.session_id, "QoD creation omitted sessionId"

        self.evidence["session_id"] = self.session_id

    def wait_state(self, expected):

        def check():

            response = self.call("GET", "/" + self.session_id)

            assert response.status_code == 200, f"QoD GET returned HTTP{response.status_code}"

            body = response.json()

            assert body["sessionId"] == self.session_id

            assert body["qosProfile"] == self.evidence["profile"], "QoD profile identity changed"

            requested = target_identity(self.profile["qod"]["session_request"])

            assert target_identity(body) == requested, "Public QoD target differs from request"

            state = body["qosStatus"]

            assert state == "REQUESTED" or state == expected, f"QoD reached unexpected state{state}"

            return body if state == expected else None

        result = poll(check, self.profile["timeout_seconds"], "QoD " + expected)

        self.evidence["qos_status"] = result["qosStatus"]

    def verify_supported(self, profile_name):

        self.create(profile_name)

        self.wait_state("AVAILABLE")

        def observed():

            after = subscription_map(self.read_nef())

            new = set(after) - set(self.before)

            assert len(new) <= 1, "Concurrent NEF writes invalidate this isolated proof"

            return after[next(iter(new))] if new else None

        entry = poll(observed, self.profile["timeout_seconds"], "NEF QoD subscription")

        assert entry["qosReference"] == profile_name, "NEF profile differs from public request"

        device = self.profile["qod"]["session_request"]["device"]["ipv4Address"]

        assert entry["ueIpv4Addr"] == device.get("privateAddress", device["publicAddress"])

        verify_flow(entry, entry["ueIpv4Addr"],

                    self.profile["qod"]["session_request"]["applicationServer"]["ipv4Address"])

        assert entry["notificationDestination"] == self.profile["qod"]["notification_url"]

        self.evidence.update(nef_profile=entry["qosReference"], nef_subscription_observed=True)

    def verify_refused(self, profile_name, backend=False):

        self.create(profile_name)

        self.wait_state("UNAVAILABLE")

        assert set(subscription_map(self.read_nef())) == set(self.before), "Refused request left a NEF subscription"

        self.evidence["refusal"] = "backend" if backend else "unsupported_profile"

        if backend:

            logs = self.operator.kubectl("logs", "deploy/" + self.profile["srm"]["deployment"],

                                         "--container", self.profile["srm"]["container"],

                                         "--since-time=" + self.started)

            assert has_backend_bad_request(logs), "Missing SDK HTTP400/bad_request diagnostic"

            self.evidence["backend_diagnostic"] = {"backend_status": 400, "reason": "bad_request"}

    def cleanup(self):

        errors = []

        if self.session_id:

            try:

                response = self.call("DELETE", "/" + self.session_id)

                assert response.status_code in (204, 404), "QoD deletion failed"

                poll(lambda: self.call("GET", "/" + self.session_id).status_code == 404,

                     self.profile["timeout_seconds"], "QoD removal from OEG")

            except Exception as error:

                errors.append("OEG QoD cleanup: " + type(error).__name__)

        elif self.attempted:

            errors.append("QoD creation response lost; public session identity unavailable")

        if self.before is not None:

            try:

                poll(lambda: set(subscription_map(self.read_nef())) == set(self.before),

                     self.profile["timeout_seconds"], "NEF baseline restoration")

            except Exception as error:

                errors.append("NEF QoD cleanup: " + type(error).__name__)

        self.evidence["cleanup_verified"] = not errors

        return errors