fixed lint and format errors (017e4d81) · Commits · OOP / code / ai2

mcp_module/models/edge_application_models.py

+17 −14

Changes for mcp_module/models/edge_application_models.py: 17 added lines, 14 removed lines.
Original line number	Diff line number	Diff line
		from typing import Literal, Optional
		from typing import Literal

		from pydantic import BaseModel, ConfigDict, Field, field_validator

		@@ -87,29 +87,31 @@ class GpuPool(BaseModel):
		class ApplicationResources(BaseModel):
		model_config = ConfigDict(extra="forbid")

		cpuPool: Optional[CpuPool] = None
		gpuPool: Optional[GpuPool] = None
		cpuPool: CpuPool \| None = None
		gpuPool: GpuPool \| None = None


		class K8sPrimaryNetwork(BaseModel):
		model_config = ConfigDict(extra="forbid")

		provider: Optional[str] = Field(default=None, max_length=64)
		version: Optional[str] = Field(default=None, max_length=64)
		provider: str \| None = Field(default=None, max_length=64)
		version: str \| None = Field(default=None, max_length=64)


		class K8sAdditionalNetwork(BaseModel):
		model_config = ConfigDict(extra="forbid")

		name: Optional[str] = Field(default=None, max_length=64)
		interfaceType: Optional[Literal["netdevice", "vfio-pci", "interface"]] = None
		name: str \| None = Field(default=None, max_length=64)
		interfaceType: Literal["netdevice", "vfio-pci", "interface"] \| None = None


		class K8sNetworking(BaseModel):
		model_config = ConfigDict(extra="forbid")

		primaryNetwork: K8sPrimaryNetwork
		additionalNetworks: Optional[list[K8sAdditionalNetwork]] = Field(default=None, max_length=100)
		additionalNetworks: list[K8sAdditionalNetwork] \| None = Field(
		default=None, max_length=100
		)


		K8sAddon = Literal["monitoring", "ingress"]
		@@ -121,20 +123,21 @@ class KubernetesResources(BaseModel):
		infraKind: Literal["kubernetes"]
		applicationResources: ApplicationResources
		isStandalone: bool
		additionalStorage: Optional[str] = Field(default=None, max_length=32, pattern=r"^\d+(GB\|MB)$")
		version: Optional[str] = Field(default=None, max_length=64)
		networking: Optional[K8sNetworking] = None
		addons: Optional[list[K8sAddon]] = Field(default=None, max_length=2)
		additionalStorage: str \| None = Field(
		default=None, max_length=32, pattern=r"^\d+(GB\|MB)$"
		)
		version: str \| None = Field(default=None, max_length=64)
		networking: K8sNetworking \| None = None
		addons: list[K8sAddon] \| None = Field(default=None, max_length=2)

		@field_validator("addons")
		@classmethod
		def _addons_unique(cls, v: Optional[list[str]]) -> Optional[list[str]]:
		def _addons_unique(cls, v: list[str] \| None) -> list[str] \| None:
		if v is not None and len(set(v)) != len(v):
		raise ValueError("addons must be unique")
		return v



		class RegisterAppDefinitionInput(BaseModel):
		appId: str \| None = Field(
		default=None,

tests/mcp_module/test_authorization.py

0 → 100644

+32 −0

Changes for tests/mcp_module/test_authorization.py: 32 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import pytest
		from fastmcp.exceptions import ToolError

		from mcp_module.security import authorization


		@pytest.mark.parametrize(
		"header", ["Bearer user-token", "bearer user-token", "BEARER user-token "]
		)
		def test_forward_request_bearer_token(monkeypatch, header):
		monkeypatch.setattr(
		authorization,
		"get_http_headers",
		lambda: {"authorization": header, "cookie": "private"},
		)
		assert authorization.get_oeg_authorization_headers() == {
		"Authorization": "Bearer user-token"
		}


		@pytest.mark.parametrize(
		"header", [None, "", "Bearer", "Bearer ", "Basic user-token", "Bearer\tuser-token"]
		)
		def test_reject_missing_or_invalid_bearer_token(monkeypatch, header):
		monkeypatch.setenv("OEG_TOKEN", "must-not-fall-back-to-service-token")
		monkeypatch.setattr(
		authorization,
		"get_http_headers",
		lambda: {} if header is None else {"authorization": header},
		)
		with pytest.raises(ToolError, match="Bearer token required"):
		authorization.get_oeg_authorization_headers()

tests/mcp_module/test_input_models.py

0 → 100644

+43 −0

Changes for tests/mcp_module/test_input_models.py: 43 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import pytest
		from pydantic import ValidationError

		from mcp_module.models.location_models import RetrieveLocationInput
		from mcp_module.models.qod_models import (
		CreateQoDSessionOAIInput,
		DeviceInput,
		IPv4Address,
		)
		from mcp_module.models.traffic_influence_models import CreateTrafficInfluenceInput


		@pytest.mark.parametrize(
		"body", [{}, {"phoneNumber": "123"}, {"ipv6Address": "invalid"}]
		)
		def test_device_rejects_missing_or_malformed_identifiers(body):
		with pytest.raises(ValidationError):
		DeviceInput(**body)


		def test_ipv4_requires_port_or_private_address():
		with pytest.raises(ValidationError):
		IPv4Address(publicAddress="198.51.100.1")


		@pytest.mark.parametrize("body", [{"maxAge": -1}, {"maxSurface": 0}])
		def test_location_limits(body):
		with pytest.raises(ValidationError):
		RetrieveLocationInput(**body)


		def test_qod_rejects_zero_duration():
		with pytest.raises(ValidationError):
		CreateQoDSessionOAIInput(
		applicationServer={"ipv4Address": "198.51.100.1"},
		qosProfile="qos-e",
		duration=0,
		)


		def test_traffic_rejects_non_uuid_application():
		with pytest.raises(ValidationError):
		CreateTrafficInfluenceInput(apiConsumerId="consumer", appId="not-a-uuid")

tests/mcp_module/test_tool_requests.py

0 → 100644

+384 −0

Changes for tests/mcp_module/test_tool_requests.py: 384 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Wire-contract tests: real tools and models, mocked HTTP only."""

		import asyncio
		import json
		from dataclasses import dataclass, field

		import httpx
		import pytest
		from fastmcp.exceptions import ToolError

		from mcp_module.models import edge_application_models as em
		from mcp_module.models.location_models import RetrieveLocationInput
		from mcp_module.models.qod_models import CreateQoDSessionOAIInput, GetQoDSessionInput
		from mcp_module.models.traffic_influence_models import CreateTrafficInfluenceInput
		from mcp_module.security import authorization
		from mcp_module.tools import edge_application as edge
		from mcp_module.tools import location, qod
		from mcp_module.tools import traffic_influence as traffic

		APP = "11111111-1111-4111-8111-111111111111"
		EDGE = "/edge-application-management/v0.1alpha1"
		TI = "/traffic-influence/v0.10/traffic-influences"
		QOD = "/quality-on-demand/v1/sessions"
		RESOURCES = {
		"infraKind": "kubernetes",
		"applicationResources": {},
		"isStandalone": True,
		}
		MANIFEST = {
		"name": "demo_app",
		"appProvider": "demo_provider",
		"version": "1.0.0",
		"packageType": "CONTAINER",
		"appRepo": {"type": "PUBLICREPO", "imagePath": "nginx:latest"},
		"componentSpec": [],
		"requiredResources": RESOURCES,
		}
		SESSION = {
		"device": {"phoneNumber": "+306900000000"},
		"applicationServer": {"ipv4Address": "198.51.100.10"},
		"qosProfile": "qos-e",
		"duration": 60,
		"devicePorts": {"ranges": [{"from": 80, "to": 90}]},
		}


		@dataclass
		class Case:
		module: object
		name: str
		kwargs: dict
		method: str
		path: str
		body: dict \| None = None
		params: dict = field(default_factory=dict)
		status: int = 200
		reply: object = field(default_factory=dict)
		idempotency: bool = False


		def cases():
		instance = {
		"name": "demo_app",
		"appId": APP,
		"edgeCloudZoneId": "zone-1",
		"kubernetesClusterRef": "cluster-1",
		}
		deployment = {
		"appDeploymentName": "demo_app",
		"appId": APP,
		"edgeCloudZones": ["zone-1"],
		"kubernetesClusterRefs": ["cluster-1"],
		}
		return [
		Case(edge, "get_app_definitions", {}, "GET", EDGE + "/apps", reply=[]),
		Case(
		edge,
		"register_app_definition",
		{"inp": em.RegisterAppDefinitionInput(**MANIFEST)},
		"POST",
		EDGE + "/apps",
		MANIFEST,
		status=201,
		reply={"appId": APP},
		),
		Case(
		edge,
		"get_app_definition",
		{"inp": em.GetAppDefinitionInput(appId=APP)},
		"GET",
		EDGE + "/apps/" + APP,
		reply={"appManifest": MANIFEST},
		),
		Case(
		edge,
		"delete_app_definition",
		{"inp": em.DeleteAppDefinitionInput(appId=APP)},
		"DELETE",
		EDGE + "/apps/" + APP,
		status=204,
		),
		Case(
		edge,
		"get_edge_cloud_zones",
		{"region": "eu", "status": "active"},
		"GET",
		EDGE + "/edge-cloud-zones",
		params={"region": "eu", "status": "active"},
		reply=[],
		),
		Case(
		edge,
		"instantiate_app",
		{"inp": em.InstantiateAppInput(**instance, idempotencyKey="retry-1")},
		"POST",
		EDGE + "/appinstances",
		instance,
		status=202,
		idempotency=True,
		),
		Case(
		edge,
		"get_app_instances",
		{"appId": APP, "appInstanceId": "instance-1", "region": "eu"},
		"GET",
		EDGE + "/appinstances",
		params={"appId": APP, "appInstanceId": "instance-1", "region": "eu"},
		reply=[],
		),
		Case(
		edge,
		"delete_app_instance",
		{"inp": em.DeleteAppInstanceInput(appInstanceId="instance-1")},
		"DELETE",
		EDGE + "/appinstances/instance-1",
		status=202,
		),
		Case(
		edge,
		"create_app_deployment",
		{
		"inp": em.CreateAppDeploymentInput(
		**deployment, idempotencyKey="retry-1"
		)
		},
		"POST",
		EDGE + "/deployments",
		deployment,
		status=202,
		reply={"appDeploymentId": "deployment-1"},
		idempotency=True,
		),
		Case(
		edge,
		"get_app_deployments",
		{"appId": APP, "appDeploymentId": "deployment-1"},
		"GET",
		EDGE + "/deployments",
		params={"appId": APP, "appDeploymentId": "deployment-1"},
		reply=[],
		),
		Case(
		edge,
		"delete_app_deployment",
		{"inp": em.DeleteAppDeploymentInput(appDeploymentId="deployment-1")},
		"DELETE",
		EDGE + "/deployments/deployment-1",
		status=202,
		),
		Case(
		location,
		"retrieve_location",
		{
		"inp": RetrieveLocationInput(
		device=SESSION["device"], maxAge=0, maxSurface=100
		)
		},
		"POST",
		"/location-retrieval/v0.5/retrieve",
		{"device": SESSION["device"], "maxAge": 0, "maxSurface": 100},
		),
		Case(
		qod,
		"create_qod_session_oai",
		{"inp": CreateQoDSessionOAIInput(**SESSION)},
		"POST",
		QOD,
		SESSION,
		status=201,
		),
		Case(
		qod,
		"get_qod_session_oai",
		{"inp": GetQoDSessionInput(sessionId="session-1")},
		"GET",
		QOD + "/session-1",
		reply={**SESSION, "sessionId": "session-1", "qosStatus": "AVAILABLE"},
		),
		Case(
		qod,
		"delete_qod_session_oai",
		{"inp": GetQoDSessionInput(sessionId="session-1")},
		"DELETE",
		QOD + "/session-1",
		status=204,
		),
		Case(
		traffic,
		"create_traffic_influence",
		{
		"inp": CreateTrafficInfluenceInput(apiConsumerId="consumer", appId=APP),
		"idempotencyKey": "retry-1",
		},
		"POST",
		TI,
		{"apiConsumerId": "consumer", "appId": APP},
		status=201,
		idempotency=True,
		),
		Case(
		traffic,
		"get_traffic_influences",
		{"appId": APP},
		"GET",
		TI,
		params={"appId": APP},
		reply=[],
		),
		Case(
		traffic,
		"get_traffic_influence",
		{"trafficInfluenceID": "traffic-1"},
		"GET",
		TI + "/traffic-1",
		),
		Case(
		traffic,
		"delete_traffic_influence",
		{"trafficInfluenceID": "traffic-1"},
		"DELETE",
		TI + "/traffic-1",
		status=202,
		),
		]


		CASES = cases()
		REAL_CLIENT = httpx.AsyncClient


		@pytest.fixture
		def mock_oeg(monkeypatch):
		monkeypatch.setenv("OEG_SERVICE_URL", "http://oeg/")
		monkeypatch.setattr(
		authorization,
		"get_http_headers",
		lambda: {"authorization": "Bearer test-user-token"},
		)

		def install(handler):
		transport = httpx.MockTransport(handler)
		monkeypatch.setattr(
		httpx,
		"AsyncClient",
		lambda kwargs: REAL_CLIENT(transport=transport, kwargs),
		)

		return install


		@pytest.mark.parametrize("case", CASES, ids=lambda case: case.name)
		def test_tool_wire_contract(case, mock_oeg):
		seen = []

		def respond(request):
		seen.append(request)
		assert request.method == case.method
		assert str(request.url).split("?")[0] == "http://oeg" + case.path
		assert dict(request.url.params) == case.params
		assert request.headers["authorization"] == "Bearer test-user-token"
		assert request.headers["accept"] == "application/json"
		assert request.headers["x-correlator"] == "trace-1"
		if case.idempotency:
		assert request.headers["idempotency-key"] == "retry-1"
		else:
		assert "idempotency-key" not in request.headers
		if case.body is None:
		assert request.content == b""
		else:
		assert request.headers["content-type"] == "application/json"
		assert json.loads(request.content) == case.body
		if case.method == "DELETE":
		return httpx.Response(case.status)
		return httpx.Response(case.status, json=case.reply)

		mock_oeg(respond)
		result = asyncio.run(
		getattr(case.module, case.name)(**case.kwargs, xCorrelator="trace-1")
		)
		assert len(seen) == 1
		assert result is not None


		@pytest.mark.parametrize("case", CASES, ids=lambda case: case.name)
		def test_every_tool_exposes_camara_errors(case, mock_oeg):
		mock_oeg(
		lambda request: httpx.Response(
		403, json={"code": "PERMISSION_DENIED", "message": "Missing required scope"}
		)
		)
		with pytest.raises(
		ToolError, match="OEG HTTP 403: PERMISSION_DENIED: Missing required scope"
		):
		asyncio.run(getattr(case.module, case.name)(**case.kwargs))


		@pytest.mark.parametrize("case", CASES, ids=lambda case: case.name)
		def test_every_tool_requires_auth_before_http(case, mock_oeg, monkeypatch):
		monkeypatch.setattr(authorization, "get_http_headers", dict)

		def unexpected(request):
		pytest.fail("Unauthenticated tool attempted an HTTP request")

		mock_oeg(unexpected)
		with pytest.raises(ToolError, match="Bearer token required"):
		asyncio.run(getattr(case.module, case.name)(**case.kwargs))


		@pytest.mark.parametrize("case", CASES, ids=lambda case: case.name)
		def test_transport_failure_is_not_reported_as_success(case, mock_oeg):
		def unavailable(request):
		raise httpx.ConnectError("OEG unavailable", request=request)

		mock_oeg(unavailable)
		with pytest.raises(httpx.ConnectError, match="OEG unavailable"):
		asyncio.run(getattr(case.module, case.name)(**case.kwargs))


		@pytest.mark.parametrize(
		"module,helper",
		[
		(edge, "get_edge_service_url"),
		(qod, "get_qod_service_url"),
		(location, "get_location_service_url"),
		(traffic, "get_traffic_influence_service_url"),
		],
		)
		def test_missing_service_url_fails_clearly(module, helper, monkeypatch):
		monkeypatch.delenv("OEG_SERVICE_URL", raising=False)
		with pytest.raises(OSError, match="OEG_SERVICE_URL"):
		getattr(module, helper)()


		@pytest.mark.parametrize(
		"module,helper",
		[
		(edge, "get_edge_headers"),
		(qod, "get_qod_headers"),
		(location, "get_location_headers"),
		(traffic, "get_traffic_influence_headers"),
		],
		)
		def test_correlator_header_injection_is_rejected(module, helper, mock_oeg):
		with pytest.raises(ValueError, match="Invalid x-correlator"):
		getattr(module, helper)("trace\r\nInjected: value")


		def test_cases_cover_all_registered_tools():
		# Inspect registration without starting FastMCP or its HTTP app.
		import ast
		from pathlib import Path

		tree = ast.parse(
		(Path(__file__).resolve().parents[2] / "mcp_module/app.py").read_text()
		)
		registered = next(
		node.value.elts
		for node in tree.body
		if isinstance(node, ast.AnnAssign)
		and isinstance(node.target, ast.Name)
		and node.target.id == "SAFE_TOOLS"
		)
		assert {node.id for node in registered} == {case.name for case in CASES}
		assert len(CASES) == 19