dodil/erp-brick-hr

tests/test_tier1.py

"""Tier 1 with the REAL neighbour: hr and brick-business-partner composed as one app on one bucket (one SQLite engine
here), the seam reaching bp in process through its routes — so bp's own permission checks and rules run. What hr
promises about the shared party record (HR-50, HR-38, HR-22) is checked on bp's own answers."""
import pathlib
from contextlib import ExitStack

import anyio
import httpx
import pytest
from sqlalchemy import create_engine, event

from erp_platform import seam, work
from erp_platform.scenarios import Catalogue
from erp_platform.scenarios.inprocess import stage

from conftest import headers
from seed import DOMAIN, hire_body, seed

HERE = pathlib.Path(__file__).parent
HR, FIN, TICK = headers("hr_admin"), headers("finance"), headers("clock")


@pytest.fixture
def app(tmp_path, drafter):
    e = create_engine(f"sqlite:///{tmp_path / 'bucket.db'}", connect_args={"check_same_thread": False, "timeout": 30})

    @event.listens_for(e, "connect")
    def _wal(conn, _rec):
        conn.execute("PRAGMA journal_mode=WAL")
    cat = Catalogue("tier1", compose=HERE / "tier1.compose.yaml", roles=HERE / "scenarios" / "roles.yaml")
    with ExitStack() as st:
        staged = stage(cat, st, lambda name: e)                    # every brick and the link tables: one bucket
        st.callback(work.reset_clock)
        yield staged.app
    e.dispose()


def call(app, method, path, who, body=None, params=None):
    async def go():
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://staff") as c:
            return await c.request(method, path, headers=who, json=body, params=params)
    r = anyio.run(go)
    return r.status_code, r.json()


def at(when):
    work.set_clock(at=work.parse_instant(f"{when} Europe/London"), freeze=True)


def setup(app):
    at("2026-10-19T10:00")
    seed(lambda p, b: call(app, "POST", p, HR, b), prefix="/hr")


def test_hr_50_the_partner_is_created_at_hire_with_a_name_and_nothing_private(app):
    setup(app)
    s, h = call(app, "POST", "/hr/hires", HR, hire_body("Anna", "Berg", "2026-11-02", personal_phone="+44 7700 900999"))
    assert s == 200, h
    s, p = call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)
    assert s == 200 and p["kind"] == "person" and p["bp_key"] == f"person:{h['person_id']}"
    assert p["legal_name"] == "Anna Berg" and p["roles"] == [] and p["work_email"] == ""
    assert "home.example" not in str(p) and "7700" not in str(p)
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", headers("manager"))[0] == 404   # bp:person:read only


def test_hr_38_the_role_arrives_on_the_start_date_and_leaves_after_the_last_day(app):
    setup(app)
    s, h = call(app, "POST", "/hr/hires", HR, hire_body("Ben", "Cole", "2026-11-02"))
    at("2026-11-02T00:05")
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[1]["roles"] == ["employee"]
    call(app, "POST", f"/hr/people/{h['person_id']}/leave", HR, {"last_day": "2026-11-30", "reason": "resignation"})
    at("2026-11-30T12:00")
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[1]["roles"] == ["employee"]
    at("2026-12-01T00:05")
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[1]["roles"] == []
    n = len(call(app, "GET", f"/business-partner/partners/{h['bp_id']}/history", FIN)[1]["events"])
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    assert len(call(app, "GET", f"/business-partner/partners/{h['bp_id']}/history", FIN)[1]["events"]) == n   # synced once


def test_a_company_freelancer_works_for_their_company_which_is_the_supplier(app):
    setup(app)
    s, h = call(app, "POST", "/hr/hires", HR, hire_body(
        "Cy", "Dee", "2026-11-02", worker_type="freelancer_company", rate_amount="500.0000", rate_basis="day",
        currency="GBP", company={"legal_name": "Studio Nine Ltd", "company_number": "09876543"}))
    assert s == 200, h
    s, co = call(app, "GET", f"/business-partner/partners/{h['supplier_bp_id']}", FIN)
    assert co["kind"] == "organisation" and "supplier" in co["roles"] and co["legal_name"] == "Studio Nine Ltd"
    s, fam = call(app, "GET", f"/business-partner/partners/{h['supplier_bp_id']}/family", FIN, params={"rel": "works_for"})
    assert [str(m["bp_id"]) for m in fam["below"]] == [h["bp_id"]]
    at("2026-11-02T00:05")
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[1]["roles"] == ["worker"]


def test_hr_31_the_work_email_made_by_hand_reaches_the_partner(app):
    setup(app)
    s, h = call(app, "POST", "/hr/hires", HR, hire_body("Di", "Eve", "2026-10-20"))
    rid = h["run_id"]
    call(app, "POST", f"/hr/people/{h['person_id']}/compliance", HR, {
        "kind": "right_to_work_check", "checked_on": "2026-10-19", "checked_by": "hr", "check_type": "manual_document_check"})
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    assert call(app, "POST", f"/hr/runs/{rid}/steps/3/complete", HR, {"attachment": {
        "filename": "c.pdf", "content_base64": "JVBERg=="}})[0] == 200
    at("2026-10-19T11:00")
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    s, out = call(app, "POST", f"/hr/runs/{rid}/steps/5/complete", headers("it"),
                  {"provider_ref": "u-1", "output": {"primary_email": f"di.eve@{DOMAIN}"}})
    assert s == 200 and out["state"] == "done", out
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[1]["work_email"] == f"di.eve@{DOMAIN}"


def test_business_partner_down_the_hire_still_lands_and_the_tick_catches_up(app):
    setup(app)
    local = seam._LOCAL.pop("business-partner")
    try:
        s, h = call(app, "POST", "/hr/hires", HR, hire_body("Fay", "Gee", "2026-10-19"))
        assert s == 200
    finally:
        seam._LOCAL["business-partner"] = local
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[0] == 404
    call(app, "POST", "/hr/run-due-steps", TICK, {})
    assert call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[1]["roles"] == ["employee"]


def test_hr_22_erasure_anonymises_the_partner_through_the_seam(app):
    at("2024-12-10T10:00")
    seed(lambda p, b: call(app, "POST", p, HR, b), prefix="/hr")
    s, h = call(app, "POST", "/hr/hires", HR, hire_body("Gil", "Ho", "2024-12-02"))
    call(app, "POST", f"/hr/people/{h['person_id']}/leave", HR, {"last_day": "2025-01-31", "reason": "resignation"})
    at("2026-10-20T10:00")
    s, q = call(app, "POST", "/hr/privacy/requests", HR, {"person_id": h["person_id"], "kind": "erasure",
                                                          "received_on": "2026-10-20"})
    s, out = call(app, "POST", f"/hr/privacy/requests/{q['request_id']}/erase", HR, {})
    assert s == 200, out
    p = call(app, "GET", f"/business-partner/partners/{h['bp_id']}", FIN)[1]
    assert p["legal_name"] == "Erased person" and p["anonymised"] is True
    s, person = call(app, "GET", f"/hr/people/{h['person_id']}", HR)
    assert person["detail"]["legal_given_name"] == "Erased" and person["detail"]["personal_email"] is None


def test_hr_11_in_a_composition_the_playbook_bpmn_comes_in_its_envelope(app):
    from bpmn_check import check
    setup(app)
    s, out = call(app, "GET", "/hr/playbooks/employee-offboarding/bpmn", HR)
    assert s == 200 and out["media_type"] == "application/xml" and out["version"] == 1
    check(out["bpmn"])