Merge remote-tracking branch 'origin/dev' into sub_laptop_2
This commit is contained in:
@@ -0,0 +1,235 @@
|
||||
"""Z01 공종 편집 — 규칙(순수 함수) · 요청 모델 · 저장 흐름(409·422·전부 아니면 전무) · 등록.
|
||||
|
||||
(2026-09-18 브레인 PLAN 1-0 계약 · sub_laptop_3) DB 없이 — 저장 흐름은 가짜 커넥션·저장소로 훑음.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from contextlib import asynccontextmanager
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.routing import APIRoute
|
||||
from fastapi.testclient import TestClient
|
||||
from pydantic import ValidationError
|
||||
|
||||
from common_util.common_util_auth import require_system_admin, verify_session
|
||||
from Z01_MasterData import Z01_MasterData_Router_WorkItems as router_module
|
||||
from Z01_MasterData import Z01_MasterData_WorkItems as work_items
|
||||
|
||||
SEED_ROW = {"resource_kind": "labor", "resource_key": "L-001", "amount": 1.5, "sort_order": 0}
|
||||
|
||||
|
||||
def _row(row_id, seed=SEED_ROW, **values):
|
||||
base = {"id": row_id, "variant": "", "resource_kind": "labor", "resource_key": "L-001"}
|
||||
return {**base, "amount": Decimal("1.500000"), "sort_order": 0, "seed": seed, **values}
|
||||
|
||||
|
||||
# ── 규칙 ──
|
||||
|
||||
|
||||
def test_고친_칸은_초기값과_값으로_비교함() -> None:
|
||||
assert work_items.changed_columns(_row(1), SEED_ROW, work_items.ROW_FIELDS) == []
|
||||
changed = _row(1, amount=Decimal("2"))
|
||||
assert work_items.changed_columns(changed, SEED_ROW, work_items.ROW_FIELDS) == ["amount"]
|
||||
assert work_items.changed_columns(_row(1, seed=None), None, work_items.ROW_FIELDS) == []
|
||||
|
||||
|
||||
def test_검사는_남의_줄_겹친_줄_없는_자원을_막음() -> None:
|
||||
rows = [
|
||||
{"id": 1, "resource_key": "L-001"},
|
||||
{"id": 1, "resource_key": None},
|
||||
{"id": 99, "resource_key": None},
|
||||
{"id": None, "resource_key": "없는열쇠"},
|
||||
]
|
||||
errors = work_items.check_rows(rows, existing_ids={1, 2}, known_keys={"L-001"})
|
||||
assert len(errors) == 3
|
||||
assert "두 번" in errors[0] and "id 99" in errors[1] and "없는열쇠" in errors[2]
|
||||
assert work_items.check_rows(rows[:1], {1}, {"L-001"}) == []
|
||||
|
||||
|
||||
def test_전체_줄_보내기는_바뀐_줄만_쓰고_빠진_줄은_숨기거나_지움() -> None:
|
||||
existing = [_row(1), _row(2, sort_order=1), _row(3, seed=None, sort_order=2)]
|
||||
same = {f: existing[0].get(f) for f in work_items.ROW_FIELDS} | {"id": 1, "amount": 1.5}
|
||||
new = {"id": None, "resource_kind": "material", "resource_key": None, "amount": None}
|
||||
plan = work_items.plan_rows(existing, [same, new])
|
||||
assert plan["updates"] == [] # 1.500000 = 1.5 · 순서도 그대로
|
||||
assert plan["inserts"][0]["resource_kind"] == "material"
|
||||
assert plan["inserts"][0]["sort_order"] == 1
|
||||
assert plan["hide"] == [2] # 초기값 줄 — 숨김(되살릴 수 있게)
|
||||
assert plan["drop"] == [3] # 관리자 추가 줄 — 지움
|
||||
|
||||
moved = work_items.plan_rows(existing[:2], [{**same, "id": 2}, same])
|
||||
assert [u["id"] for u in moved["updates"]] == [2, 1] # 순서만 바뀌어도 sort_order 를 씀
|
||||
|
||||
|
||||
def test_초기값으로는_seed_칸만_살리고_추가_줄은_지움() -> None:
|
||||
head = {"id": 7, "name": "고친 이름", "seed": {"name": "원 이름", "status": "partial"}}
|
||||
rows = [_row(1, amount=Decimal("9")), _row(2, is_deleted=1), _row(3, seed=None)]
|
||||
plan = work_items.reset_plan(head, rows)
|
||||
assert plan["head"] == {"name": "원 이름", "status": "partial"}
|
||||
assert [u["id"] for u in plan["updates"]] == [1, 2]
|
||||
assert plan["updates"][0]["amount"] == 1.5
|
||||
assert plan["drop"] == [3] and plan["inserts"] == [] and plan["hide"] == []
|
||||
|
||||
|
||||
def test_상세는_seed_를_안_내보내고_고친_칸만_알림() -> None:
|
||||
head = {"id": 7, "work_item_key": "FW-00001", "name": "새", "status": "partial"}
|
||||
head |= {"seed": {"name": "옛"}, "source": [{"pum_table_id": "T1"}], "notes": ["[주] 1"]}
|
||||
out = work_items.detail(head, [_row(1), _row(2, seed=None)], revision=4)
|
||||
assert out["item"]["revision"] == 4 and out["item"]["changed_columns"] == ["name"]
|
||||
assert "seed" not in out["item"] and "id" not in out["item"]
|
||||
assert out["source"] == [{"pum_table_id": "T1"}] and out["notes"] == ["[주] 1"]
|
||||
assert [r["is_added"] for r in out["rows"]] == [False, True]
|
||||
assert out["rows"][0]["id"] == 1 and "seed" not in out["rows"][0]
|
||||
|
||||
|
||||
# ── 요청 모델 ──
|
||||
|
||||
|
||||
def _body(**row):
|
||||
item = {"name": "공종", "status": "reviewed"}
|
||||
return {"base_revision": 3, "item": item, "rows": [{"resource_kind": "labor", **row}]}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"row",
|
||||
[{"amount": -1}, {"resource_key": ""}, {"resource_kind": "oil"}, {"unknown": 1}],
|
||||
)
|
||||
def test_요청_모델은_음수_빈열쇠_모르는_종류_모르는_칸을_막음(row) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
router_module.WorkItemSave.model_validate(_body(**row))
|
||||
|
||||
|
||||
# ── 저장 흐름(가짜 커넥션) ──
|
||||
|
||||
|
||||
class _Connection:
|
||||
def __init__(self) -> None:
|
||||
self.log: list[str] = []
|
||||
|
||||
async def begin(self) -> None:
|
||||
self.log.append("begin")
|
||||
|
||||
async def commit(self) -> None:
|
||||
self.log.append("commit")
|
||||
|
||||
async def rollback(self) -> None:
|
||||
self.log.append("rollback")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch):
|
||||
connection = _Connection()
|
||||
|
||||
class _Pool:
|
||||
@asynccontextmanager
|
||||
async def acquire(self):
|
||||
yield connection
|
||||
|
||||
head = {"id": 7, "work_item_key": "FW-00001", "name": "공종", "status": "partial"}
|
||||
head |= {"seed": {"name": "공종"}, "source": [], "notes": []}
|
||||
|
||||
async def fetch_item(_c, key, lock=False):
|
||||
return dict(head) if key == "FW-00001" else None
|
||||
|
||||
async def fetch_rows(_c, _item_id, with_deleted=False):
|
||||
return [_row(1)]
|
||||
|
||||
async def revision(_c, _scope, lock=False):
|
||||
return 3
|
||||
|
||||
async def known_resource_keys(_c, keys):
|
||||
return keys & {"L-001"}
|
||||
|
||||
async def write(_c, *args):
|
||||
connection.log.append("write")
|
||||
|
||||
async def bump_revision(_c, _scope):
|
||||
connection.log.append("bump")
|
||||
|
||||
async def list_items(_c, **picked):
|
||||
connection.log.append(picked["status"])
|
||||
return {"items": [], "total": 2, "counts": {"partial": 2}}
|
||||
|
||||
monkeypatch.setattr(router_module, "get_db_pool", lambda: _Pool())
|
||||
for name, fake in {
|
||||
"list_items": list_items,
|
||||
"fetch_item": fetch_item,
|
||||
"fetch_rows": fetch_rows,
|
||||
"revision": revision,
|
||||
"known_resource_keys": known_resource_keys,
|
||||
"write": write,
|
||||
"bump_revision": bump_revision,
|
||||
}.items():
|
||||
monkeypatch.setattr(router_module.repo, name, fake)
|
||||
app = FastAPI()
|
||||
app.include_router(router_module.router)
|
||||
app.dependency_overrides[verify_session] = lambda: {"user_id": 42}
|
||||
test_client = TestClient(app)
|
||||
test_client.log = connection.log
|
||||
return test_client
|
||||
|
||||
|
||||
URL = "/api/master-data/work-items/FW-00001"
|
||||
|
||||
|
||||
def test_낡은_revision_은_409_이고_아무것도_안_씀(client) -> None:
|
||||
response = client.put(URL, json={**_body(id=1, resource_key="L-001"), "base_revision": 2})
|
||||
assert response.status_code == 409
|
||||
assert client.log == ["begin", "rollback"]
|
||||
|
||||
|
||||
def test_없는_자원_열쇠는_422_이고_아무것도_안_씀(client) -> None:
|
||||
response = client.put(URL, json=_body(resource_key="없는열쇠"))
|
||||
assert response.status_code == 422 and "없는열쇠" in response.json()["detail"]
|
||||
assert client.log == ["begin", "rollback"]
|
||||
|
||||
|
||||
def test_맞는_저장은_쓰고_revision_올리고_새_상세를_줌(client) -> None:
|
||||
response = client.put(URL, json=_body(id=1, resource_key="L-001", amount=2))
|
||||
assert response.status_code == 200, response.text
|
||||
assert client.log == ["begin", "write", "bump", "commit"]
|
||||
assert response.json()["status"] == "success" and response.json()["item"]["revision"] == 3
|
||||
|
||||
|
||||
def test_목록은_상태별_개수를_다_채우고_모르는_상태는_막음(client) -> None:
|
||||
body = client.get("/api/master-data/work-items?status=").json()
|
||||
assert body["counts"] == {s: 2 if s == "partial" else 0 for s in work_items.STATUSES}
|
||||
assert client.get("/api/master-data/work-items?status=needs_input").status_code == 200
|
||||
assert client.get("/api/master-data/work-items?status=bogus").status_code == 422
|
||||
assert client.log == ["", "needs_input"]
|
||||
|
||||
|
||||
def test_초기값으로와_없는_공종(client) -> None:
|
||||
assert client.post(URL + "/reset").status_code == 200
|
||||
assert client.log == ["begin", "write", "bump", "commit"]
|
||||
assert client.get("/api/master-data/work-items/FW-99999").status_code == 404
|
||||
|
||||
|
||||
# ── 등록 ──
|
||||
|
||||
|
||||
def _calls(dependant) -> set:
|
||||
found = set()
|
||||
for sub in dependant.dependencies:
|
||||
found.add(sub.call)
|
||||
found |= _calls(sub)
|
||||
return found
|
||||
|
||||
|
||||
def test_공종_편집_경로는_시스템_관리자만() -> None:
|
||||
app = importlib.import_module("main").app
|
||||
routes = {(r.path, m) for r in app.routes if isinstance(r, APIRoute) for m in r.methods}
|
||||
wanted = {
|
||||
("/api/master-data/work-items", "GET"),
|
||||
("/api/master-data/work-items/{key}", "GET"),
|
||||
("/api/master-data/work-items/{key}", "PUT"),
|
||||
("/api/master-data/work-items/{key}/reset", "POST"),
|
||||
}
|
||||
assert wanted <= routes
|
||||
for route in app.routes:
|
||||
if isinstance(route, APIRoute) and route.path.startswith("/api/master-data/work-items"):
|
||||
assert {verify_session, require_system_admin} <= _calls(route.dependant)
|
||||
Reference in New Issue
Block a user