GovOPlaN/govoplan
2
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat: prove assisted resident permit intake
Dependency Audit / dependency-audit (push) Successful in 1m40s
Details
Deployment Installer / deployment-installer (push) Successful in 6s
Details
Security Audit / security-audit (push) Successful in 10m37s
Details

Browse Source
This commit is contained in:
zemion
2026-08-19 02:45:53 +02:00
parent e2f505eeab
commit 69519a92b4
6 changed files with 211 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_institutional_service_journey.py
+162 -1
View File
@@ -44,11 +44,16 @@ from govoplan_forms.backend.service import (
record_form_definition,
)
from govoplan_forms_runtime.backend.db.models import (
FormAssistedConfirmation,
FormInstanceEvent,
FormInstanceIdentity,
FormInstanceRevision,
FormIntakeProfile,
FormIntakeSession,
)
from govoplan_forms_runtime.backend.intake import FormIntakeService
from govoplan_forms_runtime.backend.service import (
FormRuntimeError,
FormRuntimeService,
FormsServiceLauncher,
)
@@ -243,7 +248,7 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
def test_reference_fixture_names_remaining_manual_target_evidence(self) -> None:
self.assertEqual("Anwohnerparkausweis", JOURNEY["title_de"])
self.assertEqual("de-DE", JOURNEY["locale"])
self.assertEqual(5, len(JOURNEY["acceptance"]["automated"]))
self.assertEqual(6, len(JOURNEY["acceptance"]["automated"]))
self.assertEqual(5, len(JOURNEY["acceptance"]["manual_or_target"]))
def test_portal_launches_exact_form_revision_and_persists_submission(self) -> None:
@@ -378,6 +383,162 @@ class InstitutionalServiceJourneyTests(unittest.TestCase):
session.close()
engine.dispose()
def test_assisted_intake_reuses_exact_form_and_persists_readback_provenance(
self,
) -> None:
engine = create_engine("sqlite+pysqlite:///:memory:")
for table in (
FormDefinitionRevision.__table__,
FormInstanceIdentity.__table__,
FormInstanceRevision.__table__,
FormInstanceEvent.__table__,
FormIntakeProfile.__table__,
FormIntakeSession.__table__,
FormAssistedConfirmation.__table__,
):
table.create(engine)
sessions = sessionmaker(bind=engine)
principal = _Principal()
assisted = JOURNEY["assisted_intake"]
try:
with sessions() as session:
form = record_form_definition(
session,
principal,
definition=FormDefinition(
reference=InstitutionalReference(
kind="form",
owner_module="forms",
object_id=JOURNEY["form"]["object_id"],
tenant_id="tenant-1",
version=JOURNEY["form"]["version"],
),
key=JOURNEY["form"]["object_id"],
temporal=TemporalRevision(
revision=JOURNEY["form"]["version"],
recorded_at=NOW - timedelta(days=2),
change_reason="Publish the resident parking permit application.",
),
title=JOURNEY["title"],
fields=tuple(
FormFieldDefinition(
key=key,
label=key.replace("_", " ").title(),
required=True,
constraints={"min_length": 2},
)
for key in JOURNEY["form"]["fields"]
),
publication_state="published",
allow_drafts=True,
handoff_kinds=("case",),
),
)
registry = _FormRegistry(_service())
intake = FormIntakeService(registry)
profile = intake.create_profile(
session,
principal,
definition_ref=form.reference,
mode="assisted",
custodian_ref=assisted["responsible_function_ref"],
recorded_at=NOW,
)
started = intake.start_assisted(
session,
principal,
profile_id=profile.profile_id,
values=JOURNEY["form"]["fields"],
channel=assisted["channel"],
affected_party_ref=assisted["affected_party_ref"],
represented_party_ref=assisted["represented_party_ref"],
authority_basis=assisted["authority_basis"],
purpose=assisted["purpose"],
legal_basis_ref=assisted["legal_basis_ref"],
consent_basis=assisted["consent_basis"],
notice_given=assisted["notice_given"],
responsible_function_ref=assisted["responsible_function_ref"],
language=assisted["language"],
accessibility_needs=assisted["accessibility_needs"],
field_sources={
key: {
"source": "person_statement",
"confidence": "stated",
"declared_by_ref": assisted["affected_party_ref"],
}
for key in JOURNEY["form"]["fields"]
},
idempotency_key="resident-permit-assisted-start",
recorded_at=NOW + timedelta(minutes=1),
)
self.assertEqual(form.reference, started.instance.definition_ref)
self.assertEqual(JOURNEY["form"]["fields"], started.instance.values)
self.assertEqual(
assisted["purpose"], started.instance.metadata["intake"]["purpose"]
)
instance_id = started.instance.instance_id
session.commit()
with sessions() as resumed:
runtime = FormRuntimeService(registry)
current = runtime.get_instance(
resumed,
principal,
instance_id=instance_id,
)
self.assertIsNotNone(current)
with self.assertRaisesRegex(FormRuntimeError, "read-back confirmation"):
runtime.submit_instance(
resumed,
principal,
instance_id=instance_id,
expected_revision=current.revision,
values=current.values,
attachment_refs=(),
signature_refs=(),
idempotency_key="resident-permit-assisted-unconfirmed",
recorded_at=NOW + timedelta(minutes=2),
)
confirmation = FormIntakeService(registry).record_assisted_confirmation(
resumed,
principal,
instance_id=instance_id,
expected_revision=current.revision,
values=current.values,
attachment_refs=(),
signature_refs=(),
outcome=assisted["confirmation_outcome"],
method=assisted["confirmation_method"],
confirmed_by_ref=assisted["affected_party_ref"],
confirmed_at=NOW + timedelta(minutes=3),
idempotency_key="resident-permit-assisted-readback",
field_sources={
key: {
"source": "person_statement",
"confidence": "stated",
"declared_by_ref": assisted["affected_party_ref"],
}
for key in JOURNEY["form"]["fields"]
},
)
submitted = runtime.submit_instance(
resumed,
principal,
instance_id=instance_id,
expected_revision=current.revision,
values=current.values,
attachment_refs=(),
signature_refs=(),
idempotency_key="resident-permit-assisted-submit",
recorded_at=NOW + timedelta(minutes=4),
)
resumed.commit()
self.assertEqual("submitted", submitted.status)
self.assertEqual(current.revision, confirmation.instance_revision)
self.assertEqual(assisted["affected_party_ref"], confirmation.confirmed_by_ref)
finally:
engine.dispose()
def test_workflow_handoff_survives_session_reopen_and_projects_into_tasks(
self,
) -> None: