"""7M Forms multi-step: client-assembled collection, one final submission.""" from __future__ import annotations import inspect from datetime import UTC, datetime from pathlib import Path import pytest from pydantic import ValidationError from src.domain.connectors.invoke_capabilities import ( CAPABILITY_CONVERSION_PATTERN, FORMS_FAMILY_PLAN, ) from src.domain.forms.actions import ( FormPostSubmitContext, FormPostSubmitPlan, RecordingCapabilityInvoker, execute_post_submit_actions, ) from src.domain.forms.anti_abuse import ( ChallengeVerificationResult, FormAntiAbusePolicy, InMemoryAttemptLimiter, StaticChallengeVerifier, process_form_submission_request, ) from src.domain.forms.canonical import ( NAMING_COLLISIONS, FormDefinition, FormField, FormFieldOption, FormFileRef, FormSubmission, contact_block_compatible_definition, ) from src.domain.forms.conditions import ( CONDITIONAL_FIELD_NOT_ALLOWED, ActionConditionalRule, FieldConditionalRule, FormCondition, FormConditionSet, evaluate_field_conditions, ) from src.domain.forms.email import email_send_action from src.domain.forms.files import ( FormFileRecord, InMemoryFormFileDirectory, accept_form_submission_with_files, ) from src.domain.forms.persistence import ( FormPersistenceContext, InMemoryFormSubmissionStore, get_form_submission, persist_form_submission, ) from src.domain.forms.runtime import accept_form_submission from src.domain.forms.steps import ( COLLECTION_MODEL, FORM_SECTION_IS_NOT_FORM_STEP, FORMS_MULTI_STEP, FormStep, FormStepPlan, active_steps, assemble_candidate_values, field_step_map, validate_step_plan, ) _FIXED_AT = datetime(2026, 9, 13, 22, 30, tzinfo=UTC) _FSB = "fsb_" + ("d" * 32) _FILE = "file_" + ("a" * 32) _CTX = FormPostSubmitContext(project_id="prj_a") _PERSIST = FormPersistenceContext(project_id="prj_a") def _definition() -> FormDefinition: return FormDefinition( key="lead", fields=[ FormField( key="contact_method", type="select", required=True, options=[ FormFieldOption(value="email", label="Email"), FormFieldOption(value="phone", label="Phone"), ], ), FormField(key="has_company", type="checkbox", required=False), FormField(key="opt_in", type="checkbox", required=False), FormField(key="message", type="textarea", required=False), FormField(key="phone", type="phone", required=False), FormField(key="company_name", type="text", required=False), FormField(key="email", type="email", required=False), FormField(key="attachment", type="file", required=False), FormField(key="campaign", type="hidden", required=False), ], ) def _plan() -> FormStepPlan: return FormStepPlan( form_key="lead", steps=( FormStep( key="contact", field_keys=("contact_method", "email", "has_company", "opt_in"), ), FormStep(key="details", field_keys=("company_name", "attachment")), FormStep(key="message", field_keys=("message", "phone")), ), ) def _phone_rules() -> FormConditionSet: phone = FormCondition(field_key="contact_method", operator="equals", value="phone") company = FormCondition(field_key="has_company", operator="equals", value=True) return FormConditionSet( field_rules=( FieldConditionalRule(target_field="phone", effect="show_when", condition=phone), FieldConditionalRule( target_field="phone", effect="require_when", condition=phone ), FieldConditionalRule( target_field="company_name", effect="show_when", condition=company ), ), action_rules=( ActionConditionalRule( action_key="send_confirmation", condition=FormCondition(field_key="opt_in", operator="equals", value=True), ), ), ) def test_7m_multi_step_frozen() -> None: assert FORMS_MULTI_STEP["status"] == "frozen" assert FORMS_MULTI_STEP["collection_model"] == COLLECTION_MODEL == "client_assembled" assert FORMS_MULTI_STEP["draft_persistence"] is False assert FORMS_MULTI_STEP["server_draft"] is False assert FORMS_MULTI_STEP["step_graph"] == "linear" assert FORMS_MULTI_STEP["step_expressions"] is False assert FORMS_MULTI_STEP["fsb_per_step"] is False assert FORMS_MULTI_STEP["values_remain_flat"] is True assert FORM_SECTION_IS_NOT_FORM_STEP is True assert NAMING_COLLISIONS["form.section"] == "starter_ui_layout" assert CAPABILITY_CONVERSION_PATTERN["phase_7_status"] == "complete" assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == ( "9_app_block_modernization" ) assert CAPABILITY_CONVERSION_PATTERN["opened_families"]["forms.*"] == "composition_integration" assert FORMS_FAMILY_PLAN["status"] == "complete" assert FORMS_FAMILY_PLAN["multi_step_frozen"] is True assert FORMS_FAMILY_PLAN["slices"][12] == "7m_multi_step_forms" assert FORMS_FAMILY_PLAN["next_slice"] == "9_app_block_modernization" assert "step_plan" not in inspect.signature(accept_form_submission).parameters assert "current_step" not in FormSubmission.model_fields src = Path("src/domain/forms/steps.py").read_text(encoding="utf-8").lower() for banned in ("draft_id", "resume_token", "formdraft", "goto_step"): assert banned not in src def test_step_plan_membership_rules() -> None: definition = _definition() plan = _plan() assert validate_step_plan(definition, plan) == () assert list(field_step_map(plan)) == [ "contact_method", "email", "has_company", "opt_in", "company_name", "attachment", "message", "phone", ] assert field_step_map(plan)["phone"] == "message" assert "campaign" not in field_step_map(plan) unknown = FormStepPlan( steps=(FormStep(key="contact", field_keys=("email", "does_not_exist")),) ) assert "unknown_step_field" in validate_step_plan( FormDefinition(key="lead", fields=[FormField(key="email", type="email")]), unknown, ) dup = FormStepPlan( steps=( FormStep(key="a", field_keys=("email",)), FormStep(key="b", field_keys=("email",)), ) ) assert "duplicate_step_field" in validate_step_plan( FormDefinition(key="lead", fields=[FormField(key="email", type="email")]), dup, ) with pytest.raises(ValidationError): FormStep(key="empty", field_keys=()) missing_input = FormStepPlan(steps=(FormStep(key="contact", field_keys=("email",)),)) assert "unassigned_input_field" in validate_step_plan(definition, missing_input) with pytest.raises(ValidationError): FormStepPlan( steps=( FormStep(key="contact", field_keys=("email",)), FormStep(key="contact", field_keys=("phone",)), ) ) assert [step.key for step in plan.steps] == ["contact", "details", "message"] def test_contact_form_needs_no_step_plan() -> None: contact = contact_block_compatible_definition() result = accept_form_submission( contact, {"name": "Ada", "email": "ada@example.com", "message": "Hi"}, clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) assert result.accepted is True assert result.submission is not None assert result.submission.submission_id == _FSB def test_final_payload_equivalence_single_vs_assembled() -> None: definition = _definition() values = { "contact_method": "email", "email": "ada@example.com", "has_company": False, "opt_in": True, "message": "Hello", } step1 = { "contact_method": "email", "email": "ada@example.com", "has_company": False, "opt_in": True, } step3 = {"message": "Hello"} assembled = assemble_candidate_values(step1, step3) assert assembled == values single = accept_form_submission( definition, values, conditions=_phone_rules(), clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) multi = accept_form_submission( definition, assembled, conditions=_phone_rules(), clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) assert single.accepted is True and multi.accepted is True assert single.submission is not None and multi.submission is not None assert single.submission.submission.values == multi.submission.submission.values assert "current_step" not in single.submission.submission.values assert "completed_steps" not in single.submission.submission.values def test_no_partial_identity_persistence_or_actions() -> None: definition = _definition() store = InMemoryFormSubmissionStore() invoker = RecordingCapabilityInvoker() candidate = assemble_candidate_values( {"contact_method": "email", "email": "ada@example.com", "opt_in": True}, {"message": "Hello"}, ) assert get_form_submission(store, _PERSIST, _FSB) is None assert invoker.calls == [] accepted = accept_form_submission( definition, candidate, conditions=_phone_rules(), clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) assert accepted.accepted is True assert accepted.submission is not None persist_form_submission(store, _PERSIST, accepted.submission) row = get_form_submission(store, _PERSIST, _FSB) assert row is not None execute_post_submit_actions( accepted.submission, FormPostSubmitPlan(actions=(email_send_action(),)), context=_CTX, invoker=invoker, definition=definition, conditions=_phone_rules(), ) assert len(invoker.calls) == 1 payload = invoker.calls[0]["payload"] assert "current_step" not in payload assert "completed_steps" not in payload assert "progress" not in payload def test_cross_step_requiredness_and_hidden_rejection() -> None: definition = _definition() rules = _phone_rules() missing_phone = accept_form_submission( definition, assemble_candidate_values( {"contact_method": "phone", "opt_in": False}, {"message": "Call me"}, ), conditions=rules, ) assert missing_phone.accepted is False assert any(err.field == "phone" and err.code == "required" for err in missing_phone.errors) stale = accept_form_submission( definition, assemble_candidate_values( {"contact_method": "email", "email": "a@b.co", "has_company": False}, {"company_name": "Acme"}, ), conditions=rules, ) assert stale.accepted is False assert any( err.field == "company_name" and err.code == CONDITIONAL_FIELD_NOT_ALLOWED for err in stale.errors ) back_nav = {"contact_method": "email", "has_company": True, "company_name": "Acme"} back_nav["has_company"] = False rejected = accept_form_submission(definition, back_nav, conditions=rules) assert rejected.accepted is False assert any(err.code == CONDITIONAL_FIELD_NOT_ALLOWED for err in rejected.errors) def test_dynamically_empty_step_is_not_an_acceptance_rule() -> None: definition = _definition() plan = FormStepPlan( form_key="lead", steps=( FormStep( key="contact", field_keys=("contact_method", "email", "has_company", "opt_in"), ), FormStep(key="details", field_keys=("company_name",)), FormStep(key="message", field_keys=("message", "phone", "attachment")), ), ) rules = _phone_rules() raw = { "contact_method": "email", "email": "ada@example.com", "has_company": False, "opt_in": False, "message": "Hi", } state = evaluate_field_conditions(definition, raw, rules) assert "details" not in active_steps(plan, state, definition=definition) result = accept_form_submission( definition, raw, conditions=rules, clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) assert result.accepted is True def test_file_across_steps_uses_7f() -> None: definition = _definition() files = InMemoryFormFileDirectory() files.put(FormFileRecord(project_id="prj_a", file_id=_FILE)) ok = accept_form_submission_with_files( definition, assemble_candidate_values( {"contact_method": "email", "email": "ada@example.com"}, {"attachment": FormFileRef(file_id=_FILE)}, {"message": "Quote"}, ), project_id="prj_a", files=files, conditions=_phone_rules(), clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) assert ok.accepted is True missing = accept_form_submission_with_files( definition, assemble_candidate_values( {"contact_method": "email", "email": "ada@example.com"}, {"attachment": FormFileRef(file_id=_FILE)}, ), project_id="prj_a", files=InMemoryFormFileDirectory(), conditions=_phone_rules(), clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) assert missing.accepted is False assert any(err.code == "invalid_file" for err in missing.errors) assert missing.submission is None def test_anti_abuse_counts_final_request_only() -> None: definition = _definition() limiter = InMemoryAttemptLimiter() policy = FormAntiAbusePolicy( challenge_required=False, honeypot_enabled=False, rate_limit_max_attempts=20, ) verifier = StaticChallengeVerifier(ChallengeVerificationResult(passed=True)) assert limiter._hits == {} result = process_form_submission_request( definition, { "contact_method": "email", "email": "ada@example.com", "has_company": False, "message": "Hi", }, policy=policy, project_id="prj_a", client_identity="client-1", verifier=verifier, limiter=limiter, conditions=_phone_rules(), clock=lambda: _FIXED_AT, submission_id_factory=lambda: _FSB, ) assert result.accepted is True assert sum(len(hits) for hits in limiter._hits.values()) == 1 unknown = accept_form_submission( definition, { "contact_method": "email", "email": "ada@example.com", "step1_valid": True, }, conditions=_phone_rules(), ) assert unknown.accepted is False assert any(err.code == "unknown_field" for err in unknown.errors)