Commit 6e7c6289 authored by Konstantinos Togias's avatar Konstantinos Togias
Browse files

test: verify public request failures and API boundaries

parent d55c253b
Loading
Loading
Loading
Loading
Loading
+9 −0
Changes for ansible/tests/r2_e2e/checkpoints.py: 9 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -9,11 +9,20 @@ PHASES = (
        "test_platform.py::test_keycloak_issuer_is_the_public_url[federation]",
        "test_auth.py::test_oeg_refuses_calls_without_a_valid_token",
        "test_auth.py::test_oeg_keeps_app_apis_to_app_providers",
        "test_auth.py::test_oeg_keeps_federation_to_operators",
        "test_exposure.py::test_fm_internal_api_is_not_public",
    )),
    ("onboarding", "Cloud application catalogue onboarding", (
        "test_apps.py::test_app_provider_registers_reads_and_deletes_an_app",
        "test_apps.py::test_app_provider_lists_its_apps",
    )),
    ("failures", "Invalid and conflicting public requests are rejected", (
        "test_release_failures.py::test_invalid_application_is_not_registered[missing-name]",
        "test_release_failures.py::test_invalid_application_is_not_registered[unsupported-package]",
        "test_release_failures.py::test_duplicate_onboarding_preserves_the_registered_application",
        "test_release_failures.py::test_unknown_application_cannot_be_instantiated",
        "test_release_failures.py::test_unknown_instance_cannot_be_terminated",
    )),
    ("zones", "Configured workload zones exposed by OEG", (
        "test_deploy_to_ti.py::test_cloud_zones_match_configured_workload_targets",
    )),
+8 −3
Changes for ansible/tests/r2_e2e/conftest.py: 8 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -44,7 +44,7 @@ def pytest_configure(config):
        image = config.getoption("e2e_app_image") or ""
        if not re.fullmatch(r"[^\s@]+@sha256:[0-9a-f]{64}", image):
            raise pytest.UsageError("--e2e-suite requires --e2e-app-image pinned by sha256 digest")
    if config.getoption("sdk_checkpoint") or platform_checkpoint not in (None, "health", "auth", "onboarding"):
    if config.getoption("sdk_checkpoint") or platform_checkpoint not in (None, "health", "auth", "onboarding", "failures"):
        from sdk_scenario import load_profile

        if not config.getoption("sdk_profile"):
@@ -305,6 +305,11 @@ EAM_SCOPES = (
)


INSTANCE_SCOPES = tuple(
    f"edge-application-management:instances:{action}" for action in ("write", "read", "delete")
)


@pytest.fixture(scope="session")
def callers(operators):
    """Per operator, a token each for an operator and two App Providers of different tenants."""
@@ -316,8 +321,8 @@ def callers(operators):
            out[op.host] = {}
            for name, role, oeg_roles, scopes in (
                ("operator", "operator", (), ()),
                ("provider", "app-provider", ("edge-application-management",), EAM_SCOPES),
                ("other-provider", "app-provider", ("edge-application-management",), EAM_SCOPES),
                ("provider", "app-provider", ("edge-application-management",), EAM_SCOPES + INSTANCE_SCOPES),
                ("other-provider", "app-provider", ("edge-application-management",), EAM_SCOPES + INSTANCE_SCOPES),
            ):
                client_id = f"e2e-{name}-{run}"
                secret = admin.make_client(client_id, role, oeg_roles, scopes)
+1 −4
Changes for ansible/tests/r2_e2e/test_deploy_to_ti.py: 1 added line, 4 removed lines.
Original line number Diff line number Diff line
@@ -7,12 +7,9 @@ from uuid import uuid4

import pytest

from conftest import EAM_SCOPES, KeycloakAdmin, Token, http
from conftest import EAM_SCOPES, INSTANCE_SCOPES, KeycloakAdmin, Token, http
from sdk_scenario import Scenario, check_endpoint_metadata, check_reachability, check_requested_resources, preflight

INSTANCE_SCOPES = tuple(
    f"edge-application-management:instances:{action}" for action in ("write", "read", "delete")
)
TI_SCOPES = tuple(
    f"traffic-influence:traffic-influences:{action}" for action in ("write", "read", "delete")
)
+69 −0
Changes for ansible/tests/r2_e2e/test_release_failures.py: 69 added lines, 0 removed lines.
Original line number Diff line number Diff line
"""Likely public errors, with absence/readback assertions and bounded cleanup."""

from copy import deepcopy
from uuid import uuid4

import pytest

from app_manifest import container_manifest
from conftest import http
from test_apps import APPS, _delete_app, registered_app  # noqa: F401

INSTANCES = "/edge-application-management/v0.1alpha1/app-instances"
pytestmark = pytest.mark.platform_acceptance


@pytest.mark.parametrize("fault,status", [("missing-name", 422), ("unsupported-package", 501)],
                         ids=["missing-name", "unsupported-package"])
def test_invalid_application_is_not_registered(operators, callers, fault, status):
    for op in operators:
        manifest = container_manifest(op.application_image)
        if fault == "missing-name":
            del manifest["name"]
        else:
            manifest["packageType"] = "QCOW2"
        token = callers[op.host]["provider"]
        url = op.oeg_url + APPS + "/" + manifest["appId"]
        try:
            response = http("POST", op.oeg_url + APPS, token, json=manifest)
            assert response.status_code == status, f"{fault}: HTTP {response.status_code}"
            response = http("GET", url, token)
            assert response.status_code == 404, "Rejected application remains readable"
            response = http("GET", op.oeg_url + APPS, token)
            assert response.status_code == 200
            assert manifest["appId"] not in [item["appId"] for item in response.json()]
        finally:
            _delete_app(op, url, token)


def test_duplicate_onboarding_preserves_the_registered_application(operators, callers, registered_app):
    for op in operators:
        url, original = registered_app[op.host]
        duplicate = deepcopy(original)
        duplicate["name"] = original["name"] + "_changed"
        token = callers[op.host]["provider"]
        response = http("POST", op.oeg_url + APPS, token, json=duplicate)
        assert response.status_code == 409, f"Duplicate: HTTP {response.status_code}"
        response = http("GET", url, token)
        assert response.status_code == 200
        assert response.json()["appManifest"]["name"] == original["name"], "Conflict replaced the existing app"


def test_unknown_application_cannot_be_instantiated(operators, callers):
    for op in operators:
        app_id = str(uuid4())
        token = callers[op.host]["provider"]
        response = http("POST", op.oeg_url + INSTANCES, token, json={
            "name": "e2e_unknown_" + uuid4().hex[:8], "appId": app_id,
            "edgeCloudZoneId": str(uuid4()),
        })
        assert response.status_code == 400, f"Unknown app: HTTP {response.status_code}"
        response = http("GET", op.oeg_url + INSTANCES, token, params={"appId": app_id})
        assert response.status_code == 200 and response.json() == [], "Rejected request created an instance"


def test_unknown_instance_cannot_be_terminated(operators, callers):
    for op in operators:
        token = callers[op.host]["provider"]
        response = http("DELETE", op.oeg_url + INSTANCES + "/" + str(uuid4()), token)
        assert response.status_code == 404, f"Unknown instance: HTTP {response.status_code}"
+1 −1
Changes for ansible/tests/unit/test_feature_report.py: 1 added line, 1 removed line.
Original line number Diff line number Diff line
@@ -38,7 +38,7 @@ def test_matrix_reports_actual_features_and_keeps_unselected_scopes_unverified(t
    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 sum(x["status"] == "passed" for x in report["features"]) == 11
    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')

Loading