diff --git a/efile_app/efile/migrations/0003_branch_aware_workflow.py b/efile_app/efile/migrations/0003_branch_aware_workflow.py new file mode 100644 index 0000000..af28837 --- /dev/null +++ b/efile_app/efile/migrations/0003_branch_aware_workflow.py @@ -0,0 +1,57 @@ +from django.db import migrations, models + + +def normalize_existing_case(apps, schema_editor): + FilingDraft = apps.get_model("efile", "FilingDraft") + FilingDraft.objects.filter(existing_case="no").update(existing_case="new") + FilingDraft.objects.filter(existing_case__in=["yes", "responding"]).update(existing_case="existing") + + +class Migration(migrations.Migration): + dependencies = [("efile", "0002_filing_drafts")] + + operations = [ + migrations.RunPython(normalize_existing_case, migrations.RunPython.noop), + migrations.AddField( + model_name="filingdraft", + name="workflow_version", + field=models.PositiveSmallIntegerField(default=1), + ), + migrations.AlterField( + model_name="filingdraft", + name="existing_case", + field=models.CharField( + blank=True, + choices=[("new", "New"), ("existing", "Existing"), ("unsure", "Unsure")], + max_length=20, + ), + ), + migrations.AlterField( + model_name="filingdraft", + name="current_step", + field=models.CharField( + choices=[ + ("options", "Options"), + ("filing_path", "Filing"), + ("upload_documents", "Upload documents"), + ("extraction_review", "Confirm filing"), + ("case_lookup", "Find your case"), + ("case_confirmation", "Confirm your case"), + ("document_checklist", "Check documents"), + ("organize_documents", "Organize documents"), + ("your_information", "Your information"), + ("parties", "People in this filing"), + ("party_details", "Person details"), + ("case_questions", "Case questions"), + ("payment", "Fees"), + ("review", "Review"), + ("confirmation", "Confirmation"), + ("upload_first", "Upload lead document"), + ("case_information", "Case information"), + ("documents", "Documents"), + ], + default="options", + max_length=64, + ), + ), + ] diff --git a/efile_app/efile/models.py b/efile_app/efile/models.py index 0b37f82..6195062 100644 --- a/efile_app/efile/models.py +++ b/efile_app/efile/models.py @@ -4,7 +4,7 @@ from django.db import models from django.utils import timezone -from efile.workflow import WorkflowStepKey, get_workflow_step_choices +from efile.workflow import ExistingCase, WorkflowStepKey, get_workflow_step_choices class UserProfile(AbstractUser): @@ -54,8 +54,13 @@ class Status(models.TextChoices): choices=get_workflow_step_choices(), default=WorkflowStepKey.OPTIONS, ) + workflow_version = models.PositiveSmallIntegerField(default=1) - existing_case = models.CharField(max_length=20, blank=True) + existing_case = models.CharField( + max_length=20, + choices=[(value.value, value.name.title()) for value in ExistingCase], + blank=True, + ) court_code = models.CharField(max_length=100, blank=True) court_name = models.CharField(max_length=255, blank=True) case_category_code = models.CharField(max_length=100, blank=True) diff --git a/efile_app/efile/services/drafts.py b/efile_app/efile/services/drafts.py index 0967151..00c9a50 100644 --- a/efile_app/efile/services/drafts.py +++ b/efile_app/efile/services/drafts.py @@ -13,7 +13,7 @@ from django.db.models import QuerySet from efile.models import FilingDocument, FilingDraft, FilingParty -from efile.workflow import WorkflowStepKey +from efile.workflow import WorkflowStepKey, legacy_existing_case_value, normalize_existing_case ACTIVE_DRAFT_STATUSES = (FilingDraft.Status.DRAFT, FilingDraft.Status.ERROR) # A draft mid-submission is still the user's current draft (so the submit flow can @@ -198,6 +198,8 @@ def write_case_data( if value is _MISSING: continue value = _as_str(value) + if field == "existing_case": + value = normalize_existing_case(value) if getattr(draft, field) != value: setattr(draft, field, value) update_fields.append(field) @@ -246,7 +248,9 @@ def read_case_data(draft: FilingDraft | None) -> dict[str, Any]: data: dict[str, Any] = {} _put(data, "jurisdiction", draft.jurisdiction) _put(data, "jurisdiction_id", draft.jurisdiction) - _put(data, "existing_case", draft.existing_case) + # Old screens still branch on yes/no. The durable value is normalized now; + # remove this translation when the last legacy screen is retired. + _put(data, "existing_case", legacy_existing_case_value(draft.existing_case)) _put(data, "court", draft.court_code) _put(data, "court_name", draft.court_name) _put(data, "case_category", draft.case_category_code) @@ -475,6 +479,7 @@ def draft_snapshot(draft: FilingDraft | None) -> dict[str, Any] | None: "jurisdiction": draft.jurisdiction, "status": draft.status, "current_step": draft.current_step, + "workflow_version": draft.workflow_version, "existing_case": draft.existing_case, "court_code": draft.court_code, "court_name": draft.court_name, diff --git a/efile_app/efile/tests/test_workflow.py b/efile_app/efile/tests/test_workflow.py index ed9ceac..ac7c548 100644 --- a/efile_app/efile/tests/test_workflow.py +++ b/efile_app/efile/tests/test_workflow.py @@ -1,117 +1,181 @@ +from types import SimpleNamespace + import pytest from django.urls import reverse from efile.workflow import ( FILING_WORKFLOW, + LEGACY_WORKFLOW, + ExistingCase, WorkflowStepKey, get_next_step, get_previous_step, get_resume_step_url, get_step, get_step_url, + get_visible_workflow, get_workflow_context, get_workflow_steps, + legacy_existing_case_value, + normalize_existing_case, ) -EXPECTED_WORKFLOW_KEYS = [ - WorkflowStepKey.OPTIONS, - WorkflowStepKey.UPLOAD_FIRST, - WorkflowStepKey.CASE_INFORMATION, - WorkflowStepKey.DOCUMENTS, - WorkflowStepKey.PAYMENT, - WorkflowStepKey.REVIEW, - WorkflowStepKey.CONFIRMATION, -] + +def draft(**overrides): + values = { + "current_step": WorkflowStepKey.FILING_PATH, + "workflow_version": 2, + "existing_case": ExistingCase.NEW, + "case_questions_required": False, + "parties": [], + } + values.update(overrides) + return SimpleNamespace(**values) + + +def keys(workflow): + return [step.key for step in workflow] + + +def test_target_workflow_declares_every_reorganized_screen(): + assert get_workflow_steps() == FILING_WORKFLOW + assert keys(FILING_WORKFLOW) == [ + WorkflowStepKey.OPTIONS, + WorkflowStepKey.FILING_PATH, + WorkflowStepKey.UPLOAD_DOCUMENTS, + WorkflowStepKey.EXTRACTION_REVIEW, + WorkflowStepKey.CASE_LOOKUP, + WorkflowStepKey.CASE_CONFIRMATION, + WorkflowStepKey.DOCUMENT_CHECKLIST, + WorkflowStepKey.ORGANIZE_DOCUMENTS, + WorkflowStepKey.YOUR_INFORMATION, + WorkflowStepKey.PARTIES, + WorkflowStepKey.PARTY_DETAILS, + WorkflowStepKey.CASE_QUESTIONS, + WorkflowStepKey.PAYMENT, + WorkflowStepKey.REVIEW, + WorkflowStepKey.CONFIRMATION, + ] + + +def test_legacy_drafts_keep_the_current_linear_route_during_migration(): + legacy_draft = draft(current_step=WorkflowStepKey.DOCUMENTS, workflow_version=1) + + assert get_visible_workflow(legacy_draft) == LEGACY_WORKFLOW + assert get_previous_step(WorkflowStepKey.PAYMENT, legacy_draft).key == WorkflowStepKey.DOCUMENTS + assert get_next_step(WorkflowStepKey.PAYMENT, legacy_draft).key == WorkflowStepKey.REVIEW @pytest.mark.parametrize( - ("step_key", "label"), + "shared_step", [ - (WorkflowStepKey.OPTIONS, "Options"), - (WorkflowStepKey.UPLOAD_FIRST, "Upload lead document"), - (WorkflowStepKey.CASE_INFORMATION, "Case information"), - (WorkflowStepKey.DOCUMENTS, "Documents"), - (WorkflowStepKey.PAYMENT, "Payment"), - (WorkflowStepKey.REVIEW, "Review"), - (WorkflowStepKey.CONFIRMATION, "Confirmation"), + WorkflowStepKey.OPTIONS, + WorkflowStepKey.PAYMENT, + WorkflowStepKey.REVIEW, + WorkflowStepKey.CONFIRMATION, ], ) -def test_get_step_returns_registered_step(step_key, label): - step = get_step(step_key) +def test_shared_steps_without_draft_context_default_to_legacy(shared_step): + assert get_visible_workflow(current_step=shared_step) == LEGACY_WORKFLOW - assert step.key == step_key - assert step.label == label +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("no", ExistingCase.NEW), + ("yes", ExistingCase.EXISTING), + ("responding", ExistingCase.EXISTING), + ("unsure", ExistingCase.UNSURE), + ("", ""), + ], +) +def test_existing_case_values_are_normalized(value, expected): + assert normalize_existing_case(value) == expected -def test_get_workflow_steps_returns_ordered_workflow(): - assert get_workflow_steps() == FILING_WORKFLOW - assert [step.key for step in get_workflow_steps()] == EXPECTED_WORKFLOW_KEYS +def test_normalized_case_values_can_be_read_by_legacy_clients(): + assert legacy_existing_case_value(ExistingCase.NEW) == "no" + assert legacy_existing_case_value(ExistingCase.EXISTING) == "yes" -def test_get_step_raises_key_error_for_invalid_step(): - with pytest.raises(KeyError): - get_step("invalid_step") +def test_new_case_skips_lookup_and_confirmation(): + new_case = draft(existing_case=ExistingCase.NEW) -def test_get_previous_step_returns_none_for_first_step(): - assert get_previous_step(WorkflowStepKey.OPTIONS) is None + assert WorkflowStepKey.CASE_LOOKUP not in keys(get_visible_workflow(new_case)) + assert WorkflowStepKey.CASE_CONFIRMATION not in keys(get_visible_workflow(new_case)) + assert get_next_step(WorkflowStepKey.EXTRACTION_REVIEW, new_case).key == WorkflowStepKey.DOCUMENT_CHECKLIST -def test_get_previous_step_returns_prior_step(): - previous_step = get_previous_step(WorkflowStepKey.CASE_INFORMATION) +def test_existing_case_uses_lookup_and_confirmation(): + existing_case = draft(existing_case=ExistingCase.EXISTING) - assert previous_step is not None - assert previous_step.key == WorkflowStepKey.UPLOAD_FIRST + assert WorkflowStepKey.CASE_LOOKUP in keys(get_visible_workflow(existing_case)) + assert WorkflowStepKey.CASE_CONFIRMATION in keys(get_visible_workflow(existing_case)) + assert get_next_step(WorkflowStepKey.EXTRACTION_REVIEW, existing_case).key == WorkflowStepKey.CASE_LOOKUP -def test_get_next_step_returns_following_step(): - next_step = get_next_step(WorkflowStepKey.CASE_INFORMATION) +def test_unsure_case_stays_on_extraction_review(): + unsure_case = draft(existing_case=ExistingCase.UNSURE) - assert next_step is not None - assert next_step.key == WorkflowStepKey.DOCUMENTS + assert get_next_step(WorkflowStepKey.EXTRACTION_REVIEW, unsure_case) is None -def test_get_next_step_returns_none_for_last_step(): - assert get_next_step(WorkflowStepKey.CONFIRMATION) is None +def test_party_details_only_appear_for_incomplete_parties(): + incomplete = SimpleNamespace(first_name="Ada", last_name="", organization_name="") + complete = SimpleNamespace(first_name="Ada", last_name="Lovelace", organization_name="") + assert WorkflowStepKey.PARTY_DETAILS in keys(get_visible_workflow(draft(parties=[incomplete]))) + assert WorkflowStepKey.PARTY_DETAILS not in keys(get_visible_workflow(draft(parties=[complete]))) + + +def test_case_questions_only_appear_when_required(): + assert WorkflowStepKey.CASE_QUESTIONS not in keys(get_visible_workflow(draft())) + assert WorkflowStepKey.CASE_QUESTIONS in keys(get_visible_workflow(draft(case_questions_required=True))) -def test_get_step_url_reverses_workflow_route(): - expected_url = reverse("payment", kwargs={"jurisdiction": "illinois"}) +def test_get_step_raises_key_error_for_invalid_step(): + with pytest.raises(KeyError): + get_step("invalid_step") + + +def test_get_step_url_reverses_an_available_route(): + expected_url = reverse("payment", kwargs={"jurisdiction": "illinois"}) assert get_step_url(WorkflowStepKey.PAYMENT, "illinois") == expected_url -def test_get_resume_step_url_returns_the_drafts_own_step(): +def test_resume_preserves_legacy_draft_routes(): expected_url = reverse("upload", kwargs={"jurisdiction": "illinois"}) - assert get_resume_step_url(WorkflowStepKey.DOCUMENTS, "illinois") == expected_url -def test_get_resume_step_url_skips_options_so_resuming_does_not_loop(): - # OPTIONS is the model default for older drafts; resuming there would just - # return the user to the page that offered to resume. +def test_resume_skips_options_for_pre_migration_drafts(): expected_url = reverse("upload_first", kwargs={"jurisdiction": "illinois"}) - assert get_resume_step_url(WorkflowStepKey.OPTIONS, "illinois") == expected_url -def test_get_resume_step_url_falls_back_for_an_unrecognised_step(): +def test_resume_falls_back_for_an_unrecognised_step(): expected_url = reverse("upload_first", kwargs={"jurisdiction": "illinois"}) - assert get_resume_step_url("a_step_that_was_removed", "illinois") == expected_url -def test_get_resume_step_url_returns_none_without_a_draft(): +def test_resume_returns_none_without_a_draft(): assert get_resume_step_url(None, "illinois") is None -def test_get_workflow_context_includes_current_previous_and_next_urls(): - context = get_workflow_context(WorkflowStepKey.PAYMENT, "illinois") - previous_url = reverse("upload", kwargs={"jurisdiction": "illinois"}) - next_url = reverse("case_review", kwargs={"jurisdiction": "illinois"}) +def test_workflow_context_uses_draft_branch_and_includes_stage_progress(): + legacy_draft = draft(current_step=WorkflowStepKey.PAYMENT, workflow_version=1) + context = get_workflow_context(WorkflowStepKey.PAYMENT, "illinois", legacy_draft) assert context["workflow_current_step"].key == WorkflowStepKey.PAYMENT assert context["workflow_previous_step"].key == WorkflowStepKey.DOCUMENTS assert context["workflow_next_step"].key == WorkflowStepKey.REVIEW - assert context["workflow_previous_url"] == previous_url - assert context["workflow_next_url"] == next_url + assert context["workflow_previous_url"] == reverse("upload", kwargs={"jurisdiction": "illinois"}) + assert context["workflow_next_url"] == reverse("case_review", kwargs={"jurisdiction": "illinois"}) + assert [stage.value for stage in context["workflow_stages"]] == [ + "filing", + "upload", + "confirm_case", + "organize_documents", + "fees", + "review", + ] diff --git a/efile_app/efile/views/expert_form.py b/efile_app/efile/views/expert_form.py index a77d3a8..87772f1 100644 --- a/efile_app/efile/views/expert_form.py +++ b/efile_app/efile/views/expert_form.py @@ -65,6 +65,6 @@ def efile_expert_form(request, jurisdiction): "missing_required_fields": not has_all_required, "missing_party_info": has_all_required and not has_party_info, } - context.update(get_workflow_context(WorkflowStepKey.CASE_INFORMATION, jurisdiction)) + context.update(get_workflow_context(WorkflowStepKey.CASE_INFORMATION, jurisdiction, filing_draft)) return render(request, "efile/expert_form.html", context) diff --git a/efile_app/efile/views/options.py b/efile_app/efile/views/options.py index 1916296..1ab4c1c 100644 --- a/efile_app/efile/views/options.py +++ b/efile_app/efile/views/options.py @@ -33,6 +33,6 @@ def efile_options(request, jurisdiction): "resume_url": get_resume_step_url(active_draft.current_step if active_draft else None, jurisdiction), "has_case_data": bool(case_data or active_draft), } - context.update(get_workflow_context(WorkflowStepKey.OPTIONS, jurisdiction)) + context.update(get_workflow_context(WorkflowStepKey.OPTIONS, jurisdiction, active_draft)) return render(request, "efile/options.html", context) diff --git a/efile_app/efile/views/payment.py b/efile_app/efile/views/payment.py index e5c90ab..746a39a 100644 --- a/efile_app/efile/views/payment.py +++ b/efile_app/efile/views/payment.py @@ -46,6 +46,6 @@ def efile_payment(request, jurisdiction): "case_data": case_data, "filing_draft": draft_snapshot(filing_draft), } - context.update(get_workflow_context(WorkflowStepKey.PAYMENT, jurisdiction)) + context.update(get_workflow_context(WorkflowStepKey.PAYMENT, jurisdiction, filing_draft)) return render(request, "efile/payment.html", context) diff --git a/efile_app/efile/views/review.py b/efile_app/efile/views/review.py index ce062d9..1753792 100644 --- a/efile_app/efile/views/review.py +++ b/efile_app/efile/views/review.py @@ -120,6 +120,6 @@ def case_review(request, jurisdiction): "document_type": friendly_document_type, }, } - context.update(get_workflow_context(WorkflowStepKey.REVIEW, jurisdiction)) + context.update(get_workflow_context(WorkflowStepKey.REVIEW, jurisdiction, filing_draft)) return render(request, "efile/review.html", context) diff --git a/efile_app/efile/views/upload.py b/efile_app/efile/views/upload.py index 2573f75..dce8dfd 100644 --- a/efile_app/efile/views/upload.py +++ b/efile_app/efile/views/upload.py @@ -66,6 +66,6 @@ def efile_upload(request, jurisdiction): "filing_type_raw": case_classification["filing_type"], "court_raw": case_classification["court"], } - context.update(get_workflow_context(WorkflowStepKey.DOCUMENTS, jurisdiction)) + context.update(get_workflow_context(WorkflowStepKey.DOCUMENTS, jurisdiction, filing_draft)) return render(request, "efile/upload.html", context) diff --git a/efile_app/efile/views/upload_first.py b/efile_app/efile/views/upload_first.py index 1a04ef0..950b55c 100644 --- a/efile_app/efile/views/upload_first.py +++ b/efile_app/efile/views/upload_first.py @@ -61,6 +61,6 @@ def efile_upload_first(request, jurisdiction): "name_sought_info": name_sought_info, "case_classification": case_classification, } - context.update(get_workflow_context(WorkflowStepKey.UPLOAD_FIRST, jurisdiction)) + context.update(get_workflow_context(WorkflowStepKey.UPLOAD_FIRST, jurisdiction, filing_draft)) return render(request, "efile/upload_first.html", context) diff --git a/efile_app/efile/workflow.py b/efile_app/efile/workflow.py index 2f4eddd..46e023b 100644 --- a/efile_app/efile/workflow.py +++ b/efile_app/efile/workflow.py @@ -1,143 +1,329 @@ -"""Central filing workflow registry. +"""Central filing workflow registry and branch-aware navigation. -Use FILING_WORKFLOW as the single high-level map of the filing flow. +The reorganized flow is stateful: new and existing cases take different paths, +party details may repeat, and case questions only appear when configured. Keep +those decisions here so templates and JavaScript do not each invent their own +redirect rules. -To add a step: -1. Add a WorkflowStepKey member for the new step. -2. Add the URL route and view. -3. Add a WorkflowStep entry in the desired position below. -4. Add get_workflow_context(WorkflowStepKey.YOUR_STEP, jurisdiction) to that view's context. -5. Update any navigation copy that mentions the surrounding steps. -6. Update efile/tests/test_workflow.py. - -To rearrange steps: -1. Reorder FILING_WORKFLOW. -2. Update affected labels, navigation copy, and workflow tests. - -This registry is intentionally linear for now. Future branching should be added -here after the durable filing draft model exists as the workflow state source. +``LEGACY_WORKFLOW`` remains temporarily available while the reorganized screens +land in stacked changes. A draft on a legacy step continues through the old flow; +as soon as it enters a reorganized step it uses ``FILING_WORKFLOW``. """ from dataclasses import dataclass from enum import StrEnum +from typing import Any +from django.db.models import Q from django.urls import reverse +class ExistingCase(StrEnum): + """Controlled vocabulary used by durable drafts and workflow branches.""" + + NEW = "new" + EXISTING = "existing" + UNSURE = "unsure" + + +LEGACY_EXISTING_CASE_VALUES = { + "no": ExistingCase.NEW, + "yes": ExistingCase.EXISTING, + "responding": ExistingCase.EXISTING, +} + + +def normalize_existing_case(value: str | ExistingCase | None) -> str: + """Normalize old yes/no values without making legacy clients branch incorrectly.""" + + if value in (None, ""): + return "" + normalized = str(value).strip().lower() + return str(LEGACY_EXISTING_CASE_VALUES.get(normalized, normalized)) + + +def legacy_existing_case_value(value: str | ExistingCase | None) -> str: + """Translate normalized state for the old screens during the migration.""" + + normalized = normalize_existing_case(value) + if normalized == ExistingCase.NEW: + return "no" + if normalized == ExistingCase.EXISTING: + return "yes" + return normalized + + +class WorkflowStage(StrEnum): + FILING = "filing" + UPLOAD = "upload" + CONFIRM_CASE = "confirm_case" + CHECK_DOCUMENTS = "check_documents" + ORGANIZE_DOCUMENTS = "organize_documents" + PEOPLE = "people" + FEES = "fees" + REVIEW = "review" + + class WorkflowStepKey(StrEnum): - """Stable identifiers for filing workflow steps.""" + """Stable identifiers for both reorganized and transitional workflow steps.""" OPTIONS = "options" - UPLOAD_FIRST = "upload_first" - CASE_INFORMATION = "case_information" - DOCUMENTS = "documents" + FILING_PATH = "filing_path" + UPLOAD_DOCUMENTS = "upload_documents" + EXTRACTION_REVIEW = "extraction_review" + CASE_LOOKUP = "case_lookup" + CASE_CONFIRMATION = "case_confirmation" + DOCUMENT_CHECKLIST = "document_checklist" + ORGANIZE_DOCUMENTS = "organize_documents" + YOUR_INFORMATION = "your_information" + PARTIES = "parties" + PARTY_DETAILS = "party_details" + CASE_QUESTIONS = "case_questions" PAYMENT = "payment" REVIEW = "review" CONFIRMATION = "confirmation" + # Removed after all screens have migrated. Keeping these values temporarily + # lets saved drafts and each independently reviewable stacked PR keep working. + UPLOAD_FIRST = "upload_first" + CASE_INFORMATION = "case_information" + DOCUMENTS = "documents" + @dataclass(frozen=True) class WorkflowStep: - """A single screen in the filing workflow.""" - key: WorkflowStepKey label: str url_name: str + stage: WorkflowStage FILING_WORKFLOW: tuple[WorkflowStep, ...] = ( - WorkflowStep(WorkflowStepKey.OPTIONS, "Options", "efile_options"), - WorkflowStep(WorkflowStepKey.UPLOAD_FIRST, "Upload lead document", "upload_first"), - WorkflowStep(WorkflowStepKey.CASE_INFORMATION, "Case information", "expert_form"), - WorkflowStep(WorkflowStepKey.DOCUMENTS, "Documents", "upload"), - WorkflowStep(WorkflowStepKey.PAYMENT, "Payment", "payment"), - WorkflowStep(WorkflowStepKey.REVIEW, "Review", "case_review"), - WorkflowStep(WorkflowStepKey.CONFIRMATION, "Confirmation", "filing_confirmation"), + WorkflowStep(WorkflowStepKey.OPTIONS, "Options", "efile_options", WorkflowStage.FILING), + WorkflowStep(WorkflowStepKey.FILING_PATH, "Filing", "filing_path", WorkflowStage.FILING), + WorkflowStep(WorkflowStepKey.UPLOAD_DOCUMENTS, "Upload documents", "upload_documents", WorkflowStage.UPLOAD), + WorkflowStep(WorkflowStepKey.EXTRACTION_REVIEW, "Confirm filing", "extraction_review", WorkflowStage.UPLOAD), + WorkflowStep(WorkflowStepKey.CASE_LOOKUP, "Find your case", "case_lookup", WorkflowStage.CONFIRM_CASE), + WorkflowStep( + WorkflowStepKey.CASE_CONFIRMATION, + "Confirm your case", + "case_confirmation", + WorkflowStage.CONFIRM_CASE, + ), + WorkflowStep( + WorkflowStepKey.DOCUMENT_CHECKLIST, + "Check documents", + "document_checklist", + WorkflowStage.CHECK_DOCUMENTS, + ), + WorkflowStep( + WorkflowStepKey.ORGANIZE_DOCUMENTS, + "Organize documents", + "organize_documents", + WorkflowStage.ORGANIZE_DOCUMENTS, + ), + WorkflowStep( + WorkflowStepKey.YOUR_INFORMATION, + "Your information", + "your_information", + WorkflowStage.PEOPLE, + ), + WorkflowStep(WorkflowStepKey.PARTIES, "People in this filing", "parties", WorkflowStage.PEOPLE), + WorkflowStep(WorkflowStepKey.PARTY_DETAILS, "Person details", "party_details", WorkflowStage.PEOPLE), + WorkflowStep(WorkflowStepKey.CASE_QUESTIONS, "Case questions", "case_questions", WorkflowStage.PEOPLE), + WorkflowStep(WorkflowStepKey.PAYMENT, "Fees", "payment", WorkflowStage.FEES), + WorkflowStep(WorkflowStepKey.REVIEW, "Review", "case_review", WorkflowStage.REVIEW), + WorkflowStep(WorkflowStepKey.CONFIRMATION, "Confirmation", "filing_confirmation", WorkflowStage.REVIEW), +) + +LEGACY_WORKFLOW: tuple[WorkflowStep, ...] = ( + WorkflowStep(WorkflowStepKey.OPTIONS, "Options", "efile_options", WorkflowStage.FILING), + WorkflowStep(WorkflowStepKey.UPLOAD_FIRST, "Upload lead document", "upload_first", WorkflowStage.UPLOAD), + WorkflowStep(WorkflowStepKey.CASE_INFORMATION, "Case information", "expert_form", WorkflowStage.CONFIRM_CASE), + WorkflowStep(WorkflowStepKey.DOCUMENTS, "Documents", "upload", WorkflowStage.ORGANIZE_DOCUMENTS), + WorkflowStep(WorkflowStepKey.PAYMENT, "Fees", "payment", WorkflowStage.FEES), + WorkflowStep(WorkflowStepKey.REVIEW, "Review", "case_review", WorkflowStage.REVIEW), + WorkflowStep(WorkflowStepKey.CONFIRMATION, "Confirmation", "filing_confirmation", WorkflowStage.REVIEW), ) +_STEPS_BY_KEY = {step.key: step for step in (*FILING_WORKFLOW, *LEGACY_WORKFLOW)} +_LEGACY_KEYS = {step.key for step in LEGACY_WORKFLOW} - { + WorkflowStepKey.OPTIONS, + WorkflowStepKey.PAYMENT, + WorkflowStepKey.REVIEW, + WorkflowStepKey.CONFIRMATION, +} + def get_workflow_steps() -> tuple[WorkflowStep, ...]: + """Return the complete target workflow for choices, docs, and tests.""" + return FILING_WORKFLOW def get_workflow_step_choices() -> tuple[tuple[str, str], ...]: - """Return Django model choices derived from the workflow registry.""" + """Return choices for target and temporarily supported legacy draft steps.""" - return tuple((step.key.value, step.label) for step in FILING_WORKFLOW) + return tuple((step.key.value, step.label) for step in _STEPS_BY_KEY.values()) def get_step(step_key: WorkflowStepKey | str) -> WorkflowStep: try: - return next(step for step in FILING_WORKFLOW if step.key == step_key) - except StopIteration as exc: + return _STEPS_BY_KEY[WorkflowStepKey(step_key)] + except (KeyError, ValueError) as exc: raise KeyError(f"Unknown workflow step: {step_key}") from exc -def get_step_index(step_key: WorkflowStepKey | str) -> int: - for index, step in enumerate(FILING_WORKFLOW): - if step.key == step_key: +def _draft_value(draft: Any | None, name: str, default: Any = None) -> Any: + return getattr(draft, name, default) if draft is not None else default + + +def _has_incomplete_parties(draft: Any | None) -> bool: + if draft is None: + return False + parties = getattr(draft, "parties", None) + if parties is None: + return bool(_draft_value(draft, "has_incomplete_parties", False)) + if hasattr(parties, "filter"): + incomplete = Q(organization_name="") & (Q(first_name="") | Q(last_name="")) + return parties.filter(incomplete).exists() + try: + party_list = list(parties.all()) + except (AttributeError, TypeError): + party_list = list(parties) + return any(not (party.organization_name or (party.first_name and party.last_name)) for party in party_list) + + +def _has_case_questions(draft: Any | None) -> bool: + if draft is None: + return False + explicit = _draft_value(draft, "case_questions_required", None) + if explicit is not None: + return bool(explicit) + return bool((_draft_value(draft, "supplemental_fields", {}) or {}).get("_case_questions_required")) + + +def _uses_legacy_workflow(current_step: WorkflowStepKey | str | None, draft: Any | None) -> bool: + raw_step = current_step or _draft_value(draft, "current_step") + try: + key = WorkflowStepKey(raw_step) + except (TypeError, ValueError): + return False + if key in _LEGACY_KEYS: + return True + if draft is None and key in { + WorkflowStepKey.OPTIONS, + WorkflowStepKey.PAYMENT, + WorkflowStepKey.REVIEW, + WorkflowStepKey.CONFIRMATION, + }: + return True + if draft is not None: + return int(_draft_value(draft, "workflow_version", 1)) < 2 + return False + + +def get_visible_workflow( + draft: Any | None = None, + *, + current_step: WorkflowStepKey | str | None = None, +) -> tuple[WorkflowStep, ...]: + """Resolve the screens visible for this draft's branch.""" + + if _uses_legacy_workflow(current_step, draft): + return LEGACY_WORKFLOW + + existing_case = normalize_existing_case(_draft_value(draft, "existing_case")) + current_key = None + try: + current_key = WorkflowStepKey(current_step or _draft_value(draft, "current_step")) + except (TypeError, ValueError): + pass + + visible: list[WorkflowStep] = [] + for step in FILING_WORKFLOW: + if step.key in {WorkflowStepKey.CASE_LOOKUP, WorkflowStepKey.CASE_CONFIRMATION}: + if existing_case != ExistingCase.EXISTING and step.key != current_key: + continue + if step.key == WorkflowStepKey.PARTY_DETAILS: + if not _has_incomplete_parties(draft) and step.key != current_key: + continue + if step.key == WorkflowStepKey.CASE_QUESTIONS: + if not _has_case_questions(draft) and step.key != current_key: + continue + visible.append(step) + return tuple(visible) + + +def get_step_index( + step_key: WorkflowStepKey | str, + draft: Any | None = None, +) -> int: + workflow = get_visible_workflow(draft, current_step=step_key) + key = WorkflowStepKey(step_key) + for index, step in enumerate(workflow): + if step.key == key: return index raise KeyError(f"Unknown workflow step: {step_key}") -def get_previous_step(step_key: WorkflowStepKey | str) -> WorkflowStep | None: - index = get_step_index(step_key) +def get_previous_step(step_key: WorkflowStepKey | str, draft: Any | None = None) -> WorkflowStep | None: + workflow = get_visible_workflow(draft, current_step=step_key) + index = get_step_index(step_key, draft) if index == 0: return None - return FILING_WORKFLOW[index - 1] + return workflow[index - 1] + +def get_next_step(step_key: WorkflowStepKey | str, draft: Any | None = None) -> WorkflowStep | None: + key = WorkflowStepKey(step_key) + if key == WorkflowStepKey.EXTRACTION_REVIEW: + existing_case = normalize_existing_case(_draft_value(draft, "existing_case")) + if existing_case == ExistingCase.NEW: + return get_step(WorkflowStepKey.DOCUMENT_CHECKLIST) + if existing_case == ExistingCase.EXISTING: + return get_step(WorkflowStepKey.CASE_LOOKUP) + return None -def get_next_step(step_key: WorkflowStepKey | str) -> WorkflowStep | None: - index = get_step_index(step_key) + workflow = get_visible_workflow(draft, current_step=key) + index = get_step_index(key, draft) try: - return FILING_WORKFLOW[index + 1] + return workflow[index + 1] except IndexError: return None def get_step_url(step_key: WorkflowStepKey | str, jurisdiction: str) -> str: - step = get_step(step_key) - return reverse(step.url_name, kwargs={"jurisdiction": jurisdiction}) + return reverse(get_step(step_key).url_name, kwargs={"jurisdiction": jurisdiction}) def get_resume_step_url(current_step: WorkflowStepKey | str | None, jurisdiction: str) -> str | None: - """Return the URL to send someone back to when they resume a draft. - - OPTIONS is the model default for drafts saved before ``current_step`` was - tracked, but resuming there just returns the user to the page offering to - resume. Those start at the first real filing step instead. An unrecognised - value is treated the same way rather than breaking the options page. - """ if current_step is None: return None - try: step_key = WorkflowStepKey(current_step) except ValueError: step_key = WorkflowStepKey.UPLOAD_FIRST - if step_key == WorkflowStepKey.OPTIONS: step_key = WorkflowStepKey.UPLOAD_FIRST - return get_step_url(step_key, jurisdiction) -def get_workflow_context(current_step: WorkflowStepKey | str, jurisdiction: str) -> dict: - previous_step = get_previous_step(current_step) - next_step = get_next_step(current_step) - previous_url = None - next_url = None - - if previous_step: - previous_url = get_step_url(previous_step.key, jurisdiction) - if next_step: - next_url = get_step_url(next_step.key, jurisdiction) +def get_workflow_context( + current_step: WorkflowStepKey | str, + jurisdiction: str, + draft: Any | None = None, +) -> dict[str, Any]: + previous_step = get_previous_step(current_step, draft) + next_step = get_next_step(current_step, draft) + visible_workflow = get_visible_workflow(draft, current_step=current_step) return { - "workflow_steps": get_workflow_steps(), + "workflow_steps": visible_workflow, + "workflow_stages": tuple(dict.fromkeys(step.stage for step in visible_workflow)), "workflow_current_step": get_step(current_step), "workflow_previous_step": previous_step, "workflow_next_step": next_step, - "workflow_previous_url": previous_url, - "workflow_next_url": next_url, + "workflow_previous_url": get_step_url(previous_step.key, jurisdiction) if previous_step else None, + "workflow_next_url": get_step_url(next_step.key, jurisdiction) if next_step else None, }