test(eam): register app befor /appinstances-dependent tests (392cbcde) · Commits · OOP / code / open-exposure-gateway

pyproject.toml

+1 −0

Original line number	Diff line number	Diff line
		@@ -58,6 +58,7 @@ python_version = "3.12"
		cache_dir = ".cache/mypy"
		strict = true
		ignore_missing_imports = true
		mypy_path = "src"

		[[tool.mypy.overrides]]
		module = "tests.conformance.*"

tests/unit/conftest.py

+46 −1

Original line number	Diff line number	Diff line
		import json as _json
		from collections.abc import Generator
		from collections.abc import Callable, Generator
		from typing import Any

		import pytest
		from fastapi.testclient import TestClient

		from open_exposure_gateway.api.camara.edge_application_management.vwip.router import (
		BASE_PATH as EAM_BASE,
		)
		from open_exposure_gateway.application.services.edge_application_management_service import (
		EdgeApplicationManagementService,
		)
		@@ -89,6 +92,48 @@ def qod_service(fake_srm: FakeSRMClient) -> QualityOnDemandService:
		return QualityOnDemandService(srm_client=fake_srm)


		@pytest.fixture()
		def register_app(api_client: TestClient) -> Callable[[Any], None]:
		"""Registers an app via a real POST /apps call, for tests that exercise
		/appinstances or /deployments and need appId -> app_registrations to
		resolve. Returns a callable so each test picks its own appId."""

		def _register(app_id: Any) -> None:
		response = api_client.post(
		f"{EAM_BASE}/apps",
		json={
		"appId": str(app_id),
		"name": "myvideoapp",
		"appProvider": "acme_provider",
		"version": "1.0.0",
		"packageType": "HELM",
		"appRepo": {
		"type": "PUBLICREPO",
		"imagePath": "oci://registry.example.com/app:1.0",
		},
		"requiredResources": {
		"infraKind": "kubernetes",
		"applicationResources": {
		"cpuPool": {
		"numCPU": 2,
		"memory": 4096,
		"topology": {
		"minNumberOfNodes": 1,
		"minNodeCpu": 1,
		"minNodeMemory": 1024,
		},
		}
		},
		"isStandalone": False,
		},
		"componentSpec": [],
		},
		)
		assert response.status_code == 201

		return _register


		@pytest.fixture()
		def api_client(
		eam_service: EdgeApplicationManagementService,

tests/unit/test_eam_contract.py

+9 −0

Original line number	Diff line number	Diff line
		@@ -6,6 +6,7 @@ code (or the fakes, where noted) until the test passes; do not weaken the test.

		"""

		from collections.abc import Callable
		from datetime import datetime
		from typing import Any
		from unittest.mock import AsyncMock, MagicMock
		@@ -60,6 +61,10 @@ def _assert_envelope(payload: dict[str, Any]) -> None:


		class TestCommandEnvelope:
		@pytest.fixture(autouse=True)
		def _registered_app(self, register_app: Callable[[Any], None]) -> None:
		register_app(APP_ID)

		def test_deploy_and_terminate_carry_the_full_envelope(
		self, api_client: TestClient, fake_bus: FakeDataBus, live_srm: FakeSRMClient
		) -> None:
		@@ -74,6 +79,10 @@ class TestCommandEnvelope:


		class TestTerminateCommandConformance:
		@pytest.fixture(autouse=True)
		def _registered_app(self, register_app: Callable[[Any], None]) -> None:
		register_app(APP_ID)

		def test_terminate_carries_required_terminate_object(
		self, api_client: TestClient, fake_bus: FakeDataBus, live_srm: FakeSRMClient
		) -> None:

tests/unit/test_eam_flows.py

+5 −13

Original line number	Diff line number	Diff line
		@@ -9,6 +9,7 @@ Tests in this module assert intended behavior. A failing test here means a
		real hole in the flow, not a broken test.
		"""

		from collections.abc import Callable
		from typing import Any
		from unittest.mock import AsyncMock
		from uuid import uuid4
		@@ -65,19 +66,10 @@ def _manifest_body(num_cpu: float \| int) -> dict[str, Any]:
		}


		def _register_app(api_client: TestClient, app_id: Any = APP_ID) -> None:
		"""Registers app_id via the real POST /apps flow, so create_app_instance's
		appId -> app_registrations resolution finds it."""
		response = api_client.post(
		f"{EAM_BASE}/apps", json={**_manifest_body(num_cpu=2), "appId": str(app_id)}
		)
		assert response.status_code == 201


		class TestCreateAppInstanceFlow:
		@pytest.fixture(autouse=True)
		def _registered_app(self, api_client: TestClient) -> None:
		_register_app(api_client)
		def _registered_app(self, register_app: Callable[[Any], None]) -> None:
		register_app(APP_ID)

		def test_returns_202_with_location_header(
		self, api_client: TestClient, live_srm: FakeSRMClient
		@@ -174,8 +166,8 @@ class TestCreateAppInstanceUnregisteredAppFlow:

		class TestDeleteAppInstanceFlow:
		@pytest.fixture(autouse=True)
		def _registered_app(self, api_client: TestClient) -> None:
		_register_app(api_client)
		def _registered_app(self, register_app: Callable[[Any], None]) -> None:
		register_app(APP_ID)

		def test_returns_202_and_srm_terminates_the_instance(
		self, api_client: TestClient, fake_bus: FakeDataBus, live_srm: FakeSRMClient

Admin message