"""Deterministic mutations for the complete maintenance rejection matrix.""" from __future__ import annotations from dataclasses import dataclass, replace from datetime import date, datetime, timedelta from enum import StrEnum import pytest from gateway.platforms.nutrition_weekly_maintenance_contract import ( Topic59MaintenanceDecisionKind, Topic59MaintenanceHoldV1, derive_hold_id, maintenance_scope_digest, ) from gateway.platforms.nutrition_weekly_maintenance_store import Topic59MaintenanceGate from tests.gateway.nutrition_weekly_maintenance_fs_mutations import ( extra as _extra, fake_identity as _fake_identity, fifo as _fifo, hardlink as _hardlink, noncanonical as _noncanonical, replace_directory_with_symlink as _replace_directory_with_symlink, rewrite as _rewrite, symlink as _symlink, ) from tests.gateway.nutrition_weekly_maintenance_support import ( MaintenanceFixture, change_group, write_bound_hold, ) class MatrixCase(StrEnum): CREDENTIAL_MISSING = "credential_missing" AUTHORITY_EMPTY = "authority_empty" AUTHORITY_MALFORMED = "authority_malformed" AUTHORITY_OVERSIZED = "authority_oversized" AUTHORITY_NONCANONICAL = "authority_noncanonical" AUTHORITY_EXTRA = "authority_extra" AUTHORITY_HASH = "authority_hash" HOLD_EMPTY = "hold_empty" HOLD_MALFORMED = "hold_malformed" HOLD_OVERSIZED = "hold_oversized" HOLD_NONCANONICAL = "hold_noncanonical" HOLD_EXTRA = "hold_extra" HOLD_EARLY = "hold_early" HOLD_EXPIRED = "hold_expired" HOLD_REPLAY = "hold_replay" WRONG_CANDIDATE = "wrong_candidate" WRONG_CONFIG = "wrong_config" WRONG_ROUTE = "wrong_route" WRONG_CUSTOMER = "wrong_customer" WRONG_SLOT = "wrong_slot" WRONG_DAY = "wrong_day" CREDENTIAL_SYMLINK = "credential_symlink" CREDENTIAL_HARDLINK = "credential_hardlink" CREDENTIAL_MODE = "credential_mode" CREDENTIAL_KIND = "credential_kind" CREDENTIAL_OWNER = "credential_owner" DIRECTORY_SYMLINK = "directory_symlink" DIRECTORY_MODE = "directory_mode" DIRECTORY_GID = "directory_gid" DIRECTORY_OWNER = "directory_owner" ACTIVE_SYMLINK = "active_symlink" ACTIVE_HARDLINK = "active_hardlink" ACTIVE_MODE = "active_mode" ACTIVE_KIND = "active_kind" ACTIVE_GID = "active_gid" ACTIVE_OWNER = "active_owner" LOCK_SYMLINK = "lock_symlink" LOCK_HARDLINK = "lock_hardlink" LOCK_MODE = "lock_mode" LOCK_KIND = "lock_kind" LOCK_GID = "lock_gid" LOCK_OWNER = "lock_owner" AUDIT_SYMLINK = "audit_symlink" AUDIT_HARDLINK = "audit_hardlink" AUDIT_MODE = "audit_mode" AUDIT_KIND = "audit_kind" AUDIT_GID = "audit_gid" AUDIT_OWNER = "audit_owner" ACTIVE_AND_CONSUMING = "active_and_consuming" PARTIAL_TEMP = "partial_temp" PARTIAL_AUDIT = "partial_audit" @dataclass(frozen=True, slots=True) class HoldBinding: candidate: str config: str route: str customer: str slot: str day: date not_before: datetime expires_at: datetime @classmethod def exact(cls, fixture: MaintenanceFixture) -> HoldBinding: hold = fixture.hold return cls( hold.candidate_digest, hold.config_digest, hold.route_digest, hold.customer_identity_digest, hold.card_slot, hold.kst_day, hold.not_before, hold.expires_at, ) def _bind(fixture: MaintenanceFixture, binding: HoldBinding) -> None: scope = fixture.hold.scope.model_copy(update={ "candidate_digest": binding.candidate, "config_digest": binding.config, "route_digest": binding.route, "customer_identity_digest": binding.customer, "card_slot": binding.slot, "kst_day": binding.day, "not_before": binding.not_before, "expires_at": binding.expires_at, }) binding_digest = fixture.hold.package_binding_digest write_bound_hold(fixture, Topic59MaintenanceHoldV1( hold_id=derive_hold_id(binding_digest, maintenance_scope_digest(scope)), package_binding_digest=binding_digest, scope=scope, )) def apply_case( case: MatrixCase, fixture: MaintenanceFixture, monkeypatch: pytest.MonkeyPatch, ) -> bool: """Apply one mutation and report whether setup created a valid audit.""" active = fixture.maintenance / "hold.json" lock = fixture.maintenance / "maintenance.lock" match case: case MatrixCase.CREDENTIAL_MISSING: fixture.credential.unlink() case MatrixCase.AUTHORITY_EMPTY: _rewrite(fixture.credential, b"", 0o400) case MatrixCase.AUTHORITY_MALFORMED: _rewrite(fixture.credential, b"{\n", 0o400) case MatrixCase.AUTHORITY_OVERSIZED: _rewrite(fixture.credential, b"x" * 8193, 0o400) case MatrixCase.AUTHORITY_NONCANONICAL: _rewrite(fixture.credential, _noncanonical(fixture.credential.read_bytes()), 0o400) case MatrixCase.AUTHORITY_EXTRA: _rewrite(fixture.credential, _extra(fixture.credential.read_bytes()), 0o400) case MatrixCase.AUTHORITY_HASH: raw = fixture.credential.read_bytes().replace( fixture.authority.hold_sha256.encode(), b"0" * 64, ) _rewrite(fixture.credential, raw, 0o400) case MatrixCase.HOLD_EMPTY: _rewrite(active, b"", 0o600) case MatrixCase.HOLD_MALFORMED: _rewrite(active, b"{\n", 0o600) case MatrixCase.HOLD_OVERSIZED: _rewrite(active, b"x" * 8193, 0o600) case MatrixCase.HOLD_NONCANONICAL: _rewrite(active, _noncanonical(fixture.hold_bytes), 0o600) case MatrixCase.HOLD_EXTRA: _rewrite(active, _extra(fixture.hold_bytes), 0o600) case MatrixCase.HOLD_EARLY: binding = replace( HoldBinding.exact(fixture), not_before=fixture.now + timedelta(minutes=1), expires_at=fixture.now + timedelta(minutes=6), ) _bind(fixture, binding) case MatrixCase.HOLD_EXPIRED: binding = replace( HoldBinding.exact(fixture), not_before=fixture.now - timedelta(minutes=6), expires_at=fixture.now - timedelta(minutes=1), ) _bind(fixture, binding) case MatrixCase.HOLD_REPLAY: _create_valid_audit(fixture) _rewrite(active, fixture.hold_bytes, 0o600) case MatrixCase.WRONG_CANDIDATE: _bind(fixture, replace(HoldBinding.exact(fixture), candidate="8" * 64)) case MatrixCase.WRONG_CONFIG: _bind(fixture, replace(HoldBinding.exact(fixture), config="8" * 64)) case MatrixCase.WRONG_ROUTE: _bind(fixture, replace(HoldBinding.exact(fixture), route="8" * 64)) case MatrixCase.WRONG_CUSTOMER: _bind(fixture, replace(HoldBinding.exact(fixture), customer="8" * 64)) case MatrixCase.WRONG_SLOT: _bind(fixture, replace(HoldBinding.exact(fixture), slot="8" * 64)) case MatrixCase.WRONG_DAY: _bind(fixture, replace( HoldBinding.exact(fixture), day=fixture.hold.kst_day + timedelta(days=1), not_before=fixture.hold.not_before + timedelta(days=1), expires_at=fixture.hold.expires_at + timedelta(days=1), )) case MatrixCase.CREDENTIAL_SYMLINK: _symlink(fixture.credential, active) case MatrixCase.CREDENTIAL_HARDLINK: _hardlink(fixture.credential) case MatrixCase.CREDENTIAL_MODE: _ = fixture.credential.chmod(0o600) case MatrixCase.CREDENTIAL_KIND: _fifo(fixture.credential) case MatrixCase.CREDENTIAL_OWNER: _fake_identity(monkeypatch, fixture.credential, change_uid=True) case MatrixCase.DIRECTORY_SYMLINK: _replace_directory_with_symlink(fixture) case MatrixCase.DIRECTORY_MODE: _ = fixture.maintenance.chmod(0o755) case MatrixCase.DIRECTORY_GID: _ = change_group(fixture.maintenance) case MatrixCase.DIRECTORY_OWNER: _fake_identity(monkeypatch, fixture.maintenance, change_uid=True) case MatrixCase.ACTIVE_SYMLINK: _symlink(active, fixture.credential) case MatrixCase.ACTIVE_HARDLINK: _hardlink(active) case MatrixCase.ACTIVE_MODE: _ = active.chmod(0o640) case MatrixCase.ACTIVE_KIND: _fifo(active) case MatrixCase.ACTIVE_GID: _ = change_group(active) case MatrixCase.ACTIVE_OWNER: _fake_identity(monkeypatch, active, change_uid=True) case MatrixCase.LOCK_SYMLINK: _symlink(lock, fixture.credential) case MatrixCase.LOCK_HARDLINK: _hardlink(lock) case MatrixCase.LOCK_MODE: _ = lock.chmod(0o640) case MatrixCase.LOCK_KIND: _fifo(lock) case MatrixCase.LOCK_GID: _ = change_group(lock) case MatrixCase.LOCK_OWNER: _fake_identity(monkeypatch, lock, change_uid=True) case MatrixCase.AUDIT_SYMLINK: _create_valid_audit(fixture) _symlink(fixture.audit, fixture.credential) case MatrixCase.AUDIT_HARDLINK: _create_valid_audit(fixture) _hardlink(fixture.audit) case MatrixCase.AUDIT_MODE: _create_valid_audit(fixture) _ = fixture.audit.chmod(0o640) case MatrixCase.AUDIT_KIND: _create_valid_audit(fixture) _fifo(fixture.audit) case MatrixCase.AUDIT_GID: _create_valid_audit(fixture) _ = change_group(fixture.audit) case MatrixCase.AUDIT_OWNER: _create_valid_audit(fixture) _fake_identity(monkeypatch, fixture.audit, change_uid=True) case MatrixCase.ACTIVE_AND_CONSUMING: _rewrite(fixture.maintenance / "hold.consuming.json", fixture.hold_bytes, 0o600) case MatrixCase.PARTIAL_TEMP: _rewrite(fixture.maintenance / ".skip-audit.json.tmp", b"{", 0o600) case MatrixCase.PARTIAL_AUDIT: _rewrite(fixture.audit, b"{", 0o600) return False def _create_valid_audit(fixture: MaintenanceFixture) -> None: result = Topic59MaintenanceGate.for_profile(fixture.profile).evaluate( fixture.projection, fixture.now, ) assert result.kind is Topic59MaintenanceDecisionKind.NOOP_MAINTENANCE