"""7Q Forms end-to-end application tests. No new Forms architecture.""" from __future__ import annotations import inspect from concurrent.futures import ThreadPoolExecutor from pathlib import Path from src.core.middleware import LoggingMiddleware from src.domain.app_blocks.catalogs import FEATURE_PACK_ALIASES from src.domain.codegen.forms_codegen import classify_form_intent, select_forms_generation from src.domain.connectors.invoke_capabilities import ( CAPABILITY_CONVERSION_PATTERN, FORMS_FAMILY_PLAN, FORMS_HYPOTHESIZED_VERBS_NOT_FROZEN, INVOKE_CAPABILITIES, ) from src.domain.connectors.public_gateway import PHASE_7_STATUS, PUBLIC_INVOCABLE_CAPABILITIES from src.domain.feature_packs.catalog_v2 import CATALOG_OVERLAYS from src.domain.forms.actions import FormCapabilityInvokeResult, RecordingCapabilityInvoker from src.domain.forms.anti_abuse import ( ChallengeVerificationResult, StaticChallengeVerifier, ) from src.domain.forms.canonical import FormDefinition, FormField from src.domain.forms.composition import ( FORM_BINDING_CONFLICT, CompositionFormRuntimeStore, FormBindingSource, publish_form_readiness, resolve_form_bindings, ) from src.domain.forms.conditions import ( CONDITIONAL_FIELD_NOT_ALLOWED, FieldConditionalRule, FormCondition, FormConditionSet, ) from src.domain.forms.e2e import ( BROWSER_LEAK_TOKENS, CONTACT_VALUES, FORMS_END_TO_END_PHASE_7Q, PUBLISH_READINESS_DEPLOY_GAP, browser_leak_hits, browser_submit, conflict_sources, generated_browser_sources, install_contact_app, install_quote_app, new_install_uses_canonical_forms, passing_verifier, put_platform_file, rebind, ) from src.domain.forms.persistence import ( FormPersistenceContext, FormPersistenceError, get_form_submission, ) from src.domain.forms.public import ( ResolvedFormRuntime, UnwiredChallengeVerifier, project_public_forms_response, submit_form, ) from src.domain.forms.router import forms_router from src.domain.forms.steps import FormStep, FormStepPlan from src.domain.starter_library.catalog import known_contract_ids from src.main import app as fastapi_app _FILE_A = "file_" + ("a" * 32) _FILE_B = "file_" + ("b" * 32) def _ctx(project_id: str) -> FormPersistenceContext: return FormPersistenceContext(project_id=project_id) class BoomStore: def persist(self, *args, **kwargs): raise FormPersistenceError("submission_persistence_failed") def get(self, *args, **kwargs): return None class CountingStore: def __init__(self, inner): self.inner = inner self.gets = 0 def get(self, project_id, form_key): self.gets += 1 return self.inner.get(project_id, form_key) def test_7q_closes_forms_phase_without_new_architecture() -> None: assert PHASE_7_STATUS == "complete" assert CAPABILITY_CONVERSION_PATTERN["phase_7_status"] == "complete" assert CAPABILITY_CONVERSION_PATTERN["forms_domain_status"] == "complete" assert CAPABILITY_CONVERSION_PATTERN["closed_families"]["forms.*"] == "complete" assert CAPABILITY_CONVERSION_PATTERN["next_program_slice"] == ( "9_app_block_modernization" ) assert CAPABILITY_CONVERSION_PATTERN["current_program_phase"] == 8 assert FORMS_FAMILY_PLAN["status"] == "complete" assert FORMS_FAMILY_PLAN["e2e_frozen"] is True assert FORMS_FAMILY_PLAN["slices"][16] == "7q_end_to_end_application_tests" assert FORMS_FAMILY_PLAN["next_slice"] == "9_app_block_modernization" assert FORMS_END_TO_END_PHASE_7Q["new_architecture"] is False assert FORMS_END_TO_END_PHASE_7Q["schema_migration"] is False assert FORMS_END_TO_END_PHASE_7Q["publish_form_readiness_deploy"] == ( PUBLISH_READINESS_DEPLOY_GAP ) assert "form.submit" not in INVOKE_CAPABILITIES assert "form.submit" in FORMS_HYPOTHESIZED_VERBS_NOT_FROZEN assert "form.submit" not in PUBLIC_INVOCABLE_CAPABILITIES def test_contact_happy_path_through_composition_and_installed_bff() -> None: app = install_contact_app(notify_to_email="ops@example.com") assert "forms.contact" in [block.id for block in app.plan.blocks] assert "contact-capture" in app.plan.required_feature_packs assert app.store.get(app.project_id, "contact") is not None client = app.client_config("contact") assert client["submit"]["path"] == "/api/forms/contact/submissions" assert "persist_first" not in client assert "plan" not in client result = browser_submit(app, form_key="contact", values=CONTACT_VALUES) assert result.body == { "ok": True, "status": "accepted", "submission_id": result.submission_id, } assert result.submission_id and result.submission_id.startswith("fsb_") stored = get_form_submission( app.submissions, _ctx(app.project_id), result.submission_id ) assert stored is not None assert stored.accepted.submission.form_key == "contact" assert stored.accepted.submission.values["email"] == "ada@example.com" assert stored.project_id == app.project_id assert any(call["capability"] == "crm.contact.upsert" for call in app.invoker.calls) assert any(call["capability"] == "email.send" for call in app.invoker.calls) assert all( "hubspot" not in call["capability"] and "salesforce" not in call["capability"] for call in app.invoker.calls ) def test_canonical_codegen_and_legacy_routes_coexist() -> None: assert new_install_uses_canonical_forms("forms.contact for a new install") assert new_install_uses_canonical_forms("forms.quote_request") legacy = select_forms_generation("keep /api/contact for old apps") assert legacy.runtime == "legacy_contact_adapter" contact = Path("src/domain/codegen/runtime/p2hs-contact.mjs").read_text(encoding="utf-8") assert "export async function submitContact" in contact legacy_forms = Path("src/domain/codegen/runtime/p2hs-forms.mjs").read_text( encoding="utf-8" ) assert "submitFormEntry" in legacy_forms assert "/api/forms/" not in legacy_forms client = Path("src/domain/codegen/runtime/p2hs-app-client.js").read_text( encoding="utf-8" ) assert "submitGeneratedForm" in client assert "submitContact" in client assert "/api/contact" in client assert "/api/forms/" in client assert classify_form_intent("login form") == "auth" assert classify_form_intent("edit record") == "crud" assert classify_form_intent("search form") == "search" assert classify_form_intent("checkout payment form") == "checkout" def test_quote_file_flow_and_cross_project_rejection() -> None: app_a = install_quote_app() app_b = install_quote_app(project_id="prj_quote_b") assert app_a.store.get(app_a.project_id, "quote_request") is not None assert "contact" not in app_a.resolution.runtimes put_platform_file(app_a, _FILE_A) put_platform_file(app_b, _FILE_B) values = { "name": "Ada", "email": "ada@example.com", "message": "Need a roof", "attachment": {"file_id": _FILE_A}, } ok = browser_submit(app_a, form_key="quote_request", values=values) assert ok.body["ok"] is True stored = get_form_submission(app_a.submissions, _ctx(app_a.project_id), ok.submission_id) assert stored.accepted.submission.values["attachment"].file_id == _FILE_A stolen = browser_submit(app_b, form_key="quote_request", values=values) assert stolen.http_status == 422 assert stolen.body["fields"]["attachment"] == ["invalid_file"] assert stolen.submission_id is None app_a.files.delete(app_a.project_id, _FILE_A) missing = browser_submit(app_a, form_key="quote_request", values=values) assert missing.body["fields"]["attachment"] == ["invalid_file"] dumped = str(values) assert "r2" not in dumped.lower() assert "bucket" not in dumped def test_conditions_and_multi_step_server_authority() -> None: app = install_contact_app() definition = FormDefinition( key="intake", fields=[ FormField(key="has_company", type="checkbox", required=False), FormField(key="company_name", type="text", required=False), FormField(key="email", type="email", required=True), FormField(key="accepted", type="checkbox", required=True), ], ) conditions = FormConditionSet( field_rules=( FieldConditionalRule( target_field="company_name", effect="show_when", condition=FormCondition( field_key="has_company", operator="equals", value=True ), ), ) ) steps = FormStepPlan( form_key="intake", steps=( FormStep(key="one", field_keys=("has_company", "company_name")), FormStep(key="two", field_keys=("email", "accepted")), ), ) app.overlay.put( app.project_id, ResolvedFormRuntime( definition=definition, conditions=conditions, step_plan=steps, persist_first=True, ), ) cleaned = browser_submit( app, form_key="intake", values={"has_company": False, "email": "ada@example.com", "accepted": False}, ) assert cleaned.body["ok"] is True stale = browser_submit( app, form_key="intake", values={ "has_company": False, "company_name": "Acme", "email": "ada@example.com", "accepted": False, }, ) assert stale.http_status == 422 assert stale.body["fields"]["company_name"] == [CONDITIONAL_FIELD_NOT_ALLOWED] assert stale.submission_id is None client = app.client_config("intake") assert client["steps"][0]["key"] == "one" assert "run_when" not in str(client) def test_validation_aggregation_and_coercion_boundaries() -> None: app = install_contact_app() many = browser_submit( app, form_key="contact", values={"name": "", "email": "not-an-email", "extra": "nope", "budget": ""}, ) assert many.body["error"] == "validation_failed" fields = many.body["fields"] assert "required" in fields["name"] assert "invalid_email" in fields["email"] assert "unknown_field" in fields["extra"] assert "unknown_field" in fields["budget"] runtime = app.store.get(app.project_id, "contact") number_def = FormDefinition( key="nums", fields=[ FormField(key="qty", type="number", required=False), FormField(key="flag", type="checkbox", required=False), FormField(key="accepted", type="checkbox", required=True), ], ) app.overlay.put(app.project_id, ResolvedFormRuntime(definition=number_def, persist_first=True)) blank = browser_submit(app, form_key="nums", values={"qty": "", "accepted": False}) assert "invalid_number" in blank.body["fields"]["qty"] coerced = browser_submit( app, form_key="nums", values={"flag": "on", "accepted": False} ) assert "invalid_type" in coerced.body["fields"]["flag"] ok_false = browser_submit(app, form_key="nums", values={"accepted": False}) assert ok_false.body["ok"] is True assert runtime is not None def test_anti_abuse_honeypot_challenge_and_rate_limit() -> None: app = install_contact_app() honey = browser_submit( app, form_key="contact", values={**CONTACT_VALUES, "website": "http://spam"}, ) assert honey.body == {"ok": True, "status": "accepted"} assert honey.submission_id is None assert honey.honeypot_triggered is True assert get_form_submission(app.submissions, _ctx(app.project_id), "fsb_" + "0" * 32) is None challenged = install_contact_app(project_id="prj_chal", challenge_required=True) missing = browser_submit(challenged, form_key="contact", values=CONTACT_VALUES) assert missing.body["error"] == "challenge_required" assert "turnstile" not in str(missing.body).lower() failed = browser_submit( challenged, form_key="contact", values=CONTACT_VALUES, challenge="bad", verifier=StaticChallengeVerifier( ChallengeVerificationResult(passed=False, failure_code="challenge_failed") ), ) assert failed.body["error"] == "challenge_failed" outage = browser_submit( challenged, form_key="contact", values=CONTACT_VALUES, challenge="tok", verifier=UnwiredChallengeVerifier(), ) assert outage.body["error"] == "challenge_unavailable" ok = browser_submit( challenged, form_key="contact", values=CONTACT_VALUES, challenge="tok", verifier=passing_verifier(), ) assert ok.body["ok"] is True limited = install_contact_app(project_id="prj_rate") other = install_contact_app(project_id="prj_rate_b") other.limiter = limited.limiter for _ in range(20): browser_submit(limited, form_key="contact", values=CONTACT_VALUES) blocked = browser_submit(limited, form_key="contact", values=CONTACT_VALUES) assert blocked.http_status == 429 assert blocked.body["error"] == "rate_limited" assert "retry" not in str(blocked.body).lower() still = browser_submit(other, form_key="contact", values=CONTACT_VALUES) assert still.body["ok"] is True def test_best_effort_and_persistence_failure_are_observably_different() -> None: crm_fail = RecordingCapabilityInvoker( { "crm.contact.upsert": FormCapabilityInvokeResult( status="failed", error_code="binding_ambiguous" ), "email.send": FormCapabilityInvokeResult( status="succeeded", execution_ref="mail_1" ), } ) app = install_contact_app(notify_to_email="ops@example.com") accepted = browser_submit( app, form_key="contact", values=CONTACT_VALUES, invoker=crm_fail ) assert accepted.body["ok"] is True assert "binding_ambiguous" not in str(accepted.body) assert "execution_ref" not in accepted.body assert set(crm_fail.calls[i]["capability"] for i in range(len(crm_fail.calls))) >= { "crm.contact.upsert", "email.send", } email_fail = RecordingCapabilityInvoker( { "crm.contact.upsert": FormCapabilityInvokeResult( status="succeeded", execution_ref="crm_1" ), "email.send": FormCapabilityInvokeResult( status="failed", error_code="provider_error" ), } ) second = browser_submit( app, form_key="contact", values=CONTACT_VALUES, invoker=email_fail ) assert second.body["ok"] is True boom = BoomStore() silent = RecordingCapabilityInvoker() failed = browser_submit( app, form_key="contact", values=CONTACT_VALUES, invoker=silent, submissions=boom, ) assert failed.http_status == 500 assert failed.body == {"ok": False, "error": "submission_failed"} assert failed.submission_id and failed.submission_id.startswith("fsb_") assert silent.calls == [] leaked = project_public_forms_response( ok=True, http_status=200, submission_id="fsb_" + "c" * 32 ) assert set(leaked) <= {"ok", "status", "submission_id"} def test_project_isolation_conflicts_and_lifecycle() -> None: a = install_contact_app(project_id="prj_a", notify_to_email="a@example.com") b = install_contact_app(project_id="prj_b", notify_to_email="b@example.com") first = browser_submit(a, form_key="contact", values=CONTACT_VALUES) browser_submit(b, form_key="contact", values=CONTACT_VALUES) assert a.store.get("prj_b", "contact") is None assert get_form_submission(a.submissions, _ctx("prj_b"), first.submission_id) is None assert a.invoker.calls[0]["project_id"] == "prj_a" conflicted = resolve_form_bindings(conflict_sources()) assert "contact" in conflicted.conflicts store = CompositionFormRuntimeStore("prj_x", conflicted) assert store.get("prj_x", "contact") is None assert publish_form_readiness( generated_form_keys=("contact",), resolution=conflicted ) == (FORM_BINDING_CONFLICT,) historical = first.submission_id rebind(a, ()) gone = browser_submit(a, form_key="contact", values=CONTACT_VALUES) assert gone.body["error"] == "form_unavailable" assert get_form_submission(a.submissions, _ctx("prj_a"), historical) is not None from src.domain.forms.e2e import contact_pack rebind(a, (contact_pack(notify_to_email="new@example.com"),)) restored = browser_submit(a, form_key="contact", values=CONTACT_VALUES) assert restored.body["ok"] is True runtime = a.store.get("prj_a", "contact") owner = next(action for action in runtime.plan.actions if action.key == "owner_notify") assert owner.constants["to"] == "new@example.com" old = get_form_submission(a.submissions, _ctx("prj_a"), historical) assert old.accepted.submission.values["message"] == CONTACT_VALUES["message"] def test_publish_readiness_and_unbound_malformed() -> None: app = install_contact_app() assert publish_form_readiness( generated_form_keys=("contact",), resolution=app.resolution ) == () assert publish_form_readiness( generated_form_keys=("missing",), resolution=app.resolution ) == ("form_unavailable",) missing = browser_submit(app, form_key="nonexistent", values=CONTACT_VALUES) assert missing.body["error"] == "form_unavailable" bad = FormBindingSource( form_key="contact", owner_id="forms.contact", owner_kind="app_block", pack_slug="contact-capture", pack_config={}, ) runtime = resolve_form_bindings((bad,)).runtimes["contact"] broken = ResolvedFormRuntime( definition=runtime.definition, plan=runtime.plan.model_copy( update={ "actions": ( runtime.plan.actions[0].model_copy( update={"mapping": {"email": "not_a_field"}} ), ) } ), persist_first=True, ) app.overlay.put(app.project_id, broken) unavailable = browser_submit(app, form_key="contact", values=CONTACT_VALUES) assert unavailable.body["error"] == "form_unavailable" assert "not_a_field" not in str(unavailable.body) def test_client_trust_boundary_spoofs_are_rejected() -> None: app = install_contact_app() for extra in ( {"project_id": "prj_other"}, {"provider_hint": "hubspot"}, {"workflow": "attacker_workflow"}, {"target": "https://evil.example"}, {"persist_first": False}, {"actions": [{"type": "email.send"}]}, {"client_identity": "spoof"}, {"definition": {"key": "x"}}, ): result = browser_submit( app, form_key="contact", values=CONTACT_VALUES, extra=extra ) assert result.body["error"] == "invalid_request" assert result.http_status == 400 twice = [ browser_submit(app, form_key="contact", values=CONTACT_VALUES) for _ in range(2) ] assert twice[0].submission_id != twice[1].submission_id quoted = browser_submit( app, form_key="contact", values={ "name": "Ada", "email": "ada@example.com", "message": "hi", "project_type": "repair", }, ) assert "unknown_field" in quoted.body["fields"]["project_type"] def test_generated_bundle_and_composition_identities() -> None: source = generated_browser_sources() assert "/api/forms/" in source assert "submitGeneratedForm" in source for token in ( "/capabilities/crm.", "/capabilities/email.send", "hubspot.create_contact", "persist_first", "P2HS_CORE_URL", ): assert token not in Path("src/domain/codegen/runtime/p2hs-forms-runtime.mjs").read_text( encoding="utf-8" ) runtime = Path("src/domain/codegen/runtime/p2hs-forms-runtime.mjs").read_text( encoding="utf-8" ) assert "retry" not in runtime.lower() or "Do not retry" in Path( "src/domain/codegen/forms_codegen.py" ).read_text(encoding="utf-8") assert FEATURE_PACK_ALIASES["forms.public_submission"] == "contact-capture" assert CATALOG_OVERLAYS["contact-capture"]["provides"][0].id == "form.public_submission" assert "form.standard" in known_contract_ids() assert classify_form_intent("embed HubSpot form") == "provider_escape" paths = [getattr(route, "path", "") for route in fastapi_app.routes] assert "/api/projects/{project_id}/forms/{form_key}/submissions" in paths methods = { tuple(sorted(route.methods or ())) for route in forms_router.routes if getattr(route, "path", "").endswith("/submissions") } assert methods == {("POST",)} src = inspect.getsource(LoggingMiddleware.dispatch) assert "email" not in src assert "request_id" in src assert browser_leak_hits(runtime) == () for token in BROWSER_LEAK_TOKENS: assert token not in runtime def test_request_snapshot_concurrency_and_stale_client() -> None: app = install_contact_app() counted = CountingStore(app.runtimes) submit_form( project_id=app.project_id, form_key="contact", values=CONTACT_VALUES, client_identity="203.0.113.10", runtimes=counted, invoker=app.invoker, submissions=app.submissions, files=app.files, verifier=app.verifier, limiter=app.limiter, ) assert counted.gets == 1 ids = [] def _one(): ids.append( browser_submit(app, form_key="contact", values=CONTACT_VALUES).submission_id ) with ThreadPoolExecutor(max_workers=4) as pool: list(pool.map(lambda _: _one(), range(8))) assert len(set(ids)) == 8 required = FormDefinition( key="contact", fields=[ FormField(key="name", type="text", required=True), FormField(key="email", type="email", required=True), FormField(key="phone", type="phone", required=False), FormField(key="message", type="textarea", required=True), FormField(key="company", type="text", required=True), ], ) current = app.store.get(app.project_id, "contact") app.overlay.put( app.project_id, ResolvedFormRuntime( definition=required, anti_abuse=current.anti_abuse, persist_first=True, ), ) stale = browser_submit(app, form_key="contact", values=CONTACT_VALUES) assert "required" in stale.body["fields"]["company"] extra = browser_submit( app, form_key="contact", values={**CONTACT_VALUES, "company": "Acme", "old_field": "x"}, ) assert "unknown_field" in extra.body["fields"]["old_field"]