dodil/erp-brick-time-off

DODIL ERP brick: time off

tests/fakes.py

"""An in-memory stand-in for hr, registered as a local seam (erp_platform.seam.register_local): the person by sign-in, the employment, the derived
manager and the work pattern of a range - contracted minutes, the public holidays of the person's OWN calendar, employed or not, FTE - per day. It
holds plain dicts the tests fill, answers the contract infrastructure/seams.py documents, and can be taken DOWN. conftest.py checks every call
against hr's signed spec."""
from datetime import date, timedelta
from typing import Optional

from fastapi import FastAPI, Query, Response
from fastapi.responses import JSONResponse

MON_FRI = {0: 450, 1: 450, 2: 450, 3: 450, 4: 450}
MON_WED = {0: 450, 1: 450, 2: 450}

# the public holidays, per calendar, that the stories name (hr holds them as hr_holiday rows)
GB_EAW = {"2026-01-01": "New Year's Day", "2026-04-03": "Good Friday", "2026-04-06": "Easter Monday", "2026-05-04": "Early May", "2026-05-25": "Spring",
          "2026-08-31": "Summer", "2026-12-25": "Christmas Day", "2026-12-28": "Boxing Day (substitute)", "2027-01-01": "New Year's Day"}
GB_SCT = {**GB_EAW, "2026-11-30": "St Andrew's Day", "2027-01-02": "2 January"}


class Hr:
    def __init__(self):
        self.by_sub: dict = {}          # sub -> person_id
        self.people: dict = {}          # person_id -> {company, email, sub, weekdays, fte, calendar, start, end, fte_from, cal_name}
        self.managers: dict = {}        # person_id -> [level 1, level 2]: each {person_id, bp_id, pool_user_sub, email} or None
        self.status: Optional[int] = None       # 500 takes hr down
        self.malformed = False          # a pattern without the shape the contract promises
        self.calls: list = []           # (path, params) of every request, for the tests that count them
        self.app = self._app()

    def person(self, person_id: int, email: str, *, company: str = "UK01", weekdays=MON_FRI, fte: str = "1.0000", calendar=GB_EAW, start=date(2020, 1, 1),
               end: Optional[date] = None, fte_from: Optional[tuple] = None, manager=None, manager2=None) -> dict:
        sub = f"sub-{person_id}"
        self.people[person_id] = {"company": company, "email": email, "sub": sub, "weekdays": weekdays, "fte": fte, "calendar": calendar, "start": start,
                                  "end": end, "fte_from": fte_from}
        self.by_sub[sub] = person_id
        self.managers[person_id] = [manager, manager2]
        return self.people[person_id]

    def as_manager(self, person_id: int) -> dict:
        p = self.people[person_id]
        return {"person_id": person_id, "bp_id": 70000 + person_id, "pool_user_sub": p["sub"], "email": p["email"]}

    def _day(self, p: dict, d: date) -> dict:
        employed = p["start"] <= d and (p["end"] is None or d <= p["end"])
        fte = p["fte"]
        if p["fte_from"] and d >= p["fte_from"][0]:
            fte = p["fte_from"][1]
        return {"date": d.isoformat(), "employed": employed, "contracted_minutes": p["weekdays"].get(d.weekday(), 0) if employed else 0,
                "public_holiday": p["calendar"].get(d.isoformat()) if employed else None, "location_code": "LON", "holiday_calendar": "GB-EAW",
                "fte": fte if employed else "0.0000"}

    def _app(self) -> FastAPI:
        app, me = FastAPI(), self

        def down(response: Response):
            if me.status:
                response.status_code = me.status
                return {"detail": "down"}

        @app.get("/persons")
        def by_sub(response: Response, pool_user_sub: str):
            """hr HR-58: 200 {person_id, bp_id, company_code}; 404 when no person carries the sub."""
            me.calls.append(("/persons", {"pool_user_sub": pool_user_sub}))
            if me.status:
                return down(response)
            pid = me.by_sub.get(pool_user_sub)
            if pid is None:
                response.status_code = 404
                return {"detail": "no such person"}
            return {"person_id": pid, "bp_id": 70000 + pid, "company_code": me.people[pid]["company"]}

        @app.get("/persons/{person_id}/employment")
        def employment(person_id: int, on: str, response: Response):
            """hr HR-56: 200 {person_id, on, employed, engagement_id, worker_type, company_code, org_unit_id, location_code, cost_centre, bp_id}."""
            me.calls.append(("employment", {"person_id": person_id, "on": on}))
            if me.status:
                return down(response)
            p = me.people.get(person_id)
            if p is None:
                response.status_code = 404
                return {"detail": "no such person"}
            d = me._day(p, date.fromisoformat(on))
            return {"person_id": person_id, "on": on, "employed": d["employed"], "worker_type": "employee", "company_code": p["company"],
                    "location_code": "LON"}

        @app.get("/persons/{person_id}/manager")
        def manager(person_id: int, on: str, level: int, response: Response):
            """hr HR-57: 200 {person_id, on, level, manager: {...} | null, source}."""
            me.calls.append(("manager", {"person_id": person_id, "on": on, "level": level}))
            if me.status:
                return down(response)
            chain = me.managers.get(person_id, [None, None])
            return {"person_id": person_id, "on": on, "level": level, "manager": chain[level - 1] if level <= len(chain) else None, "source": "org_tree"}

        @app.get("/persons/{person_id}/work-pattern")
        def work_pattern(person_id: int, response: Response, frm: str = Query(alias="from"), to: str = Query()):
            """hr HR-62: 200 {person_id, from, to, days: [{date, employed, contracted_minutes, public_holiday, location_code, holiday_calendar, fte}]};
            422 {gate: bounds} for a range of more than 400 days."""
            me.calls.append(("work-pattern", {"person_id": person_id, "from": frm, "to": to}))
            if me.status:
                return down(response)
            p = me.people.get(person_id)
            if p is None:
                response.status_code = 404
                return {"detail": "no such person"}
            a, b = date.fromisoformat(frm), date.fromisoformat(to)
            if me.malformed:
                return {"person_id": person_id, "days": [{"date": frm}]}
            if (b - a).days > 400 or b < a:
                return JSONResponse({"gate": "bounds", "reason": "at most 400 days", "fix": "ask in two parts"}, status_code=422)
            return {"person_id": person_id, "from": frm, "to": to, "days": [me._day(p, a + timedelta(days=i)) for i in range((b - a).days + 1)]}
        return app