"""Guarded catalog candidate and publication regression tests.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import json import unittest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from licensing_shared.catalog import LicensingCatalog, load_builtin_catalog from licensing_server.app.catalog_administration import ( CatalogAdministrationService, CatalogBoundSnapshotSigner, inspect_catalog_registry, ) from licensing_server.app.database import Base from licensing_server.app.errors import ServerErrorCode, ServerLicensingError from licensing_server.app.models import AuditEvent, CatalogRelease, User from licensing_server.app.security import Ed25519SnapshotSigner TEST_NOW = datetime(2026, 8, 16, 20, 0, tzinfo=timezone.utc) class CatalogAdministrationTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(self.engine) self.catalog = load_builtin_catalog() self.signer = Ed25519SnapshotSigner( "license.test.catalog", Ed25519PrivateKey.generate(), ) with Session(self.engine) as session, session.begin(): session.add_all( ( User( id="user.test.catalog_admin", external_issuer="https://identity.test", external_subject="catalog-admin", verified_email="catalog-admin@example.test", status="active", is_server_admin=True, ), User( id="user.test.catalog_customer", external_issuer="https://identity.test", external_subject="catalog-customer", verified_email="catalog-customer@example.test", status="active", is_server_admin=False, ), ) ) def tearDown(self) -> None: self.engine.dispose() @staticmethod def _service( session: Session, runtime_catalog: LicensingCatalog, ) -> CatalogAdministrationService: return CatalogAdministrationService( session, runtime_catalog, now_factory=lambda: TEST_NOW, ) def _add_active_release( self, session: Session, catalog: LicensingCatalog, ) -> None: session.add( CatalogRelease( id=f"catalog_release.test.revision_{catalog.revision}", 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.catalog_admin", staged_at=TEST_NOW - timedelta(days=1), published_by_user_id="user.test.catalog_admin", published_at=TEST_NOW - timedelta(days=1), retired_at=None, ) ) def _next_catalog(self) -> LicensingCatalog: document = self.catalog.to_mapping() document["revision"] = self.catalog.revision + 1 entitlements = document["entitlements"] assert isinstance(entitlements, dict) advanced = entitlements["feature.analysis.advanced"] assert isinstance(advanced, dict) advanced["label"] = "Advanced engineering analysis" return LicensingCatalog.from_mapping(document) def test_initial_publication_requires_admin_digest_and_exact_idempotency(self) -> None: with Session(self.engine) as session: service = self._service(session, self.catalog) with self.assertRaises(ServerLicensingError) as denied: service.stage_candidate( self.catalog.to_mapping(), "user.test.catalog_customer", "initial_provisioning", "idempotency-catalog-stage-denied", "correlation.test.catalog_stage_denied", ) self.assertEqual(denied.exception.code, ServerErrorCode.AUTHORIZATION_DENIED) with Session(self.engine) as session: staged = self._service(session, self.catalog).stage_candidate( self.catalog.to_mapping(), "user.test.catalog_admin", "initial_provisioning", "idempotency-catalog-stage-one", "correlation.test.catalog_stage_one", ) self.assertTrue(staged["registered"]) self.assertEqual(staged["catalogSha256"], self.catalog.sha256()) substituted = self.catalog.to_mapping() entitlements = substituted["entitlements"] assert isinstance(entitlements, dict) personal = entitlements["right.personal_use"] assert isinstance(personal, dict) personal["label"] = "Substituted content" with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as conflict: self._service(session, self.catalog).stage_candidate( substituted, "user.test.catalog_admin", "support_correction", "idempotency-catalog-stage-substitution", "correlation.test.catalog_stage_substitution", ) self.assertEqual(conflict.exception.code, ServerErrorCode.CONFLICT) with Session(self.engine) as session: preview = self._service(session, self.catalog).preview_publication( self.catalog.revision, "user.test.catalog_admin", "initial_provisioning", "correlation.test.catalog_publish_one", ) self.assertTrue(preview["canPublish"]) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as stale: self._service(session, self.catalog).publish_candidate( self.catalog.revision, "user.test.catalog_admin", "initial_provisioning", "00" * 32, "idempotency-catalog-publish-one", "correlation.test.catalog_publish_one", ) self.assertEqual(stale.exception.code, ServerErrorCode.CONFLICT) with Session(self.engine) as session: published = self._service(session, self.catalog).publish_candidate( self.catalog.revision, "user.test.catalog_admin", "initial_provisioning", preview["stateDigest"], "idempotency-catalog-publish-one", "correlation.test.catalog_publish_one", ) self.assertTrue(published["catalogReady"]) self.assertTrue(published["executed"]) with Session(self.engine) as session: replay = self._service(session, self.catalog).publish_candidate( self.catalog.revision, "user.test.catalog_admin", "initial_provisioning", preview["stateDigest"], "idempotency-catalog-publish-one", "correlation.test.catalog_publish_one", ) audit = session.scalars( select(AuditEvent).where(AuditEvent.target_type == "catalog_release") ).all() self.assertTrue(replay["idempotentReplay"]) self.assertEqual(len(audit), 2) serialized_audit = json.dumps( [event.metadata_json for event in audit], sort_keys=True, ) self.assertNotIn("entitlements", serialized_audit) self.assertNotIn("devicePolicies", serialized_audit) def test_append_only_rotation_rejects_removal_and_stale_preview(self) -> None: next_catalog = self._next_catalog() with Session(self.engine) as session, session.begin(): self._add_active_release(session, self.catalog) removed_sku = next_catalog.to_mapping() skus = removed_sku["skus"] assert isinstance(skus, dict) del skus["subscription.ai_assist"] with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as removal: self._service(session, next_catalog).stage_candidate( removed_sku, "user.test.catalog_admin", "planned_catalog_release", "idempotency-catalog-stage-removal", "correlation.test.catalog_stage_removal", ) self.assertEqual(removal.exception.code, ServerErrorCode.CONFLICT) self.assertIn("stable_sku_removed", str(removal.exception)) changed_policy = next_catalog.to_mapping() policies = changed_policy["devicePolicies"] assert isinstance(policies, dict) standard_policy = policies["policy.named_user.standard"] assert isinstance(standard_policy, dict) standard_policy["connectedRefreshHours"] = 1 with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as policy_change: self._service(session, next_catalog).stage_candidate( changed_policy, "user.test.catalog_admin", "planned_catalog_release", "idempotency-catalog-stage-policy-change", "correlation.test.catalog_stage_policy_change", ) self.assertEqual(policy_change.exception.code, ServerErrorCode.CONFLICT) self.assertIn("existing_device_policy_changed", str(policy_change.exception)) with Session(self.engine) as session: staged = self._service(session, next_catalog).stage_candidate( next_catalog.to_mapping(), "user.test.catalog_admin", "planned_catalog_release", "idempotency-catalog-stage-next", "correlation.test.catalog_stage_next", ) self.assertTrue(staged["registered"]) self.assertEqual( staged["changes"]["changedEntitlementMetadata"]["count"], 1, ) with Session(self.engine) as session: old_runtime_preview = self._service( session, self.catalog, ).preview_publication( next_catalog.revision, "user.test.catalog_admin", "planned_catalog_release", "correlation.test.catalog_publish_old_runtime", ) self.assertFalse(old_runtime_preview["canPublish"]) self.assertIn( "candidate_runtime_revision_mismatch", old_runtime_preview["issues"], ) with Session(self.engine) as session: preview = self._service(session, next_catalog).preview_publication( next_catalog.revision, "user.test.catalog_admin", "planned_catalog_release", "correlation.test.catalog_publish_next", ) self.assertTrue(preview["canPublish"]) with Session(self.engine) as session, session.begin(): staged_row = session.scalar( select(CatalogRelease).where( CatalogRelease.revision == next_catalog.revision ) ) assert staged_row is not None staged_row.staged_at = staged_row.staged_at + timedelta(seconds=1) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as stale: self._service(session, next_catalog).publish_candidate( next_catalog.revision, "user.test.catalog_admin", "planned_catalog_release", preview["stateDigest"], "idempotency-catalog-publish-next", "correlation.test.catalog_publish_next", ) self.assertEqual(stale.exception.code, ServerErrorCode.CONFLICT) with Session(self.engine) as session: refreshed = self._service(session, next_catalog).preview_publication( next_catalog.revision, "user.test.catalog_admin", "planned_catalog_release", "correlation.test.catalog_publish-next-refreshed", ) with Session(self.engine) as session: published = self._service(session, next_catalog).publish_candidate( next_catalog.revision, "user.test.catalog_admin", "planned_catalog_release", refreshed["stateDigest"], "idempotency-catalog-publish-next-refreshed", "correlation.test.catalog_publish-next-refreshed", ) self.assertEqual(published["previousActiveRevision"], self.catalog.revision) self.assertEqual(published["publishedRevision"], next_catalog.revision) with Session(self.engine) as session: rows = session.scalars( select(CatalogRelease).order_by(CatalogRelease.revision) ).all() self.assertEqual([row.status for row in rows], ["retired", "active"]) ready = inspect_catalog_registry(session, next_catalog) self.assertTrue(ready["catalogReady"]) signed = CatalogBoundSnapshotSigner( session, next_catalog, self.signer, ).sign_payload( {"schema": "catalog.test", "revision": next_catalog.revision} ) self.assertEqual(signed.key_id, self.signer.key_id) with self.assertRaises(ServerLicensingError) as old_runtime: CatalogBoundSnapshotSigner( session, self.catalog, self.signer, ).sign_payload( {"schema": "catalog.test", "revision": self.catalog.revision} ) self.assertEqual( old_runtime.exception.code, ServerErrorCode.CATALOG_UNAVAILABLE, ) def test_schema_one_catalog_can_roll_forward_to_schema_two_cadence(self) -> None: legacy_document = self.catalog.to_mapping() legacy_document["schemaVersion"] = 1 legacy_document["revision"] = self.catalog.revision - 1 policies = legacy_document["devicePolicies"] assert isinstance(policies, dict) for policy in policies.values(): assert isinstance(policy, dict) policy.pop("connectedRefreshHours") policy.pop("offlineRefreshReminderDays") legacy_catalog = LicensingCatalog.from_mapping(legacy_document) with Session(self.engine) as session, session.begin(): self._add_active_release(session, legacy_catalog) with Session(self.engine) as session: staged = self._service(session, self.catalog).stage_candidate( self.catalog.to_mapping(), "user.test.catalog_admin", "planned_catalog_release", "idempotency-catalog-schema-two", "correlation.test.catalog_schema_two", ) self.assertTrue(staged["registered"]) self.assertEqual(staged["revision"], self.catalog.revision) if __name__ == "__main__": unittest.main()