dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_doors.py
"""The use cases through both doors, on SQLite: every use case is a route AND a tool, a create answers the
same through the API and MCP, a missing row is 404, and no permission means nothing is written."""
from conftest import ALICE, NOBODY, api, tool, tools
ROUTES = [
("list_groups", "GET /groups"),
("get_group", "GET /groups/{group_id}"),
("keep_group", "POST /groups"),
("open_group_period", "POST /groups/{group_id}/group-periods"),
("upload_reported_data", "POST /groups/{group_id}/group-periods/{period_id}/upload-reported-data"),
("accept_reported_upload", "POST /groups/{group_id}/group-periods/{period_id}/accept-reported-upload"),
("collect_group_data", "POST /groups/{group_id}/group-periods/{period_id}/collect-group-data"),
("recollect_group_data", "POST /groups/{group_id}/group-periods/{period_id}/recollect-group-data"),
("translate_currencies", "POST /groups/{group_id}/group-periods/{period_id}/translate-currencies"),
("run_eliminations", "POST /groups/{group_id}/group-periods/{period_id}/run-eliminations"),
("sign_off_group_close", "POST /groups/{group_id}/group-periods/{period_id}/sign-off-group-close"),
("reopen_group_period", "POST /groups/{group_id}/group-periods/{period_id}/reopen-group-period"),
("keep_group_rules", "POST /groups/{group_id}/rules"),
("enter_ic_inventory", "POST /groups/{group_id}/group-periods/{period_id}/ic-inventory"),
("list_structure_changes", "GET /structure-changes"),
("get_structure_change", "GET /structure-changes/{change_id}"),
("propose_structure_change", "POST /structure-changes"),
("approve_structure_change", "POST /structure-changes/{change_id}/approve-structure-change"),
("reject_structure_change", "POST /structure-changes/{change_id}/reject-structure-change"),
("list_cons_journals", "GET /cons-journals"),
("get_cons_journal", "GET /cons-journals/{cons_journal_id}"),
("propose_cons_entry", "POST /cons-journals/propose-cons-entry"),
("approve_cons_entry", "POST /cons-journals/{cons_journal_id}/approve-cons-entry"),
("reject_cons_entry", "POST /cons-journals/{cons_journal_id}/reject-cons-entry"),
("reverse_cons_entry", "POST /cons-journals/{cons_journal_id}/reverse-cons-entry"),
("list_statements", "GET /statements"),
("get_statement", "GET /statements/{report}/{group_id}/{period_id}/{node_id}"),
("list_ai_findings", "GET /ai-findings"),
("get_ai_finding", "GET /ai-findings/{finding_id}"),
("list_cons_load_runs", "GET /migration-runs"),
("get_cons_load_run", "GET /migration-runs/{run_id}"),
("list_cons_id_maps", "GET /migration-maps"),
("get_cons_migration_map", "GET /migration-maps/{odoo_db}/{odoo_model}/{odoo_id}"),
("list_cons_migration_runs", "GET /cons-migration-runs"),
("get_cons_migration_run", "GET /cons-migration-runs/{cons_migration_run_id}"),
("load_cons_from_odoo", "POST /cons-migration-runs"),
("reconcile_cons_migration", "POST /cons-migration-runs/{cons_migration_run_id}/reconcile-cons-migration"),
("explain_cons_difference", "POST /cons-migration-runs/{cons_migration_run_id}/explain-cons-difference"),
("reload_cons_migration", "POST /cons-migration-runs/{cons_migration_run_id}/reload-cons-migration"),
("sign_cons_reconciliation", "POST /cons-migration-runs/{cons_migration_run_id}/sign-cons-reconciliation"),
("discard_cons_migration", "POST /cons-migration-runs/{cons_migration_run_id}/discard-cons-migration"),
("list_cons_settings", "GET /settings"),
("set_cons_setting", "POST /settings/{setting_key}"),
]
REMOVED = [("post_cons_auto_entry", "POST /cons-journals"), ("keep_statement", "POST /statements"), ("keep_ai_finding", "POST /ai-findings"),
("keep_cons_load_run", "POST /migration-runs"), ("keep_cons_id_map", "POST /migration-maps")] # audit v1 MUST-2a/2b + SHOULDs: raw scaffold writers of guarded rows
def test_the_raw_scaffold_writers_are_not_doors():
"""An entry is posted only by the runs or through propose + approve; a statement only by materialise; a load run only through
/cons-migration-runs; an AI finding only by the AI step."""
status, spec = api("GET", "/openapi.json", ALICE)
assert status == 200
for name, route in REMOVED:
method, path = route.split(" ", 1)
assert method.lower() not in spec["paths"].get(path, {}), route
assert not {n for n, _ in REMOVED} & set(tools(ALICE))
def test_every_use_case_is_a_route_and_a_tool():
status, spec = api("GET", "/openapi.json", ALICE)
assert status == 200
for name, route in ROUTES:
method, path = route.split(" ", 1)
assert method.lower() in spec["paths"].get(path, {}), route
assert {n for n, _ in ROUTES} <= set(tools(ALICE))
def test_a_group_through_both_doors():
body = {"group_id": "GROUP-ID-1"}
status, made = api("POST", "/groups", ALICE, body)
assert status == 200, made
assert api("GET", "/groups/GROUP-ID-1", ALICE)[1] == made
assert tool(ALICE, "get_group", {"group_id": "GROUP-ID-1"}) == ("ok", made)
assert made["version"] == 1
def test_a_missing_group_is_not_found():
assert api("GET", "/groups/GROUP-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_group", {"group_id": "GROUP-ID-1"})[0] == "refused"
def test_a_missing_structure_change_is_not_found():
assert api("GET", "/structure-changes/CHANGE-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_structure_change", {"change_id": "CHANGE-ID-1"})[0] == "refused"
def test_a_missing_cons_journal_is_not_found():
assert api("GET", "/cons-journals/CONS-JOURNAL-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_cons_journal", {"cons_journal_id": "CONS-JOURNAL-ID-1"})[0] == "refused"
def test_a_missing_statement_is_not_found():
assert api("GET", "/statements/REPORT-1/GROUP-ID-1/PERIOD-ID-1/NODE-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_statement", {"report": "REPORT-1", "group_id": "GROUP-ID-1", "period_id": "PERIOD-ID-1", "node_id": "NODE-ID-1"})[0] == "refused"
def test_a_missing_ai_finding_is_not_found():
assert api("GET", "/ai-findings/FINDING-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_ai_finding", {"finding_id": "FINDING-ID-1"})[0] == "refused"
def test_a_missing_migration_run_is_not_found():
assert api("GET", "/migration-runs/RUN-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_cons_load_run", {"run_id": "RUN-ID-1"})[0] == "refused"
def test_a_missing_migration_map_is_not_found():
assert api("GET", "/migration-maps/ODOO-DB-1/ODOO-MODEL-1/ODOO-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_cons_migration_map", {"odoo_db": "ODOO-DB-1", "odoo_model": "ODOO-MODEL-1", "odoo_id": "ODOO-ID-1"})[0] == "refused"
def test_a_cons_migration_run_through_both_doors():
assert api("POST", "/cons-migration-runs", ALICE, {"cons_migration_run_id": "CONS-MIGRATION-RUN-ID-1"})[0] == 409 # the doors are not decided
body = {"cons_migration_run_id": "CONS-MIGRATION-RUN-ID-1", "doors_json": '{"group_id": "GRP", "carried_periods": ["2026-10"]}'}
status, made = api("POST", "/cons-migration-runs", ALICE, body)
assert status == 200, made
assert api("GET", "/cons-migration-runs/CONS-MIGRATION-RUN-ID-1", ALICE)[1] == made
assert tool(ALICE, "get_cons_migration_run", {"cons_migration_run_id": "CONS-MIGRATION-RUN-ID-1"}) == ("ok", made)
assert made["version"] == 1
def test_a_missing_cons_migration_run_is_not_found():
assert api("GET", "/cons-migration-runs/CONS-MIGRATION-RUN-ID-1", ALICE)[0] == 404
assert tool(ALICE, "get_cons_migration_run", {"cons_migration_run_id": "CONS-MIGRATION-RUN-ID-1"})[0] == "refused"
def test_without_the_permission_nothing_is_written():
status, _ = api("POST", "/groups", NOBODY, {"group_id": "GROUP-ID-1"})
assert status == 403
assert api("GET", "/groups/GROUP-ID-1", ALICE)[0] == 404
def test_the_process_view_shows_the_workflows_as_bpmn():
status, body = api("GET", "/process/bpmn", ALICE)
assert status == 200 and body["processes"] == ["group_period", "cons_journal", "structure_change", "cons_migration_run"]
def test_the_settings_are_seeded_one_row_per_setting_typed():
status, body = api("GET", "/settings", ALICE)
assert status == 200
assert {s["setting_key"]: s["kind"] for s in body["settings"]} == {"ai.explain_elimination_difference.mode": "mode", "allow_open_period_collect": "bool", "nci_measurement": "str", "signoff_second_person": "bool", "translation_pl_rate": "str"}
assert all(s["value"] is None for s in body["settings"])
def test_a_setting_of_the_wrong_kind_is_refused():
path = "/controls/signoff_second_person"
assert api("POST", path, ALICE, {"value": "x"})[0] == 422