154 lines
6.8 KiB
Python
154 lines
6.8 KiB
Python
 
|
"""The profile CSV import: preview, every error, apply refused with errors, apply clean, the audit rows."""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
from fakedir import FakeDirectory, FakeUser
|
|||
|
|
from fastapi.testclient import TestClient
|
|||
|
|
from pshelpers import app_conn, needs_pg, sign_in
|
|||
|
|
|
|||
|
|
from packetstation.domain import profiles
|
|||
|
|
from packetstation.settings import Settings
|
|||
|
|
|
|||
|
|
pg = needs_pg
|
|||
|
|
|
|||
|
|
PEOPLE = ("weld.andy", "lead.lou", "sup.wes", "dup.dave", "dup.dina", "wh.wade", "qc.quinn")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _import(client: TestClient, text: str, apply: bool = False) -> Any:
|
|||
|
|
return client.post("/api/admin/users/import", json={"csv": text, "apply": apply})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _profiles(settings: Settings) -> dict[str, tuple[Any, ...]]:
|
|||
|
|
with app_conn(settings) as conn:
|
|||
|
|
rows = conn.execute(
|
|||
|
|
"SELECT u.sam_account, u.welder_stamp, u.profile_active,"
|
|||
|
|
" (SELECT array_agg(s.dept_code || ':' || s.name ORDER BY s.dept_code, s.name)"
|
|||
|
|
" FROM user_station_limits l JOIN stations s ON s.id = l.station_id WHERE l.user_id = u.id)"
|
|||
|
|
" FROM users u"
|
|||
|
|
).fetchall()
|
|||
|
|
return {r[0]: (r[1], r[2], r[3] or []) for r in rows}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_parse_csv_shapes() -> None:
|
|||
|
|
cols, rows = profiles.parse_csv("Employee_ID , Welder_Stamp\r\n0042,A1\r\n\r\n,\r\n7A,\r\n")
|
|||
|
|
assert cols == ["employee_id", "welder_stamp"]
|
|||
|
|
assert rows == [(2, {"employee_id": "0042", "welder_stamp": "A1"}), (5, {"employee_id": "7A", "welder_stamp": ""})]
|
|||
|
|
for text, code in (
|
|||
|
|
("", "bad_csv"),
|
|||
|
|
("welder_stamp\nA1\n", "bad_csv"),
|
|||
|
|
("employee_id,colour\n1,red\n", "bad_csv"),
|
|||
|
|
("employee_id,employee_id\n1,1\n", "bad_csv"),
|
|||
|
|
("employee_id\n" + "1\n" * (profiles.CSV_MAX_ROWS + 1), "bad_csv"),
|
|||
|
|
("employee_id\n" + "x" * profiles.CSV_MAX_BYTES, "too_large"),
|
|||
|
|
):
|
|||
|
|
with pytest.raises(Exception) as e:
|
|||
|
|
profiles.parse_csv(text)
|
|||
|
|
assert getattr(e.value, "code", None) == code, text[:40]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pg
|
|||
|
|
def test_import_preview_errors_and_apply(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None:
|
|||
|
|
# An employee ID with a leading zero, kept as text
|
|||
|
|
fake_dir.users.append(FakeUser("zero.zed", "Zed Zero", "0042", ["PS-04-SouthWeld-Worker"]))
|
|||
|
|
fake_dir.users.append(FakeUser("extra.ezra", "Ezra Extra", "7777", ["PS-03-WeldParts-Worker"]))
|
|||
|
|
for sam in (*PEOPLE, "zero.zed", "extra.ezra"):
|
|||
|
|
assert sign_in(pg_client, sam).status_code == 200, sam
|
|||
|
|
sign_in(pg_client, "admin.ann")
|
|||
|
|
with app_conn(pg_settings) as conn:
|
|||
|
|
conn.execute("UPDATE users SET welder_stamp = 'W9' WHERE sam_account = 'sup.wes'")
|
|||
|
|
before = _profiles(pg_settings)
|
|||
|
|
|
|||
|
|
bad = (
|
|||
|
|
"employee_id,welder_stamp,active,stations\n"
|
|||
|
|
"3003,A1,yes,03:Stall 1;03:stall 2\n" # 2 ok
|
|||
|
|
"3003,A2,,\n" # 3 repeated in the file
|
|||
|
|
"5555,,,\n" # 4 unknown employee ID
|
|||
|
|
"9001,X1,,\n" # 5 on more than one user
|
|||
|
|
"3002,bad stamp!,,\n" # 6 bad stamp
|
|||
|
|
"6001,W9,,\n" # 7 stamp taken (sup.wes)
|
|||
|
|
"5001,,maybe,\n" # 8 bad active
|
|||
|
|
"0042,,,99:Stall 1\n" # 9 unknown department
|
|||
|
|
"0042x,,,\n" # 10 unknown (not the same as 0042)
|
|||
|
|
"42,,,\n" # 11 unknown: 42 is not 0042
|
|||
|
|
",A3,,\n" # 12 no employee ID
|
|||
|
|
"2003,,,03:Stall 99\n" # 13 unknown station
|
|||
|
|
"7777,A1,,\n" # 14 A1 already claimed by line 2 in this file
|
|||
|
|
)
|
|||
|
|
r = _import(pg_client, bad)
|
|||
|
|
assert r.status_code == 200, r.text
|
|||
|
|
body = r.json()
|
|||
|
|
assert body["applied"] is False and body["ok"] == 1 and body["errors"] == 12
|
|||
|
|
rows = {x["line"]: x for x in body["rows"]}
|
|||
|
|
assert rows[2]["errors"] == [] and rows[2]["display_name"] == "Andy Welder"
|
|||
|
|
assert rows[2]["changes"] == {
|
|||
|
|
"welder_stamp": {"from": None, "to": "A1"},
|
|||
|
|
"station_limits": {"from": [], "to": ["03:Stall 1", "03:Stall 2"]},
|
|||
|
|
}
|
|||
|
|
expect = {
|
|||
|
|
3: "repeated in the file (first on line 2)",
|
|||
|
|
4: "Unknown employee ID 5555",
|
|||
|
|
5: "on more than one user",
|
|||
|
|
6: "Bad stamp",
|
|||
|
|
7: "Stamp W9 is taken by Wes Super",
|
|||
|
|
8: "Bad active value maybe",
|
|||
|
|
9: "Unknown department 99",
|
|||
|
|
10: "Unknown employee ID 0042x",
|
|||
|
|
11: "Unknown employee ID 42",
|
|||
|
|
12: "Employee ID is missing",
|
|||
|
|
13: "Unknown station Stall 99 in department 03",
|
|||
|
|
14: "Stamp A1 is taken by Andy Welder",
|
|||
|
|
}
|
|||
|
|
for line, text in expect.items():
|
|||
|
|
assert len(rows[line]["errors"]) == 1 and text in rows[line]["errors"][0], (line, rows[line]["errors"])
|
|||
|
|
|
|||
|
|
# Apply with errors: 422 with the same body, nothing changed
|
|||
|
|
r = _import(pg_client, bad, apply=True)
|
|||
|
|
assert r.status_code == 422 and r.json()["code"] == "import_errors"
|
|||
|
|
assert r.json()["rows"] == body["rows"] and r.json()["applied"] is False
|
|||
|
|
assert _profiles(pg_settings) == before
|
|||
|
|
|
|||
|
|
# A clean file: a stamp moves from Wes to Lou, Wes gets a new one, limits set and cleared, a profile goes off
|
|||
|
|
with app_conn(pg_settings) as conn:
|
|||
|
|
conn.execute(
|
|||
|
|
"INSERT INTO user_station_limits (user_id, station_id) SELECT u.id, s.id FROM users u, stations s"
|
|||
|
|
" WHERE u.sam_account = 'wh.wade' AND s.dept_code = '01' AND s.name = 'Station 1'"
|
|||
|
|
)
|
|||
|
|
clean = (
|
|||
|
|
"Employee_ID,Welder_Stamp,Active,Stations\n"
|
|||
|
|
"3003,A1,yes,03:Stall 1;03:stall 2\n"
|
|||
|
|
"3002,W9,,\n"
|
|||
|
|
"2003,W10,,\n"
|
|||
|
|
"6001,,no,-\n"
|
|||
|
|
"0042,-,,04:Stall 3\n"
|
|||
|
|
"5001,,,\n"
|
|||
|
|
)
|
|||
|
|
preview = _import(pg_client, clean).json()
|
|||
|
|
assert preview["errors"] == 0 and preview["ok"] == 6, preview
|
|||
|
|
assert {x["line"]: x["changes"] for x in preview["rows"]}[7] == {}
|
|||
|
|
r = _import(pg_client, clean, apply=True)
|
|||
|
|
assert r.status_code == 200, r.text
|
|||
|
|
assert r.json()["applied"] is True
|
|||
|
|
after = _profiles(pg_settings)
|
|||
|
|
assert after["weld.andy"] == ("A1", True, ["03:Stall 1", "03:Stall 2"])
|
|||
|
|
assert after["lead.lou"][0] == "W9" and after["sup.wes"][0] == "W10"
|
|||
|
|
assert after["wh.wade"] == (None, False, [])
|
|||
|
|
assert after["zero.zed"] == (None, True, ["04:Stall 3"])
|
|||
|
|
assert after["qc.quinn"] == before["qc.quinn"]
|
|||
|
|
|
|||
|
|
with app_conn(pg_settings) as conn:
|
|||
|
|
acts = conn.execute(
|
|||
|
|
"SELECT action, subject_id, detail FROM audit_events WHERE action LIKE 'admin.profile%' ORDER BY id"
|
|||
|
|
).fetchall()
|
|||
|
|
assert [a[0] for a in acts] == ["admin.profile_updated"] * 5 + ["admin.profile_import"]
|
|||
|
|
assert acts[-1][2] == {"rows": 6, "changed": 5, "unchanged": 1}
|
|||
|
|
assert all(a[2]["source"] == "import" for a in acts[:5])
|
|||
|
|
wade = next(a for a in acts if a[0] == "admin.profile_updated" and a[2]["after"]["profile_active"] is False)
|
|||
|
|
assert wade[2]["before"]["profile_active"] is True and wade[2]["before"]["station_limits"] != {}
|
|||
|
|
|
|||
|
|
# Over 1 MB is refused before it is read
|
|||
|
|
r = _import(pg_client, "employee_id\n" + "1" * (profiles.CSV_MAX_BYTES + 10))
|
|||
|
|
assert r.status_code == 413
|