fix: configure OAI QoD profiles per control path

credentials. Keep runtime config out of normalized responses and persisted

topology. The deployment owns mounts; the SDK does not parse a secret URI.

See `tests/contract/bootstrap_example.yaml` for an explicit file-mode example.

## OAI QoD provider profiles

Configure the profile identifiers accepted by each OAI NEF in that control

path's runtime infrastructure. For example, when the NEF offers these names:

```yaml

control_paths:

  - ref: oai-nef-lab

    adapter_family: network

    adapter_type: oai

    adapter_contract_version: v1

    infrastructure:

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

      scs_as_id: openop

      notification_url: http://notifications.example.test/nef

      qos_profiles: [qos-e, qos2, qos3, qos4]

```

The existing bootstrap runtime accessor and SRM registry pass this configuration

to the SDK factory. The adapter snapshots the list for its own client. Profiles

are opaque provider names: they reach the NEF unchanged, with no inferred aliases

or QoS policy mapping. Configure the actual names and policies offered by that

NEF; the example does not assert a universal OAI default.

Omitting qos_profiles preserves qos-e/qos-s/qos-m/qos-l. An explicit null, empty,

duplicate or malformed list fails adapter construction with ValidationError.

Requests outside the configured list fail before HTTP. Other control paths keep

their own lists, and backend refusals remain typed execution failures. The

existing notification destination requirement still applies; configuring profiles

does not implement a receiver or establish data-plane QoS guarantees.

    def __init__(self, infrastructure: Optional[dict] = None):

        infrastructure = infrastructure or {}

        self._scope = ControlPathScope(infrastructure.get("control_path_ref"))

        self._client = self.client_class(

            base_url=infrastructure.get("base_url"), scs_as_id=infrastructure.get("scs_as_id")

        )

        self._client = self.client_class(**self._client_kwargs(infrastructure))

        self._notification_url = infrastructure.get("notification_url")

    def _client_kwargs(self, infrastructure: dict) -> dict:

        return {

            "base_url": infrastructure.get("base_url"),

            "scs_as_id": infrastructure.get("scs_as_id"),

        }

    def activate(self, request: CapabilityLifecycleRequest) -> CapabilityLifecycleResult:

        self._scope.check(request)

        self._require(request.capability_kind, SupportedAction.ACTIVATE)

    client_class = OaiClient

    location_identifiers = ("network_access_id", "ipv4")

    def _client_kwargs(self, infrastructure: dict) -> dict:

        options = super()._client_kwargs(infrastructure)

        if "qos_profiles" not in infrastructure:

            return options

        profiles = infrastructure["qos_profiles"]

        message = "qos_profiles must contain nonempty, unique, valid QoS profile names"

        if (

            not isinstance(profiles, (list, tuple))

            or not profiles

            or not all(isinstance(name, str) for name in profiles)

            or len(set(profiles)) != len(profiles)

        ):

            raise ValidationError(message)

        try:

            options["qos_profiles"] = tuple(schemas.QosProfileName(name).root for name in profiles)

        except pydantic.ValidationError:

            raise ValidationError(message) from None

        return options

    @staticmethod

    def capability_surface(capability_kind: str, config: Optional[dict] = None):

        return _surface(OaiClient, capability_kind)

#   - Giulio Carota (giulio.carota@eurecom.fr)

##

from datetime import datetime, timedelta, timezone

from typing import Optional, Sequence

import requests

)

log = logger.get_logger(__name__)

supportedQos = ["qos-e", "qos-s", "qos-m", "qos-l"]

supportedQos = ("qos-e", "qos-s", "qos-m", "qos-l")

class NetworkManager(BaseNetworkClient):

    capabilities = {"qod", "traffic_influence", "location_retrieval"}

    def __init__(self, base_url: str, scs_as_id: str = None):

    def __init__(

        self, base_url: str, scs_as_id: str = None, *, qos_profiles: Optional[Sequence[str]] = None

    ):

        try:

            super().__init__()

            self.base_url = base_url

            self.scs_as_id = scs_as_id

            self._qos_profiles = tuple(supportedQos if qos_profiles is None else qos_profiles)

            self.identity_service_url = f"{base_url.rstrip('/')}/3gpp-ueid/v1"

            log.info(

                f"Initialized OaiNefClient with base_url: {self.base_url} and scs_as_id: {self.scs_as_id}"

        raises:

            ValidationError: If the session information does not meet core-specific requirements.

        """

        if session_info.qosProfile.root not in supportedQos:

        if session_info.qosProfile.root not in self._qos_profiles:

            raise OaiValidationError(

                f"QoS profile {session_info.qosProfile} not supported by OAI, supported profiles are {supportedQos}"

                f"QoS profile {session_info.qosProfile} not supported by OAI, "

                f"supported profiles are {list(self._qos_profiles)}"

            )

        if session_info.device is None or session_info.device.ipv4Address is None:

"""Provider profile names are runtime configuration of one OAI control path."""

import json

from concurrent.futures import ThreadPoolExecutor

from threading import Barrier

from types import SimpleNamespace

from uuid import uuid4

import pytest

import requests

import yaml

from openop_tfsdk.contract.adapters import build_adapter

from openop_tfsdk.contract.errors import (

    DomainExecutionFailed,

    InvalidTargetRef,

    ValidationError,

)

from openop_tfsdk.contract.execution_models import LifecycleState

from openop_tfsdk.contract.topology import BootstrapTopology

from .test_network_qod import NOTIFY, _request

@pytest.fixture

def nef(monkeypatch):

    state = SimpleNamespace(calls=[], status=201, barrier=None)

    def request(method, url, headers=None, data=None):

        body = json.loads(data) if data else {}

        state.calls.append((url, body))

        if state.barrier is not None:

            state.barrier.wait(timeout=5)

        response = requests.Response()

        response.url = url

        response.status_code = state.status

        payload = {**body, "self": f"{url}/{uuid4()}"}

        if state.status != 201:

            payload = {"detail": "PRIVATE_NEF_BODY", "cause": "UNKNOWN_QOS_REFERENCE"}

        response._content = json.dumps(payload).encode()

        return response

    monkeypatch.setattr("openop_tfsdk.network.core.common.requests.request", request)

    return state

def adapter(ref="nef", **config):

    return build_adapter(

        "oai",

        {

            "base_url": f"http://{ref}",

            "scs_as_id": "af",

            "control_path_ref": ref,

            **NOTIFY,

            **config,

        },

    )

def request(profile, ref="nef"):

    result = _request(profile=profile)

    return result.model_copy(

        update={"target": result.target.model_copy(update={"control_path_ref": ref})}

    )

@pytest.mark.parametrize("profile", ["qos-e", "qos-s", "qos-m", "qos-l"])

def test_omitted_profiles_keep_existing_release_support(nef, profile):

    assert adapter().activate(request(profile)).state is LifecycleState.ACTIVE

    assert nef.calls[0][1]["qosReference"] == profile

@pytest.mark.parametrize("profile", ["qos-e", "qos2", "qos3", "qos4"])

def test_configured_provider_names_reach_nef_without_aliases(nef, profile):

    result = adapter(qos_profiles=["qos-e", "qos2", "qos3", "qos4"]).activate(request(profile))

    assert result.state is LifecycleState.ACTIVE

    assert nef.calls[0][1]["qosReference"] == profile

    assert nef.calls[0][1]["notificationDestination"] == NOTIFY["notification_url"]

@pytest.mark.parametrize("profile", ["qos2", "qos3", "qos4"])

def test_unconfigured_names_still_fail_before_http(nef, profile):

    with pytest.raises(ValidationError, match="not supported by OAI"):

        adapter().activate(request(profile))

    assert not nef.calls

@pytest.mark.parametrize("profile", ["qos-s", "qos-m", "qos-l", "unknown-profile"])

def test_explicit_configuration_replaces_default_support(nef, profile):

    with pytest.raises(ValidationError, match="not supported by OAI"):

        adapter(qos_profiles=["qos-e", "qos2"]).activate(request(profile))

    assert not nef.calls

@pytest.mark.parametrize(

    "profiles",

    [

        None,

        [],

        "qos2",

        {},

        True,

        [None],

        [1],

        [""],

        ["ab"],

        ["qos 2"],

        ["qos2", "qos2"],

        ["qos2", "qos3 "],

        ["q" * 257],

    ],

)

def test_invalid_profile_configuration_fails_before_client_construction(monkeypatch, profiles):

    clients = []

    monkeypatch.setattr(

        "openop_tfsdk.contract.adapters.network.OaiCapabilityAdapter.client_class",

        staticmethod(lambda **kwargs: clients.append(kwargs)),

    )

    with pytest.raises(ValidationError, match="qos_profiles"):

        adapter(qos_profiles=profiles)

    assert clients == []

def test_profile_configuration_is_snapshotted_and_does_not_change_defaults(nef):

    profiles = ["qos2"]

    configured = adapter(qos_profiles=profiles)

    profiles[:] = ["qos-injected"]

    assert configured.activate(request("qos2")).state is LifecycleState.ACTIVE

    with pytest.raises(ValidationError):

        configured.activate(request("qos-injected"))

    assert adapter().activate(request("qos-s")).state is LifecycleState.ACTIVE

    assert [body["qosReference"] for _, body in nef.calls] == ["qos2", "qos-s"]

def test_different_control_paths_validate_and_execute_concurrently_without_cross_talk(nef):

    paths = [("nef-a", "qos2"), ("nef-b", "qos-s")]

    adapters = {ref: adapter(ref, qos_profiles=[profile]) for ref, profile in paths}

    nef.barrier = Barrier(2)

    with ThreadPoolExecutor(max_workers=2) as pool:

        futures = [

            pool.submit(adapters[ref].activate, request(profile, ref)) for ref, profile in paths

        ]

        assert all(f.result(timeout=10).state is LifecycleState.ACTIVE for f in futures)

    nef.barrier = None

    assert {(url.split("/")[2], body["qosReference"]) for url, body in nef.calls} == set(paths)

    before = len(nef.calls)

    for ref, profile in [("nef-a", "qos-s"), ("nef-b", "qos2")]:

        with pytest.raises(ValidationError):

            adapters[ref].activate(request(profile, ref))

    assert len(nef.calls) == before

def test_profiles_do_not_bypass_the_existing_control_path_instance_check(nef):

    with pytest.raises(InvalidTargetRef):

        adapter("nef-a", qos_profiles=["qos2"]).activate(request("qos2", "nef-b"))

    assert not nef.calls

def test_bootstrap_runtime_configuration_reaches_the_existing_factory(tmp_path, nef):

    declaration = {

        "control_paths": [

            {

                "ref": "nef",

                "adapter_type": "oai",

                "infrastructure": {

                    "base_url": "http://nef",

                    "scs_as_id": "af",

                    **NOTIFY,

                    "qos_profiles": ["qos2"],

                },

            }

        ]

    }

    source = tmp_path / "bootstrap.yaml"

    source.write_text(yaml.safe_dump(declaration))

    topology = BootstrapTopology.from_file(source)

    entry = topology.get_control_path_configurations()[0]

    runtime = {**entry["infrastructure"], "control_path_ref": entry["ref"]}

    bound = build_adapter(entry["adapter_type"], runtime)

    assert bound.activate(request("qos2")).state is LifecycleState.ACTIVE

    assert nef.calls[0][1]["qosReference"] == "qos2"

def test_configured_profile_can_be_refused_by_nef_with_safe_details(nef):

    nef.status = 400

    with pytest.raises(DomainExecutionFailed) as caught:

        adapter(qos_profiles=["qos2"]).activate(request("qos2"))

    assert len(nef.calls) == 1

    assert "PRIVATE_NEF_BODY" not in str(caught.value) + str(caught.value.details)