test(fm): verify the real FM-SRM deploy loop (4284b939) · Commits · OOP / code / federation-manager

docs/running-srm.md

0 → 100644

+52 −0

Original line number	Diff line number	Diff line
		# Running SRM locally for the FM ↔ SRM loop test

		`tests/integration/test_fm_srm_loop.py` drives a real deploy: FM publishes
		`command.srm.service.deploy`, SRM consumes it and publishes
		`event.srm.operation.completed`, and FM finalises the federation transaction on that event.
		It skips unless SRM answers at `FM_SRM_URL` (default `http://127.0.0.1:8081`).

		## Which SRM

		`develop` is not enough: its DataBus router is a no-op. Use a branch with real command
		handling, currently `origin/feat/location-retrieval-api`.

		```bash
		cd ../service-resource-manager
		git worktree add ../srm-loop origin/feat/location-retrieval-api
		cd ../srm-loop && make install DEV=true
		```

		## Configuration

		SRM reads nested env vars (`__` delimiter). Point it at the same Postgres and NATS this
		repo's `docker-compose.dev.yaml` starts, using its own database:

		```bash
		docker compose -f ../federation-manager/docker-compose.dev.yaml up -d postgres nats keycloak
		docker exec fm-postgres psql -U fm -d postgres -c 'CREATE DATABASE srm_db'

		export APP_NAME="Service Resource Manager" APP_VERSION=1.5.0 APP_DESCRIPTION=SRM
		export POSTGRES_SETTINGS__URL="postgresql+asyncpg://fm:fm@localhost:5433/srm_db"
		export POSTGRES_SETTINGS__ECHO=false POSTGRES_SETTINGS__CREATE_SCHEMA_ON_STARTUP=true
		export NATS_SETTINGS__URL="nats://localhost:4222" NATS_SETTINGS__CONNECT_TIMEOUT=10
		export NATS_SETTINGS__MAX_RECONNECT_ATTEMPTS=3 NATS_SETTINGS__DRAIN_TIMEOUT=5

		uv run uvicorn srm.main:create_app --factory --host 127.0.0.1 --port 8081
		```

		The app is a factory (`create_app`), not a module-level `app`. Run it from its own
		directory: the Sunrise SDK it imports writes a `.log/` folder into the working directory.

		## What the test asserts, and what it does not

		It asserts the transaction reaches a terminal state, not that the deployment succeeded.
		With an empty catalog entry and a zone SRM has never heard of, SRM answers
		`failed_before_start` and the transaction ends `failed`. That still exercises every hop.
		Getting `completed` needs SRM-side data this repo does not own: deployment units on the
		service specification, and the target zone present in SRM's inventory.

		## Known transport gap

		SRM subscribes with core NATS, not a JetStream durable consumer. A command published while
		SRM is down is never seen, nothing is ever acked, and two SRM replicas would each run every
		deploy. Raise before relying on this in a cluster.

tests/integration/test_fm_srm_loop.py

0 → 100644

+285 −0

Original line number	Diff line number	Diff line
		import asyncio
		import os
		import socket
		import subprocess
		import sys
		import time
		from collections.abc import Iterator
		from contextlib import suppress
		from datetime import datetime, timezone
		from uuid import UUID, uuid4

		import httpx
		import pytest
		from sqlalchemy import delete, insert, select
		from sqlalchemy.ext.asyncio import AsyncSession

		from federation_manager.adapters.database.core import (
		build_engine,
		build_session_maker,
		create_schema,
		)
		from federation_manager.adapters.database.tables import (
		federation_agreements,
		federation_contexts,
		federation_transactions,
		partner_ops,
		)
		from federation_manager.adapters.databus.nats_adapter import NatsEventConsumer
		from federation_manager.contracts.srm import SUBJECT_OPERATION_COMPLETED

		pytestmark = pytest.mark.integration

		FM_DB = os.getenv("FM_POSTGRES_URL", "postgresql+asyncpg://fm:fm@localhost:5433/fm_db")
		ISSUER = os.getenv("FM_KEYCLOAK_ISSUER", "http://localhost:8090/realms/federation")
		NATS = os.getenv("FM_NATS_URL", "nats://localhost:4222")
		SRM = os.getenv("FM_SRM_URL", "http://127.0.0.1:8081")
		CLIENT_ID, CLIENT_SECRET = "originating-op-1", "dd7vNwFqjNpYwaghlEwMbw10g0klWDHb"
		CONTEXT_ID = "fed-ctx-srm-loop"
		APP_ID, APP_VERSION, FLAVOUR = "videoAnalytics", "1.2.0", "small"
		TERMINAL = {"completed", "partially_completed", "failed"}


		def _free_port() -> int:
		with socket.socket() as s:
		s.bind(("127.0.0.1", 0))
		port: int = s.getsockname()[1]
		return port


		def _require_srm() -> None:
		try:
		if httpx.get(f"{SRM}/healthz", timeout=2.0).status_code != 200:
		raise httpx.HTTPError("unhealthy")
		except httpx.HTTPError:
		pytest.skip(f"SRM not reachable at {SRM}; see docs/running-srm.md")


		def _create_specification() -> UUID:
		specification_id = uuid4()
		response = httpx.post(
		f"{SRM}/internal/catalog/service-specifications",
		json={
		"service_specification": {
		"id": str(specification_id),
		"app_provider_id": "partner-tenant",
		"ref": f"fed-loop-{specification_id.hex[:10]}",
		"name": "Federated loop test app",
		"version": APP_VERSION,
		},
		"service_deployment_units": [],
		"service_capability_requirements": [],
		},
		timeout=10.0,
		)
		response.raise_for_status()
		return specification_id


		def _token() -> str:
		response = httpx.post(
		f"{ISSUER}/protocol/openid-connect/token",
		data={
		"grant_type": "client_credentials",
		"client_id": CLIENT_ID,
		"client_secret": CLIENT_SECRET,
		"scope": "fed-mgmt",
		},
		timeout=10.0,
		)
		response.raise_for_status()
		token: str = response.json()["access_token"]
		return token


		async def _seed(specification_id: UUID, partner_id: UUID, zone_id: UUID) -> None:
		engine = build_engine(FM_DB)
		await create_schema(engine)
		async with build_session_maker(engine)() as session:
		# oauth2_client_id is unique: drop anything an interrupted run left behind
		stale = (
		(
		await session.execute(
		select(partner_ops.c.id).where(partner_ops.c.oauth2_client_id == CLIENT_ID)
		)
		)
		.scalars()
		.all()
		)
		for previous in stale:
		await _delete_partner(session, previous)
		await session.commit()

		agreement_id = uuid4()
		await session.execute(
		insert(partner_ops).values(
		id=partner_id,
		mcc_mnc=uuid4().hex[:10],
		oauth2_client_id=CLIENT_ID,
		base_url="http://127.0.0.1:9",
		status="active",
		)
		)
		await session.execute(
		insert(federation_agreements).values(
		id=agreement_id,
		partner_op_id=partner_id,
		permitted_api_types=["install-app"],
		permitted_zone_ids=[str(zone_id)],
		service_spec_mappings={
		"apps": [
		{
		"appId": APP_ID,
		"appVersion": APP_VERSION,
		"flavourId": FLAVOUR,
		"service_specification_id": str(specification_id),
		}
		],
		"api_families": {},
		},
		valid_from=datetime(2026, 1, 1, tzinfo=timezone.utc),
		status="active",
		)
		)
		await session.execute(
		insert(federation_contexts).values(
		id=uuid4(),
		partner_op_id=partner_id,
		agreement_id=agreement_id,
		direction="inbound",
		federation_context_id=CONTEXT_ID,
		status="available",
		created_at=datetime.now(timezone.utc),
		)
		)
		await session.commit()
		await engine.dispose()


		async def _delete_partner(session: AsyncSession, partner_id: UUID) -> None:
		await session.execute(
		delete(federation_transactions).where(federation_transactions.c.partner_op_id == partner_id)
		)
		await session.execute(
		delete(federation_contexts).where(federation_contexts.c.partner_op_id == partner_id)
		)
		await session.execute(
		delete(federation_agreements).where(federation_agreements.c.partner_op_id == partner_id)
		)
		await session.execute(delete(partner_ops).where(partner_ops.c.id == partner_id))


		async def _drop_consumer(durable: str) -> None:
		# durables outlive the process that made them and pile up on OOP_EVENTS
		consumer = NatsEventConsumer(NATS, durable)
		await consumer.connect()
		with suppress(Exception):
		await consumer.delete_durable(SUBJECT_OPERATION_COMPLETED)
		await consumer.close()


		async def _cleanup(partner_id: UUID) -> None:
		engine = build_engine(FM_DB)
		async with build_session_maker(engine)() as session:
		await _delete_partner(session, partner_id)
		await session.commit()
		await engine.dispose()


		async def _transaction_status(partner_id: UUID) -> tuple[str, dict[str, object] \| None]:
		engine = build_engine(FM_DB)
		async with build_session_maker(engine)() as session:
		row = (
		await session.execute(
		select(federation_transactions).where(
		federation_transactions.c.partner_op_id == partner_id
		)
		)
		).one()
		await engine.dispose()
		return row.status, row.response_summary


		@pytest.fixture
		def federated_stack() -> Iterator[tuple[int, UUID, UUID]]:
		_require_srm()
		specification_id = _create_specification()
		partner_id, zone_id = uuid4(), uuid4()
		asyncio.run(_seed(specification_id, partner_id, zone_id))

		port = _free_port()
		durable = f"fm-loop-{partner_id.hex[:8]}"
		process = subprocess.Popen(
		[
		sys.executable,
		"-m",
		"uvicorn",
		"federation_manager.main:app",
		"--host",
		"127.0.0.1",
		"--port",
		str(port),
		"--log-level",
		"warning",
		],
		env={
		**os.environ,
		"FM_POSTGRES_URL": FM_DB,
		"FM_KEYCLOAK_ISSUER": ISSUER,
		"FM_NATS_URL": NATS,
		"FM_EVENT_CONSUMER_DURABLE": durable,
		},
		stdout=subprocess.PIPE,
		stderr=subprocess.STDOUT,
		)
		deadline = time.monotonic() + 40
		try:
		while time.monotonic() < deadline:
		if process.poll() is not None:
		output = process.stdout.read().decode() if process.stdout else ""
		raise AssertionError(f"FM exited early:\n{output}")
		try:
		if httpx.get(f"http://127.0.0.1:{port}/healthz", timeout=1.0).status_code == 200:
		break
		except httpx.HTTPError:
		time.sleep(0.3)
		else:
		raise AssertionError("FM did not become ready")
		yield port, partner_id, zone_id
		finally:
		process.kill()
		process.wait(timeout=10)
		asyncio.run(_cleanup(partner_id))
		asyncio.run(_drop_consumer(durable))


		def test_install_app_reaches_srm_and_the_completion_comes_back(
		federated_stack: tuple[int, UUID, UUID],
		) -> None:
		port, partner_id, zone_id = federated_stack

		accepted = httpx.post(
		f"http://127.0.0.1:{port}/operatorplatform/federation/v1/{CONTEXT_ID}/application/lcm",
		headers={"Authorization": f"Bearer {_token()}", "Idempotency-Key": uuid4().hex},
		json={
		"appId": APP_ID,
		"appVersion": APP_VERSION,
		"appProviderId": "partnerProvider",
		"zoneInfo": {"zoneId": str(zone_id), "flavourId": FLAVOUR},
		"appInstCallbackLink": "https://partner.example/instances/callback",
		},
		timeout=10.0,
		)

		assert accepted.status_code == 202
		assert accepted.json()["zoneId"] == str(zone_id)

		# SRM consumes the command and publishes event.srm.operation.completed; FM finalises on it.
		deadline = time.monotonic() + 25
		status, summary = asyncio.run(_transaction_status(partner_id))
		while status not in TERMINAL and time.monotonic() < deadline:
		time.sleep(0.5)
		status, summary = asyncio.run(_transaction_status(partner_id))

		assert status in TERMINAL, f"transaction never finalised, still {status}"
		assert summary is not None

tests/integration/test_two_stack_federation.py

+15 −0

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ import subprocess
		import sys
		import time
		from collections.abc import Iterator
		from contextlib import suppress
		from dataclasses import dataclass
		from pathlib import Path
		from uuid import UUID, uuid4
		@@ -23,12 +24,15 @@ from federation_manager.adapters.database.federation_context_repo import (
		PostgresFederationContextRepo,
		)
		from federation_manager.adapters.database.tables import federation_contexts, partner_ops
		from federation_manager.adapters.databus.nats_adapter import NatsEventConsumer
		from federation_manager.contracts.srm import SUBJECT_OPERATION_COMPLETED
		from federation_manager.domain.models import FederationContext

		pytestmark = pytest.mark.integration

		PG_ROOT = os.getenv("FM_POSTGRES_ROOT", "postgresql+asyncpg://fm:fm@localhost:5433")
		DB_A, DB_B = "fm_db", "fm_db_b"
		NATS = os.getenv("FM_NATS_URL", "nats://localhost:4222")
		ISSUER = os.getenv("FM_KEYCLOAK_ISSUER", "http://localhost:8090/realms/federation")
		TOKEN_ENDPOINT = f"{ISSUER}/protocol/openid-connect/token"
		CLIENT_A, SECRET_A = "originating-op-1", "dd7vNwFqjNpYwaghlEwMbw10g0klWDHb"
		@@ -110,6 +114,7 @@ def _fm_env(database: str, port: int, federation_id: str, **extra: str) -> dict[
		"FM_MNCS": '["07"]',
		"FM_PARTNER_STATUS_LINK": f"http://127.0.0.1:{port}/operatorplatform/federation/v1/partner-status",
		"FM_ALLOW_INSECURE_PARTNER_ENDPOINTS": "true",
		"FM_EVENT_CONSUMER_DURABLE": f"fm-event-worker-{federation_id}",
		**extra,
		}

		@@ -193,6 +198,16 @@ def stacks(tmp_path: Path) -> Iterator[Stacks]:
		process.kill()
		process.wait(timeout=10)

		async def drop_consumers() -> None:
		for federation_id in ("op-a", "op-b"):
		consumer = NatsEventConsumer(NATS, f"fm-event-worker-{federation_id}")
		await consumer.connect()
		with suppress(Exception):
		await consumer.delete_durable(SUBJECT_OPERATION_COMPLETED)
		await consumer.close()

		asyncio.run(drop_consumers())

		async def cleanup() -> None:
		for database, partner_id in ((DB_A, partner_b_id), (DB_B, partner_a_id)):
		engine = build_engine(f"{PG_ROOT}/{database}")