"""Read-only production deployment doctor regression tests.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import json from pathlib import Path import tempfile import unittest from alembic import command from alembic.config import Config from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from sqlalchemy import create_engine from sqlalchemy.orm import Session from licensing_shared.catalog import load_builtin_catalog from licensing_server.app.config import ( NotificationWorkerSettings, ServerSettings, SubscriptionWorkerSettings, ) from licensing_server.app.deployment_readiness import inspect_deployment_readiness from licensing_server.app.models import ( CatalogRelease, OutboxEvent, SigningKeyMetadata, User, ) from licensing_server.app.security import Ed25519SnapshotSigner TEST_NOW = datetime(2026, 8, 15, 22, 0, tzinfo=timezone.utc) REPOSITORY_ROOT = Path(__file__).resolve().parents[2] ALEMBIC_CONFIGURATION_PATH = REPOSITORY_ROOT / "licensing_server/alembic.ini" class _FailingSigner: key_id = "license.test.deployment" @staticmethod def public_key_bytes() -> bytes: return b"x" * 32 @staticmethod def sign_payload(_payload): raise RuntimeError("injected signer unavailable") def _migration_config(database_url: str) -> Config: if not isinstance(database_url, str) or not database_url: raise ValueError("database_url must be a non-empty string") configuration = Config(str(ALEMBIC_CONFIGURATION_PATH)) configuration.set_main_option("sqlalchemy.url", database_url) return configuration def _settings(database_url: str) -> ServerSettings: return ServerSettings( database_url=database_url, environment_name="development", signing_key_id="license.test.deployment", signing_private_key_path=Path("unused-development-key"), serial_pepper=b"s" * 32, fingerprint_pepper=b"f" * 32, oidc_issuer="https://identity.example.test", oidc_audience="licensing-api", oidc_jwks_url="https://identity.example.test/jwks.json", oidc_authorization_endpoint=( "https://identity.example.test/oauth2/authorize" ), oidc_token_endpoint="https://identity.example.test/oauth2/token", oidc_native_client_id="apolon-native-test", allow_development_signer=True, ) class DeploymentReadinessTests(unittest.TestCase): def setUp(self) -> None: self.temporary_directory = tempfile.TemporaryDirectory() database_path = Path(self.temporary_directory.name) / "readiness.db" self.database_url = f"sqlite+pysqlite:///{database_path.as_posix()}" self.configuration = _migration_config(self.database_url) command.upgrade(self.configuration, "head") self.engine = create_engine(self.database_url) self.settings = _settings(self.database_url) self.subscription_settings = SubscriptionWorkerSettings( database_url=self.database_url, environment_name="development", ) self.notification_settings = NotificationWorkerSettings( database_url=self.database_url, environment_name="development", ) self.signer = Ed25519SnapshotSigner( "license.test.deployment", Ed25519PrivateKey.generate(), ) with Session(self.engine) as session, session.begin(): catalog = load_builtin_catalog() session.add_all( ( User( id="user.test.deployment_admin", external_issuer="https://identity.example.test", external_subject="deployment-admin", verified_email=None, status="active", is_server_admin=True, ), CatalogRelease( id="catalog_release.test.deployment", product_id=catalog.product_id, revision=catalog.revision, document_json=catalog.to_mapping(), document_digest=bytes.fromhex(catalog.sha256()), status="active", staged_by_user_id="user.test.deployment_admin", staged_at=TEST_NOW - timedelta(days=1), published_by_user_id="user.test.deployment_admin", published_at=TEST_NOW - timedelta(days=1), retired_at=None, ), SigningKeyMetadata( id="signing_key.test.deployment", key_id=self.signer.key_id, purpose="license_snapshot", public_key_bytes=self.signer.public_key_bytes(), status="active", not_before=TEST_NOW - timedelta(days=1), expires_at=None, activated_at=TEST_NOW - timedelta(days=1), retired_at=None, ), ) ) def tearDown(self) -> None: self.engine.dispose() self.temporary_directory.cleanup() def _report(self, **overrides): return inspect_deployment_readiness( self.engine, self.settings, load_builtin_catalog(), self.signer, now=TEST_NOW, production_required=False, **overrides, ) def test_current_schema_operational_signer_and_empty_backlog_are_ready(self) -> None: report = self._report() self.assertTrue(report.ready) document = report.to_mapping() self.assertEqual(document["status"], "ready") self.assertEqual(document["qualificationMode"], "nonproduction_diagnostic") checks = {value["name"]: value for value in document["checks"]} self.assertEqual(checks["migration_head"]["code"], "migration_head_current") self.assertEqual(checks["signer"]["code"], "signer_operational") self.assertEqual( checks["catalog_registry"]["code"], "catalog_registry_ready", ) self.assertEqual( checks["signing_key_registry"]["code"], "signing_key_registry_ready", ) self.assertEqual(checks["worker_backlog"]["code"], "worker_backlog_healthy") serialized = json.dumps(document, sort_keys=True) self.assertNotIn(self.database_url, serialized) self.assertNotIn("ssssssss", serialized) self.assertNotIn("ffffffff", serialized) def test_unregistered_or_substituted_signer_fails_registry_readiness(self) -> None: with Session(self.engine) as session, session.begin(): session.query(SigningKeyMetadata).delete() missing = self._report() missing_check = {value.name: value for value in missing.checks}[ "signing_key_registry" ] self.assertFalse(missing.ready) self.assertEqual( missing_check.code, "signing_key_registry_not_ready", ) self.assertIn("configured_key_not_registered", missing_check.details["issues"]) with Session(self.engine) as session, session.begin(): session.add( SigningKeyMetadata( id="signing_key.test.substituted", key_id=self.signer.key_id, purpose="license_snapshot", public_key_bytes=Ed25519SnapshotSigner( self.signer.key_id, Ed25519PrivateKey.generate(), ).public_key_bytes(), status="active", not_before=TEST_NOW - timedelta(days=1), expires_at=None, activated_at=TEST_NOW - timedelta(days=1), retired_at=None, ) ) substituted = self._report() substituted_check = {value.name: value for value in substituted.checks}[ "signing_key_registry" ] self.assertFalse(substituted.ready) self.assertIn( "configured_key_material_mismatch", substituted_check.details["issues"], ) def test_missing_or_substituted_catalog_fails_registry_readiness(self) -> None: with Session(self.engine) as session, session.begin(): session.query(CatalogRelease).delete() missing = self._report() missing_check = {value.name: value for value in missing.checks}[ "catalog_registry" ] self.assertFalse(missing.ready) self.assertEqual(missing_check.code, "catalog_registry_not_ready") self.assertIn("active_catalog_count_invalid", missing_check.details["issues"]) catalog = load_builtin_catalog() with Session(self.engine) as session, session.begin(): session.add( CatalogRelease( id="catalog_release.test.substituted", product_id=catalog.product_id, revision=catalog.revision, document_json=catalog.to_mapping(), document_digest=b"x" * 32, status="active", staged_by_user_id="user.test.deployment_admin", staged_at=TEST_NOW - timedelta(days=1), published_by_user_id="user.test.deployment_admin", published_at=TEST_NOW - timedelta(days=1), retired_at=None, ) ) substituted = self._report() substituted_check = {value.name: value for value in substituted.checks}[ "catalog_registry" ] self.assertFalse(substituted.ready) self.assertIn( "active_catalog_digest_mismatch", substituted_check.details["issues"], ) self.assertIn( "invalid_catalog_release_record", substituted_check.details["issues"], ) def test_required_disabled_capabilities_and_stale_backlog_fail(self) -> None: with Session(self.engine) as session, session.begin(): session.add( OutboxEvent( id="outbox.test.deployment.failed", event_type="stripe.subscription.project", aggregate_type="subscription", aggregate_id="subscription.test.deployment", payload_json={"schema": "bounded-test"}, status="failed", available_at=TEST_NOW - timedelta(hours=1), attempts=1, processed_at=TEST_NOW - timedelta(hours=1), last_error_code="provider_unavailable", created_at=TEST_NOW - timedelta(hours=1), updated_at=TEST_NOW - timedelta(hours=1), ) ) api_report = self._report(require_commerce=True) self.assertFalse(api_report.ready) checks = {value.name: value for value in api_report.checks} self.assertEqual(checks["commerce"].code, "commerce_required_but_disabled") self.assertEqual(checks["worker_backlog"].code, "worker_backlog_unhealthy") notification_report = inspect_deployment_readiness( self.engine, self.notification_settings, load_builtin_catalog(), None, now=TEST_NOW, production_required=False, component="notification_worker", ) self.assertFalse(notification_report.ready) checks = {value.name: value for value in notification_report.checks} self.assertEqual( checks["notifications"].code, "notifications_incomplete", ) self.assertEqual(checks["worker_backlog"].code, "worker_backlog_unhealthy") self.assertEqual( checks["worker_backlog"].details["failedOutboxEvents"], 1, ) with self.assertRaisesRegex(ValueError, "only to the API"): inspect_deployment_readiness( self.engine, self.notification_settings, load_builtin_catalog(), None, now=TEST_NOW, production_required=False, require_commerce=True, component="notification_worker", ) subscription_report = inspect_deployment_readiness( self.engine, self.subscription_settings, load_builtin_catalog(), None, now=TEST_NOW, production_required=False, component="subscription_worker", ) self.assertFalse(subscription_report.ready) checks = {value.name: value for value in subscription_report.checks} self.assertEqual( checks["subscription_provider"].code, "subscription_provider_incomplete", ) with self.assertRaisesRegex(TypeError, "SubscriptionWorkerSettings"): inspect_deployment_readiness( self.engine, self.settings, load_builtin_catalog(), None, now=TEST_NOW, production_required=False, component="subscription_worker", ) def test_migration_drift_and_signer_outage_are_explicit_failures(self) -> None: command.downgrade(self.configuration, "e2a6c8f14b73") report = inspect_deployment_readiness( self.engine, self.settings, load_builtin_catalog(), _FailingSigner(), now=TEST_NOW, production_required=False, ) self.assertFalse(report.ready) checks = {value.name: value for value in report.checks} self.assertEqual(checks["signer"].code, "signer_probe_failed") self.assertEqual(checks["database_schema"].code, "database_schema_incomplete") self.assertEqual(checks["migration_head"].code, "migration_head_mismatch") self.assertEqual( checks["worker_backlog"].code, "worker_backlog_unavailable", ) if __name__ == "__main__": unittest.main()