from __future__ import annotations import json from pathlib import Path import pytest import gateway.platforms.dualcoach_activation_cutover as cutover_module from gateway.platforms.dualcoach_activation_cutover import ( ActivationCutoverError, activate_customer_cutover, ) from gateway.platforms.telegram_customer_bootstrap import ( BootstrapState, GenerationConflict, ) from gateway.platforms.telegram_nutrition_onboarding_publication_outbox import ( GatewayOnboardingPublicationOutbox, ) from .new_customer_consent_testkit import ( awaiting_consent, consent_handoff, ) def test_consent_commit_requires_handoff_bound_before_transition( tmp_path: Path, ) -> None: store, receipted = awaiting_consent(tmp_path) with pytest.raises(GenerationConflict, match="handoff"): _ = store.reconcile_committed_consent( receipted.session_id, expected_generation=receipted.generation, publication_attempt=receipted.consent_publication_attempt, consent_card_message_id="500", ) bound = store.bind_consent_handoff( receipted.session_id, expected_generation=receipted.generation, handoff=consent_handoff(receipted), ) committed = store.reconcile_committed_consent( bound.session_id, expected_generation=bound.generation, publication_attempt=bound.consent_publication_attempt, consent_card_message_id="500", ) assert committed.state is BootstrapState.AWAITING_ACTIVATION assert committed.consent_handoff is not None def test_activation_rejects_unreconciled_consent_handoff( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: profile = tmp_path / "profile" profile.mkdir(mode=0o700) store, receipted = awaiting_consent(profile) bound = store.bind_consent_handoff( receipted.session_id, expected_generation=receipted.generation, handoff=consent_handoff(receipted), ) committed = store.reconcile_committed_consent( bound.session_id, expected_generation=bound.generation, publication_attempt=bound.consent_publication_attempt, consent_card_message_id="500", ) def reject_profile_admin(*_args: object, **_kwargs: object) -> None: pytest.fail("profile activation reached without reconciled consent") monkeypatch.setattr( cutover_module, "_profile_admin", reject_profile_admin, ) with pytest.raises(GenerationConflict, match="consent"): _ = activate_customer_cutover( profile, profile / "data/customers/customer_001", "customer_001", tmp_path / "checklist.json", tmp_path / "membership.json", bootstrap_session_id=committed.session_id, expected_generation=committed.generation, deployment_receipt_path=tmp_path / "deployment.json", ) def test_first_claim_activation_requires_completed_owner_review( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: # Given: a first-claim customer has consented but has no owner-reviewed readiness. profile = tmp_path / "profile" profile.mkdir(mode=0o700) store, receipted = awaiting_consent(profile, first_claim=True) bound = store.bind_consent_handoff( receipted.session_id, expected_generation=receipted.generation, handoff=consent_handoff(receipted), ) committed = store.reconcile_committed_consent( bound.session_id, expected_generation=bound.generation, publication_attempt=bound.consent_publication_attempt, consent_card_message_id="500", ) assert committed.consent_handoff is not None reconciled = store.mark_consent_recovery_reconciled( committed.session_id, expected_generation=committed.generation, provenance_digest=committed.consent_handoff.provenance_digest, ) def reject_profile_admin(*_args: object, **_kwargs: object) -> None: pytest.fail("profile activation reached without owner-reviewed intake") monkeypatch.setattr(cutover_module, "_profile_admin", reject_profile_admin) # When/Then: activation stops before the profile activation boundary. with pytest.raises(ActivationCutoverError, match="owner review"): _ = activate_customer_cutover( profile, profile / "data/customers/customer_001", "customer_001", tmp_path / "checklist.json", tmp_path / "membership.json", bootstrap_session_id=reconciled.session_id, expected_generation=reconciled.generation, deployment_receipt_path=tmp_path / "deployment.json", ) def test_first_claim_owner_review_receipt_unlocks_activation_preflight( tmp_path: Path, ) -> None: profile = tmp_path / "profile" profile.mkdir(mode=0o700) _store, session = awaiting_consent(profile, first_claim=True) owner_receipt = "a" * 64 onboarding = ( profile / "data" / "customers" / session.customer_key / "nutrition-onboarding" ) onboarding.mkdir(mode=0o700, parents=True) baseline = onboarding / "baseline-v1.json" readiness = onboarding / "readiness-receipt-v1.json" _ = baseline.write_text( json.dumps({ "schema_version": "2.1", "customer_key": session.customer_key, "owner_review_receipt": owner_receipt, "eating_disorder_risk": False, }), encoding="utf-8", ) _ = readiness.write_text( json.dumps({ "owner_review_receipt": owner_receipt, "delivery_enabled": False, "activation_enabled": False, "digest": "c" * 64, }), encoding="utf-8", ) _ = (onboarding / "readiness-current.json").write_text( json.dumps({ "readiness_receipt_digest": "c" * 64, "digest": "d" * 64, }), encoding="utf-8", ) _ = (onboarding / "ready.json").write_text( json.dumps({ "state": "ready", "baseline_digest": "b" * 64, "readiness_pointer_digest": "d" * 64, }), encoding="utf-8", ) _ = (onboarding / "baseline-candidate.json").write_text( json.dumps({ "source_answers_digest": "e" * 64, }), encoding="utf-8", ) baseline_document = json.loads(baseline.read_text()) baseline_document["digest"] = "b" * 64 _ = baseline.write_text(json.dumps(baseline_document), encoding="utf-8") baseline.chmod(0o600) readiness.chmod(0o600) for path in ( onboarding / "readiness-current.json", onboarding / "ready.json", onboarding / "baseline-candidate.json", ): path.chmod(0o600) outbox = GatewayOnboardingPublicationOutbox(profile) publication, created = outbox.claim( session_id=session.session_id, generation=1, payload={"state": "owner_review"}, route=("12", "0"), role="owner", render_identity="b" * 64, ) assert created is True receipted = outbox.record_receipt( session_id=session.session_id, generation=publication.generation, chat_id="12", topic_id="0", message_id=500, ) _ = outbox.mark_committed( session_id=session.session_id, generation=publication.generation, payload=publication.payload, route=("12", "0"), role="owner", render_identity="b" * 64, message_id=500, ) _ = outbox.record_owner_callback( session_id=session.session_id, customer_key=session.customer_key, action="Approve", actor_user_id=12, authority=(12, 12, 0), route=("12", "0"), message_id=500, update_id=700, consumed_updates_before=(), callback_data="owner-approve", publication_generation=receipted.generation, ) cutover_module.validate_first_claim_owner_review(profile, session) def test_first_claim_activation_rejects_unscrubbed_ready_state( tmp_path: Path, ) -> None: profile = tmp_path / "profile" profile.mkdir(mode=0o700) _store, session = awaiting_consent(profile, first_claim=True) onboarding = ( profile / "data" / "customers" / session.customer_key / "nutrition-onboarding" ) onboarding.mkdir(mode=0o700, parents=True) owner_receipt = "a" * 64 documents = { "baseline-v1.json": { "schema_version": "2.1", "customer_key": session.customer_key, "owner_review_receipt": owner_receipt, "eating_disorder_risk": False, "digest": "b" * 64, }, "readiness-receipt-v1.json": { "owner_review_receipt": owner_receipt, "delivery_enabled": False, "activation_enabled": False, "digest": "c" * 64, }, "readiness-current.json": { "readiness_receipt_digest": "c" * 64, "digest": "d" * 64, }, "ready.json": { "state": "ready", "baseline_digest": "b" * 64, "readiness_pointer_digest": "d" * 64, }, "baseline-candidate.json": { "source_answers": {"conditions": {"status": "none", "items": []}}, }, } for name, document in documents.items(): path = onboarding / name _ = path.write_text(json.dumps(document), encoding="utf-8") path.chmod(0o600) outbox = GatewayOnboardingPublicationOutbox(profile) publication, _ = outbox.claim( session_id=session.session_id, generation=1, payload={"state": "owner_review"}, route=("12", "0"), role="owner", render_identity="b" * 64, ) receipted = outbox.record_receipt( session_id=session.session_id, generation=publication.generation, chat_id="12", topic_id="0", message_id=500, ) _ = outbox.mark_committed( session_id=session.session_id, generation=publication.generation, payload=publication.payload, route=("12", "0"), role="owner", render_identity="b" * 64, message_id=500, ) _ = outbox.record_owner_callback( session_id=session.session_id, customer_key=session.customer_key, action="Approve", actor_user_id=12, authority=(12, 12, 0), route=("12", "0"), message_id=500, update_id=700, consumed_updates_before=(), callback_data="owner-approve", publication_generation=receipted.generation, ) with pytest.raises(ActivationCutoverError, match="owner review"): cutover_module.validate_first_claim_owner_review(profile, session) def test_forged_owner_review_digest_without_callback_is_rejected( tmp_path: Path, ) -> None: profile = tmp_path / "profile" profile.mkdir(mode=0o700) _store, session = awaiting_consent(profile, first_claim=True) onboarding = ( profile / "data" / "customers" / session.customer_key / "nutrition-onboarding" ) onboarding.mkdir(mode=0o700, parents=True) owner_receipt = "a" * 64 baseline = onboarding / "baseline-v1.json" readiness = onboarding / "readiness-receipt-v1.json" _ = baseline.write_text( json.dumps({ "customer_key": session.customer_key, "owner_review_receipt": owner_receipt, }), encoding="utf-8", ) _ = readiness.write_text( json.dumps({ "owner_review_receipt": owner_receipt, "delivery_enabled": False, "activation_enabled": False, }), encoding="utf-8", ) baseline.chmod(0o600) readiness.chmod(0o600) with pytest.raises(ActivationCutoverError, match="owner review"): cutover_module.validate_first_claim_owner_review(profile, session)