TSD-9273 manual QA notes for any executed-unit matrix case

README.md

TSD-9273 manual QA guide for any executed-unit matrix case

Manual QA guide for reproducing one chosen executed-unit matrix case on a TS, entering the requested CZ upgrade flow, and comparing the real DecisionFlowLog against the expected units.

References

• TSD-9273
• PR #22039

Test Server

• Namespace: replace-via-cli

Prerequisites

Apply the required TS settings before starting the flow:

twisto ts settings replace-via-cli

Paste or merge this payload in the editor:

overrides:
  USE_DEV_DECISION_FLOW_SPECIFICATIONS: false
  CREDIT_NAVIGATOR_WSDL: os.path.join(PROJECT_ROOT, "resources", "credit_navigator",
    "mock_api.wsdl")
  EMAIL_OFFLINE_VALIDATION: true
  DISABLED_KOS: '["FOREIGNER"]'

Manual Flow

Use the steps below as the primary QA guide. Backend automation snippets are included as an exact reference and can also be executed through the shared TS runner.

1. Choose the matrix case and print the expected result

Export the matrix dimensions for the case you want to test, then run the helper below. It prints the expected config.name, the expected executed units, and the case-specific notes.

Frontend / API steps:

• Choose MATRIX_MODE as baseline or patched.
• Choose REGISTRATION_KIND as new or existing.
• Choose FLOW as classic, spi, split, or bankid.
• Choose NRKI and PREPEND as 0 or 1.
• On patched new cases with PREPEND=1, also choose REPI as 0 or 1.
• On patched new cases with PREPEND=0, expect the legacy flow shape; REPI does not change the executed-unit sequence there.
• If NRKI=1 and you want a prepend-sensitive case, choose SCORE_RANGE=F for good score range or SCORE_RANGE=A for non-good score range.

Shell command (scripts/print_case_reference.sh):

#!/usr/bin/env bash
set -euo pipefail

python3 tmp/tsd-9273/playbooks/manual-qa/scripts/print_case_reference.py

2. Set the feature flags for the chosen case and flow entry

Apply the chosen matrix flag values and the entry-path flags in Django Admin before you start or rerun the registration.

Django Admin checks:

• Open Django Admin → Features → Features.
• Set NRKI according to the chosen case.
• Set NRKI_PREPEND_FOR_GOOD_SCORE_RANGE according to the chosen case.
• On patched new cases with PREPEND=1, set CZ_REPI_ENABLED according to the chosen case.
• On baseline, on existing, or on patched new with PREPEND=0, you can leave CZ_REPI_ENABLED unchanged because it does not affect the executed-unit sequence there.
• For classic, set CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=0 and CZ_BANKID_REGISTRATION_ENABLED=0.
• For spi, set CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=0 and CZ_BANKID_REGISTRATION_ENABLED=0.
• For split, set CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=100 and CZ_BANKID_REGISTRATION_ENABLED=0.
• For bankid, set CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=100 and CZ_BANKID_REGISTRATION_ENABLED=100.
• Use a unique email address and a unique CZ phone number for every registration run.
• On TS, use OTP 123456 unless the page shows a different generated code.
• Use normal adult CZ customer data, for example name TSD 9273 Tester, address Sokolovská 47/73, Praha, 18600, income 20000, expenses 1000, and loan repayments 2000.

3. Create or rerun the registration in the target flow

Use the entry path for the selected FLOW, then create a new registration or rerun an existing one.

Frontend / API steps:

• For classic, open the legacy CZ registration path, for example /registrace/.
• For classic, complete the registration with unique email, phone, adult CZ personal ID, CZ citizenship, CZ address, income, expenses, loan repayments, education, family status, and non-PEP value.
• For classic, let the upgrade scoring check be created; if the UI stops before scoring execution, open the related upgrade request or check in Django Admin and run the scoring action normally used for classic registrations.
• For classic, later confirm that the stored config.name is CZUpgradeFlowSpecification or CZUpgradeFlowSpecificationV2, depending on the case.
• For spi, open the flow-app onboarding path, for example /start/onboarding/.
• For spi, submit a unique email address and the email OTP.
• For spi, continue from the additional-info intro, submit a unique CZ phone number, and submit the SMS OTP.
• For spi, submit password 1qaz@WSX, name TSD 9273 Tester, and address Sokolovská 47/73, Praha, 18600.
• For spi, accept CZ account agreements with termsPrivacy=true, nrki=true, telco=true, marketing=false, isPoliticallyExposed=false.
• For spi, submit country of birth CZ, a valid adult CZ personal ID, education Primary, family status Single, children None, income type Employment, employment type 1, monthly income 20000, monthly expenses 1000, and monthly loan repayment 2000.
• For spi, continue until RiskCheckProcessingStep is reached and the upgrade scoring check exists.
• For spi, later confirm SPICZUpgradeFlowSpecification or SPICZUpgradeFlowSpecificationV2.
• For split, open the flow-app onboarding path, for example /start/onboarding/.
• For split, submit a unique email address and the email OTP.
• For split, continue from the additional-info intro, submit a unique CZ phone number, and submit the SMS OTP.
• For split, submit password 1qaz@WSX, name TSD 9273 Tester, and address Sokolovská 47/73, Praha, 18600.
• For split, continue until RiskCheckProcessingStep is reached and the upgrade scoring check exists.
• For split, if the page proceeds through CZ account agreements instead of the step-by-step risk checkpoint, discard the run and re-check that CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=100 was active before the run started.
• For split, later confirm SplittedCZUpgradeFlowSpecification or SplittedCZUpgradeFlowSpecificationV2.
• For bankid, open the flow-app onboarding path, for example /start/onboarding/.
• For bankid, submit a unique email address and the email OTP.
• For bankid, choose BankID when the BankID intro step is shown.
• For bankid, complete BankID authentication through the available TS sandbox or mock path.
• For bankid, wait until the flow reaches BankID confirmation and confirm the BankID customer data.
• For bankid, submit BankID consent with termsPrivacy=true, nrki=true, telco=true, marketing=false.
• For bankid, submit a unique CZ phone number and the SMS OTP.
• For bankid, submit password 1qaz@WSX and education Primary.
• For bankid, continue until RiskCheckProcessingStep is reached and the upgrade scoring check exists.
• For bankid, confirm in Django Admin → Customer → Bank ID connection logs that the related session has a successful BankIdConnectionLog. Without it this run is not a BankID flow verification.
• For bankid, later confirm BankIdUpgradeFlowSpecification or BankIdUpgradeFlowSpecificationV2.
• If you hit the wrong flow, discard the run and retry through the correct entry path.
• For new, start a fresh registration.
• On patched new, PREPEND=1 should resolve to a V2 config; PREPEND=0 should stay on the legacy config.
• For existing on baseline, create the first run, keep the same request, then trigger a rerun of that request.
• For existing on patched, first create the first run with PREPEND=0 so the first stored flow stays legacy, then switch flags to the target case and rerun the same request.
• The created Check must be production-like: is_production=True. If the resulting DecisionFlowLog points to check.is_production=False, discard that run and create the case through a production-key path.
• When the case depends on SCORE_RANGE, choose the Credit Navigator mock sample that yields the chosen score range and verify the real CBScoreRange afterwards in the stored NRKI response.

4. Inspect the stored DecisionFlowLog

Read the real executed and skipped units from the chosen DecisionFlowLog and compare them to the expected case from step 2.

Shell command (<inline>):

DECISION_FLOW_LOG_ID=<decision-flow-log-id>
python3 ../environment/tripping-avenger/tools/ts_testing/ts_django_shell_runner.py <ts-namespace> --script tmp/tsd-9273/playbooks/manual-qa/scripts/inspect_decision_flow_log.py --var DECISION_FLOW_LOG_ID=${DECISION_FLOW_LOG_ID} --pretty

Django Admin checks:

• Open Django Admin → Scoring → Decision flow logs.
• Open the relevant DecisionFlowLog.
• Copy its numeric id into DECISION_FLOW_LOG_ID.
• Confirm that the related Check has is_production=True.
• Confirm that config.name matches the expected config from step 2.
• Run the helper below from the repository root and replace <ts-namespace> plus <decision-flow-log-id> with the tested values.
• Compare executed_units from the helper output to expected_executed_units from step 1.
• On patched new V2 cases, treat skipped_units as informative only; the matrix is defined by executed units.
• If the case depends on SCORE_RANGE, also confirm the stored CBScoreRange in the related NRKI response.

5. Optional automated production matrix smoke

Run this helper when you need a quick TS proof that a production-like patched new/classic/NRKI=1/PREPEND=1/REPI=0/SCORE_RANGE=A case reaches the registry units and matches the expected matrix. The helper isolates unrelated pre-registry TS blockers so the check focuses on the NRKI/REPI unit order.

Shell command (<inline>):

python3 ../environment/tripping-avenger/tools/ts_testing/ts_django_shell_runner.py <ts-namespace> --script tmp/tsd-9273/playbooks/manual-qa/scripts/verify_production_matrix_smoke.py --pretty --exec-timeout-seconds 900

inspect_decision_flow_log.py

from __future__ import annotations

import json
import os

from apps.scoring.models import DecisionFlowLog


RELEVANT_UNITS = {
    "NRKIFetcher",
    "NRKIChecker",
    "InsolvencyChecker",
    "REPIFetcher",
    "REPIChecker",
    "ExecutionChecker",
    "SolusChecker",
}


def short_unit_name(full_name: str) -> str:
    base = full_name.split(":", 1)[0].split("@", 1)[0]
    return base.rsplit(".", 1)[-1]


flow_log_id = int(os.environ.get("DECISION_FLOW_LOG_ID", "__DECISION_FLOW_LOG_ID__"))
flow_log = DecisionFlowLog.objects.get(pk=flow_log_id)

unit_logs = list(flow_log.unit_logs.filter(date_rolled_back__isnull=True).order_by("phase", "id"))
relevant_logs = [
    {
        "unit": short_unit_name(unit_log.unit.name),
        "status": unit_log.get_status_display(),
        "phase": unit_log.phase,
        "unit_config_name": unit_log.unit.name,
    }
    for unit_log in unit_logs
    if short_unit_name(unit_log.unit.name) in RELEVANT_UNITS
]

executed_units = [row["unit"] for row in relevant_logs if row["status"].upper() != "SKIPPED"]
skipped_units = [row["unit"] for row in relevant_logs if row["status"].upper() == "SKIPPED"]

print(
    json.dumps(
        {
            "ok": True,
            "decision_flow_log_id": flow_log.id,
            "config_name": flow_log.config.name,
            "execution_status": flow_log.get_execution_status_display() if flow_log.execution_status is not None else None,
            "executed_units": executed_units,
            "skipped_units": skipped_units,
            "relevant_unit_logs": relevant_logs,
        },
        indent=2,
    )
)

playbook.yaml

version: 1

meta:
  id: tsd-9273-manual-qa-matrix-case
  title: TSD-9273 manual QA guide for any executed-unit matrix case
  owner: glutexo
  tags:
    - scoring
    - nrki
    - tsd-9273
    - ts
    - qa
  summary: Manual QA guide for reproducing one chosen executed-unit matrix case on a TS, entering the requested CZ upgrade flow, and comparing the real `DecisionFlowLog` against the expected units.

ts:
  namespace: replace-via-cli
  settings:
    overrides:
      USE_DEV_DECISION_FLOW_SPECIFICATIONS: false
      CREDIT_NAVIGATOR_WSDL: os.path.join(PROJECT_ROOT, "resources", "credit_navigator", "mock_api.wsdl")
      EMAIL_OFFLINE_VALIDATION: true
      DISABLED_KOS: '["FOREIGNER"]'

artifacts:
  print_case_reference_py: scripts/print_case_reference.py
  inspect_decision_flow_log_py: scripts/inspect_decision_flow_log.py
  verify_production_matrix_smoke_py: scripts/verify_production_matrix_smoke.py

steps:
  - id: choose_case
    kind: note
    title: Choose the matrix case and print the expected result
    description: Export the matrix dimensions for the case you want to test, then run the helper below. It prints the expected `config.name`, the expected executed units, and the case-specific notes.
    frontend_manual:
      - Choose `MATRIX_MODE` as `baseline` or `patched`.
      - Choose `REGISTRATION_KIND` as `new` or `existing`.
      - Choose `FLOW` as `classic`, `spi`, `split`, or `bankid`.
      - Choose `NRKI` and `PREPEND` as `0` or `1`.
      - On patched `new` cases with `PREPEND=1`, also choose `REPI` as `0` or `1`.
      - On patched `new` cases with `PREPEND=0`, expect the legacy flow shape; `REPI` does not change the executed-unit sequence there.
      - If `NRKI=1` and you want a prepend-sensitive case, choose `SCORE_RANGE=F` for good score range or `SCORE_RANGE=A` for non-good score range.
    shell:
      file: scripts/print_case_reference.sh

  - id: set_feature_flags
    kind: admin_check
    title: Set the feature flags for the chosen case and flow entry
    description: Apply the chosen matrix flag values and the entry-path flags in Django Admin before you start or rerun the registration.
    admin_manual:
      - Open Django Admin → Features → Features.
      - Set `NRKI` according to the chosen case.
      - Set `NRKI_PREPEND_FOR_GOOD_SCORE_RANGE` according to the chosen case.
      - On patched `new` cases with `PREPEND=1`, set `CZ_REPI_ENABLED` according to the chosen case.
      - On baseline, on `existing`, or on patched `new` with `PREPEND=0`, you can leave `CZ_REPI_ENABLED` unchanged because it does not affect the executed-unit sequence there.
      - For `classic`, set `CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=0` and `CZ_BANKID_REGISTRATION_ENABLED=0`.
      - For `spi`, set `CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=0` and `CZ_BANKID_REGISTRATION_ENABLED=0`.
      - For `split`, set `CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=100` and `CZ_BANKID_REGISTRATION_ENABLED=0`.
      - For `bankid`, set `CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=100` and `CZ_BANKID_REGISTRATION_ENABLED=100`.
      - Use a unique email address and a unique CZ phone number for every registration run.
      - On TS, use OTP `123456` unless the page shows a different generated code.
      - Use normal adult CZ customer data, for example name `TSD 9273 Tester`, address `Sokolovská 47/73, Praha, 18600`, income `20000`, expenses `1000`, and loan repayments `2000`.

  - id: drive_registration
    kind: frontend_manual
    title: Create or rerun the registration in the target flow
    description: Use the entry path for the selected `FLOW`, then create a new registration or rerun an existing one.
    frontend_manual:
      - For `classic`, open the legacy CZ registration path, for example `/registrace/`.
      - For `classic`, complete the registration with unique email, phone, adult CZ personal ID, CZ citizenship, CZ address, income, expenses, loan repayments, education, family status, and non-PEP value.
      - For `classic`, let the upgrade scoring check be created; if the UI stops before scoring execution, open the related upgrade request or check in Django Admin and run the scoring action normally used for classic registrations.
      - For `classic`, later confirm that the stored `config.name` is `CZUpgradeFlowSpecification` or `CZUpgradeFlowSpecificationV2`, depending on the case.
      - For `spi`, open the flow-app onboarding path, for example `/start/onboarding/`.
      - For `spi`, submit a unique email address and the email OTP.
      - For `spi`, continue from the additional-info intro, submit a unique CZ phone number, and submit the SMS OTP.
      - For `spi`, submit password `1qaz@WSX`, name `TSD 9273 Tester`, and address `Sokolovská 47/73, Praha, 18600`.
      - For `spi`, accept CZ account agreements with `termsPrivacy=true`, `nrki=true`, `telco=true`, `marketing=false`, `isPoliticallyExposed=false`.
      - For `spi`, submit country of birth `CZ`, a valid adult CZ personal ID, education `Primary`, family status `Single`, children `None`, income type `Employment`, employment type `1`, monthly income `20000`, monthly expenses `1000`, and monthly loan repayment `2000`.
      - For `spi`, continue until `RiskCheckProcessingStep` is reached and the upgrade scoring check exists.
      - For `spi`, later confirm `SPICZUpgradeFlowSpecification` or `SPICZUpgradeFlowSpecificationV2`.
      - For `split`, open the flow-app onboarding path, for example `/start/onboarding/`.
      - For `split`, submit a unique email address and the email OTP.
      - For `split`, continue from the additional-info intro, submit a unique CZ phone number, and submit the SMS OTP.
      - For `split`, submit password `1qaz@WSX`, name `TSD 9273 Tester`, and address `Sokolovská 47/73, Praha, 18600`.
      - For `split`, continue until `RiskCheckProcessingStep` is reached and the upgrade scoring check exists.
      - For `split`, if the page proceeds through CZ account agreements instead of the step-by-step risk checkpoint, discard the run and re-check that `CZ_MULTISTEP_RISK_ONBOARDING_ENABLED=100` was active before the run started.
      - For `split`, later confirm `SplittedCZUpgradeFlowSpecification` or `SplittedCZUpgradeFlowSpecificationV2`.
      - For `bankid`, open the flow-app onboarding path, for example `/start/onboarding/`.
      - For `bankid`, submit a unique email address and the email OTP.
      - For `bankid`, choose BankID when the BankID intro step is shown.
      - For `bankid`, complete BankID authentication through the available TS sandbox or mock path.
      - For `bankid`, wait until the flow reaches BankID confirmation and confirm the BankID customer data.
      - For `bankid`, submit BankID consent with `termsPrivacy=true`, `nrki=true`, `telco=true`, `marketing=false`.
      - For `bankid`, submit a unique CZ phone number and the SMS OTP.
      - For `bankid`, submit password `1qaz@WSX` and education `Primary`.
      - For `bankid`, continue until `RiskCheckProcessingStep` is reached and the upgrade scoring check exists.
      - For `bankid`, confirm in Django Admin → Customer → Bank ID connection logs that the related session has a successful `BankIdConnectionLog`. Without it this run is not a BankID flow verification.
      - For `bankid`, later confirm `BankIdUpgradeFlowSpecification` or `BankIdUpgradeFlowSpecificationV2`.
      - If you hit the wrong flow, discard the run and retry through the correct entry path.
      - For `new`, start a fresh registration.
      - On patched `new`, `PREPEND=1` should resolve to a `V2` config; `PREPEND=0` should stay on the legacy config.
      - For `existing` on baseline, create the first run, keep the same request, then trigger a rerun of that request.
      - For `existing` on patched, first create the first run with `PREPEND=0` so the first stored flow stays legacy, then switch flags to the target case and rerun the same request.
      - "The created `Check` must be production-like: `is_production=True`. If the resulting `DecisionFlowLog` points to `check.is_production=False`, discard that run and create the case through a production-key path."
      - When the case depends on `SCORE_RANGE`, choose the Credit Navigator mock sample that yields the chosen score range and verify the real `CBScoreRange` afterwards in the stored NRKI response.

  - id: inspect_flow_log
    kind: admin_check
    title: Inspect the stored `DecisionFlowLog`
    description: Read the real executed and skipped units from the chosen `DecisionFlowLog` and compare them to the expected case from step 2.
    admin_manual:
      - Open Django Admin → Scoring → Decision flow logs.
      - Open the relevant `DecisionFlowLog`.
      - Copy its numeric id into `DECISION_FLOW_LOG_ID`.
      - Confirm that the related `Check` has `is_production=True`.
      - Confirm that `config.name` matches the expected config from step 2.
      - Run the helper below from the repository root and replace `<ts-namespace>` plus `<decision-flow-log-id>` with the tested values.
      - Compare `executed_units` from the helper output to `expected_executed_units` from step 1.
      - On patched `new` `V2` cases, treat `skipped_units` as informative only; the matrix is defined by executed units.
      - If the case depends on `SCORE_RANGE`, also confirm the stored `CBScoreRange` in the related NRKI response.
    shell:
      inline: |
        DECISION_FLOW_LOG_ID=<decision-flow-log-id>
        python3 ../environment/tripping-avenger/tools/ts_testing/ts_django_shell_runner.py <ts-namespace> --script tmp/tsd-9273/playbooks/manual-qa/scripts/inspect_decision_flow_log.py --var DECISION_FLOW_LOG_ID=${DECISION_FLOW_LOG_ID} --pretty

  - id: optional_production_matrix_smoke
    kind: note
    title: Optional automated production matrix smoke
    description: Run this helper when you need a quick TS proof that a production-like patched `new/classic/NRKI=1/PREPEND=1/REPI=0/SCORE_RANGE=A` case reaches the registry units and matches the expected matrix. The helper isolates unrelated pre-registry TS blockers so the check focuses on the NRKI/REPI unit order.
    shell:
      inline: |
        python3 ../environment/tripping-avenger/tools/ts_testing/ts_django_shell_runner.py <ts-namespace> --script tmp/tsd-9273/playbooks/manual-qa/scripts/verify_production_matrix_smoke.py --pretty --exec-timeout-seconds 900

print_case_reference.py

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
import sys


FLOWS = {"classic", "spi", "split", "bankid"}
REGISTRATION_KINDS = {"new", "existing"}
MODES = {"baseline", "patched"}
SCORE_RANGES = {"A", "F", "-"}

LEGACY_CONFIG = {
    "classic": "apps.scoring.decision_flow.specifications.cz.upgrade_flow.CZUpgradeFlowSpecification",
    "spi": "apps.scoring.decision_flow.specifications.cz.spi_upgrade_flow.SPICZUpgradeFlowSpecification",
    "split": "apps.scoring.decision_flow.specifications.cz.splitted_upgrade_flow.SplittedCZUpgradeFlowSpecification",
    "bankid": "apps.scoring.decision_flow.specifications.cz.bank_id_upgrade_flow.BankIdUpgradeFlowSpecification",
}

V2_CONFIG = {
    "classic": "apps.scoring.decision_flow.specifications.cz.upgrade_flow.CZUpgradeFlowSpecificationV2",
    "spi": "apps.scoring.decision_flow.specifications.cz.spi_upgrade_flow.SPICZUpgradeFlowSpecificationV2",
    "split": "apps.scoring.decision_flow.specifications.cz.splitted_upgrade_flow.SplittedCZUpgradeFlowSpecificationV2",
    "bankid": "apps.scoring.decision_flow.specifications.cz.bank_id_upgrade_flow.BankIdUpgradeFlowSpecificationV2",
}


def get_env_bool(name: str, *, default: bool = False) -> bool:
    value = os.getenv(name)
    if value is None:
        return default
    return value.strip() in {"1", "true", "True", "yes", "YES"}


def baseline_units(flow: str, *, nrki: bool, prepend: bool, score_range: str) -> list[str]:
    if flow in {"classic", "spi"}:
        if not nrki:
            return ["InsolvencyChecker", "REPIFetcher", "REPIChecker", "ExecutionChecker", "SolusChecker"]
        if not prepend:
            return [
                "InsolvencyChecker",
                "REPIFetcher",
                "REPIChecker",
                "NRKIFetcher",
                "NRKIChecker",
                "ExecutionChecker",
                "SolusChecker",
            ]
        if score_range == "F":
            return ["InsolvencyChecker", "REPIFetcher", "REPIChecker", "NRKIFetcher", "NRKIChecker"]
        return [
            "InsolvencyChecker",
            "REPIFetcher",
            "REPIChecker",
            "NRKIFetcher",
            "NRKIChecker",
            "ExecutionChecker",
            "SolusChecker",
        ]

    if flow == "split":
        if not nrki:
            return ["InsolvencyChecker", "REPIFetcher", "REPIChecker", "ExecutionChecker", "SolusChecker"]
        if not prepend:
            return [
                "InsolvencyChecker",
                "REPIFetcher",
                "REPIChecker",
                "ExecutionChecker",
                "SolusChecker",
                "NRKIFetcher",
                "NRKIChecker",
            ]
        if score_range == "F":
            return ["NRKIFetcher", "NRKIChecker", "InsolvencyChecker", "REPIFetcher", "REPIChecker"]
        return [
            "NRKIFetcher",
            "NRKIChecker",
            "InsolvencyChecker",
            "REPIFetcher",
            "REPIChecker",
            "ExecutionChecker",
            "SolusChecker",
        ]

    if flow == "bankid":
        if not nrki:
            return ["InsolvencyChecker", "REPIFetcher", "REPIChecker", "ExecutionChecker", "SolusChecker"]
        return [
            "InsolvencyChecker",
            "REPIFetcher",
            "REPIChecker",
            "ExecutionChecker",
            "SolusChecker",
            "NRKIFetcher",
            "NRKIChecker",
        ]

    raise AssertionError(flow)


def patched_v2_new_units(*, nrki: bool, repi: bool, score_range: str) -> list[str]:
    units: list[str] = []
    if nrki:
        units.extend(["NRKIFetcher", "NRKIChecker"])
    if nrki and score_range == "F":
        return units
    units.append("InsolvencyChecker")
    if repi:
        units.extend(["REPIFetcher", "REPIChecker"])
    units.extend(["ExecutionChecker", "SolusChecker"])
    return units


def expected_units(
    *,
    mode: str,
    registration_kind: str,
    flow: str,
    nrki: bool,
    prepend: bool,
    repi: bool,
    score_range: str,
) -> list[str]:
    if mode == "baseline":
        return baseline_units(flow, nrki=nrki, prepend=prepend, score_range=score_range)
    if registration_kind == "existing":
        return baseline_units(flow, nrki=nrki, prepend=prepend, score_range=score_range)
    if not prepend:
        return baseline_units(flow, nrki=nrki, prepend=prepend, score_range=score_range)
    return patched_v2_new_units(nrki=nrki, repi=repi, score_range=score_range)


def expected_config_name(*, mode: str, registration_kind: str, flow: str) -> str:
    prepend = get_env_bool("PREPEND")
    if mode == "patched" and registration_kind == "new" and prepend:
        return V2_CONFIG[flow]
    return LEGACY_CONFIG[flow]


def build_notes(*, mode: str, registration_kind: str, nrki: bool, prepend: bool, repi: bool, score_range: str) -> list[str]:
    notes: list[str] = []
    if not nrki and score_range != "-":
        notes.append("`SCORE_RANGE` is ignored when `NRKI=0`; use `-` for clarity.")
    if nrki and prepend and score_range not in {"A", "F"}:
        notes.append("Use `A` as a non-good reference sample or `F` as a good-score reference sample.")
    if mode == "baseline":
        notes.append("`REPI` does not apply on baseline; keep the TS in legacy behavior.")
    elif registration_kind == "existing":
        notes.append("`REPI` does not change the legacy rerun shape; the first stored flow config stays authoritative.")
    else:
        if prepend:
            notes.append(
                "On patched `new` cases with `PREPEND=1`, `REPI=0` means default V2 behavior and `REPI=1` means rollback mode."
            )
        else:
            notes.append("On patched `new` cases with `PREPEND=0`, the request stays on the legacy flow, so `REPI` does not affect the executed-unit sequence.")
    if mode == "patched" and registration_kind == "existing":
        notes.append(
            "To create this case manually on the patched TS, first create the request with `PREPEND=0` so the first stored flow stays legacy, then switch features to the target case and rerun."
        )
    return notes


def main() -> int:
    mode = os.getenv("MATRIX_MODE", "patched")
    registration_kind = os.getenv("REGISTRATION_KIND", "new")
    flow = os.getenv("FLOW", "classic")
    score_range = os.getenv("SCORE_RANGE", "-")
    nrki = get_env_bool("NRKI")
    prepend = get_env_bool("PREPEND")
    repi = get_env_bool("REPI")

    errors = []
    if mode not in MODES:
        errors.append(f"Unsupported `MATRIX_MODE`: {mode!r}. Use one of {sorted(MODES)}.")
    if registration_kind not in REGISTRATION_KINDS:
        errors.append(
            f"Unsupported `REGISTRATION_KIND`: {registration_kind!r}. Use one of {sorted(REGISTRATION_KINDS)}."
        )
    if flow not in FLOWS:
        errors.append(f"Unsupported `FLOW`: {flow!r}. Use one of {sorted(FLOWS)}.")
    if score_range not in SCORE_RANGES:
        errors.append(f"Unsupported `SCORE_RANGE`: {score_range!r}. Use one of {sorted(SCORE_RANGES)}.")
    if not nrki and prepend:
        errors.append("`PREPEND=1` without `NRKI=1` is allowed by the matrix, but it behaves like `NRKI=0`.")

    if errors:
        print(json.dumps({"ok": False, "errors": errors}, indent=2))
        return 1

    payload = {
        "ok": True,
        "case": {
            "MATRIX_MODE": mode,
            "REGISTRATION_KIND": registration_kind,
            "FLOW": flow,
            "NRKI": nrki,
            "PREPEND": prepend,
            "REPI": repi,
            "SCORE_RANGE": score_range,
        },
        "expected_config_name": expected_config_name(mode=mode, registration_kind=registration_kind, flow=flow),
        "expected_executed_units": expected_units(
            mode=mode,
            registration_kind=registration_kind,
            flow=flow,
            nrki=nrki,
            prepend=prepend,
            repi=repi,
            score_range=score_range,
        ),
        "notes": build_notes(
            mode=mode,
            registration_kind=registration_kind,
            nrki=nrki,
            prepend=prepend,
            repi=repi,
            score_range=score_range,
        ),
    }
    print(json.dumps(payload, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

print_case_reference.sh

#!/usr/bin/env bash
set -euo pipefail

python3 tmp/tsd-9273/playbooks/manual-qa/scripts/print_case_reference.py

verify_production_matrix_smoke.py

from __future__ import annotations

import json
import traceback
import uuid
from contextlib import contextmanager
from decimal import Decimal
from ipaddress import ip_address
from unittest import mock

from django.db import transaction

from apps.account.enums import CustomerSignupType
from apps.account.models import CustomerAccount
from apps.api.enums import EshopApiVersion
from apps.features.models import Feature
from apps.flow.enums import XAppType, XDeviceType
from apps.customer.enums.upgrade_requests import UpgradeRequestOrigin
from apps.scoring.client import RiskScoringClient
from apps.scoring.client.base import ClientData, ClientPlatformIdentification
from apps.scoring.client.registration_data import RegistrationData
from apps.scoring.decision_flow.core import DecisionUnitStatus, run_decision_flow
from apps.scoring.enums import DecisionCheckpoint
from apps.scoring.models import DecisionFlowLog
from apps.scoring.services.decision_flow_specification import DecisionFlowSpecificationService
from libs.country.countries import Country
from libs.generators import PersonalIdGenerator, PhoneNumberGenerator
from libs.site import CZ


RESULT_START = "===TS_JSON_START==="
RESULT_END = "===TS_JSON_END==="
RUN_ID = uuid.uuid4().hex[:8]

FEATURES = (
    Feature.Type.NRKI,
    Feature.Type.NRKI_PREPEND_FOR_GOOD_SCORE_RANGE,
    Feature.Type.CZ_REPI_ENABLED,
    Feature.Type.CZ_MULTISTEP_RISK_ONBOARDING_ENABLED,
    Feature.Type.CZ_BANKID_REGISTRATION_ENABLED,
)

RELEVANT_UNITS = {
    "NRKIFetcher",
    "NRKIChecker",
    "InsolvencyChecker",
    "REPIFetcher",
    "REPIChecker",
    "ExecutionChecker",
    "SolusChecker",
}

EXPECTED_EXECUTED_UNITS = [
    "NRKIFetcher",
    "NRKIChecker",
    "InsolvencyChecker",
    "ExecutionChecker",
    "SolusChecker",
]
EXPECTED_SKIPPED_UNITS = [
    "REPIFetcher",
    "REPIChecker",
]


def feature_label(value):
    return getattr(value, "label", str(value))


def set_feature(feature_type, value):
    Feature.objects.update_or_create(
        id=feature_type,
        defaults={"name": feature_label(feature_type), "value": value},
    )


def snapshot_features():
    result = {}
    for feature_type in FEATURES:
        try:
            feature = Feature.objects.get(id=feature_type)
        except Feature.DoesNotExist:
            result[int(feature_type)] = None
        else:
            result[int(feature_type)] = {"name": feature.name, "value": feature.value}
    return result


def restore_features(snapshot):
    for feature_id, value in snapshot.items():
        if value is None:
            Feature.objects.filter(id=feature_id).delete()
        else:
            Feature.objects.update_or_create(id=feature_id, defaults=value)


@contextmanager
def feature_setup():
    original_features = snapshot_features()
    try:
        set_feature(Feature.Type.NRKI, 100)
        set_feature(Feature.Type.NRKI_PREPEND_FOR_GOOD_SCORE_RANGE, 100)
        set_feature(Feature.Type.CZ_REPI_ENABLED, 0)
        set_feature(Feature.Type.CZ_MULTISTEP_RISK_ONBOARDING_ENABLED, 0)
        set_feature(Feature.Type.CZ_BANKID_REGISTRATION_ENABLED, 0)
        yield original_features
    finally:
        restore_features(original_features)


def client_data():
    platform = ClientPlatformIdentification(
        x_app_type=XAppType.BROWSER.value,
        x_device_type=XDeviceType.DESKTOP.value,
    )
    return ClientData(
        api_version=EshopApiVersion.REGISTRATION,
        is_production=True,
        ip_address=ip_address("84.42.140.14"),
        platform_identification=platform,
        http_user_agent="TSD-9273 production matrix smoke",
        http_connection=None,
        http_accept_language="cs-CZ",
        x_fingerprint={"len": 16, "hash": 9273},
        x_fingerprint_v2=f"tsd9273matrix{RUN_ID}",
        x_fingerprint_v2_checksum=9273,
        x_fingerprint_v2_version="1",
        x_fp_request_id=f"req-{RUN_ID}",
        x_fp_visitor_id=f"vis-{RUN_ID}",
        x_mobile=None,
        x_version=1,
        x_session=f"matrix-{RUN_ID}",
        x_flow_session_id=f"matrix-{RUN_ID}",
        x_local_ip=[],
        x_browser=None,
        x_sites=None,
    )


def generated_personal_id():
    return CZ.get(PersonalIdGenerator).generate()


def short_unit_name(full_name):
    base = full_name.split(":", 1)[0].split("@", 1)[0]
    return base.rsplit(".", 1)[-1]


def inspect_flow_log(flow_log: DecisionFlowLog):
    unit_logs = list(flow_log.unit_logs.filter(date_rolled_back__isnull=True).order_by("phase", "id"))
    relevant_logs = [
        {
            "unit": short_unit_name(unit_log.unit.name),
            "status": unit_log.get_status_display(),
            "phase": unit_log.phase,
            "unit_config_name": unit_log.unit.name,
        }
        for unit_log in unit_logs
        if short_unit_name(unit_log.unit.name) in RELEVANT_UNITS
    ]
    return {
        "execution_status": flow_log.get_execution_status_display() if flow_log.execution_status is not None else None,
        "executed_units": [row["unit"] for row in relevant_logs if row["status"].upper() != "SKIPPED"],
        "skipped_units": [row["unit"] for row in relevant_logs if row["status"].upper() == "SKIPPED"],
        "relevant_unit_logs": relevant_logs,
        "unit_log_count": flow_log.unit_logs.count(),
    }


def run():
    with feature_setup() as original_features:
        customer = CustomerAccount(
            email=f"tsd-9273-{RUN_ID}-matrix@foo.cz",
            site=CZ,
            name="TSD 9273 Matrix",
            signup_type=CustomerSignupType.Homepage.code,
            type=CustomerAccount.TYPE_STANDARD,
            phone_number=CZ.get(PhoneNumberGenerator).generate(),
        )
        customer.set_feature_flags()
        customer.save()
        risk_client = RiskScoringClient.for_customer_registration(
            customer,
            UpgradeRequestOrigin.LEGACY,
            platform_identification=ClientPlatformIdentification(
                x_app_type=XAppType.BROWSER.value,
                x_device_type=XDeviceType.DESKTOP.value,
            ),
        )
        with transaction.atomic():
            risk_client.process_current_registration_data(
                RegistrationData(
                    name="TSD 9273 Matrix",
                    phone_number=customer.phone_number,
                    personal_id=generated_personal_id(),
                    citizenship="CZ",
                    international_address_street="Sokolovská 47/73",
                    international_address_city="Praha",
                    international_address_zipcode="18600",
                    international_address_country=Country("CZ"),
                    extra_info_income_amount=Decimal("20000"),
                    extra_info_expense_amount=Decimal("1000"),
                    extra_info_expense_debt=Decimal("2000"),
                    extra_info_income_type=1,
                    extra_info_employment_type="1",
                    extra_info_education_type=1,
                    extra_info_family_children=0,
                    extra_info_family_type=1,
                    is_pep=False,
                ),
                skip_cross_check=True,
                skip_address_normalization=True,
            )
            risk_client.initialise_check(client_data())
            flow_log = risk_client.get_decision_flow().get_or_create_log()

        check = risk_client.get_check()
        actual_spec = type(check.site.get(DecisionFlowSpecificationService).get_specification(check)).__name__
        risk_client._create_upgrade_check_related_models(check)
        from apps.scoring.decision_flow.units.common.openscoring import OpenscoringFetcher
        from apps.scoring.decision_flow.units.common.kyc_checkers import VeriffKOChecker
        from apps.scoring.decision_flow.units.common.municipal_address import MunicipalAddressChecker
        from apps.scoring.decision_flow.units.common.nikita_identity_checkers import PersonalIdChecker
        from apps.scoring.decision_flow.units.common.nikita_scoring_checkers import LowScoreChecker

        with (
            mock.patch.object(OpenscoringFetcher, "execute", return_value=DecisionUnitStatus.PASSED),
            mock.patch.object(MunicipalAddressChecker, "execute", return_value=DecisionUnitStatus.PASSED),
            mock.patch.object(PersonalIdChecker, "execute", return_value=DecisionUnitStatus.PASSED),
            mock.patch.object(LowScoreChecker, "execute", return_value=DecisionUnitStatus.PASSED),
            mock.patch.object(VeriffKOChecker, "execute", return_value=DecisionUnitStatus.PASSED),
        ):
            run_decision_flow(check, DecisionCheckpoint.ACCEPT)
        check.refresh_from_db()
        flow_log.refresh_from_db()
        inspected = inspect_flow_log(flow_log)

    after_features = snapshot_features()
    return {
        "ok": (
            check.is_production
            and actual_spec == "CZUpgradeFlowSpecificationV2"
            and inspected["executed_units"] == EXPECTED_EXECUTED_UNITS
            and inspected["skipped_units"] == EXPECTED_SKIPPED_UNITS
            and original_features == after_features
        ),
        "case": {
            "MATRIX_MODE": "patched",
            "REGISTRATION_KIND": "new",
            "FLOW": "classic",
            "NRKI": True,
            "PREPEND": True,
            "REPI": False,
            "SCORE_RANGE": "A",
        },
        "expected_config_name": "apps.scoring.decision_flow.specifications.cz.upgrade_flow.CZUpgradeFlowSpecificationV2",
        "actual_specification": actual_spec,
        "check_id": check.id,
        "decision_flow_log_id": flow_log.id,
        "is_production": check.is_production,
        "expected_executed_units": EXPECTED_EXECUTED_UNITS,
        "expected_skipped_units": EXPECTED_SKIPPED_UNITS,
        "before_features": original_features,
        "after_features": after_features,
        **inspected,
    }


try:
    payload = run()
except Exception as exc:
    payload = {"ok": False, "error": str(exc), "traceback": traceback.format_exc()}

print(RESULT_START)
print(json.dumps(payload, indent=2, sort_keys=True, default=str))
print(RESULT_END)