Aislo/resources/tester/test_718_2_alias_gather.py

"""별칭 장부 시스템 모으기(PLAN 718-2 · `M01_MasterData_Alias`) — 회사 · 개인 층 맞춤 +
내역맞춤 모음 · 같은 짝 묶기 · 회사 수 · 사람 수(익명) · 충돌 · 승인 → 마스터 줄 ·
반려 기억 · 다른 회사 새 사용자가 마스터 줄을 씀
· 시스템 관리자만 · master_data 원본 그대로. 층 · storage 는 임시 폴더.
"""

import json
from pathlib import Path

import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

from B09_Estimation import B09_Estimation_Bill_Link as lk
from B09_Estimation import B09_Estimation_Router_Link as route
from common_util.common_util_auth import require_system_admin, verify_session
from M01_MasterData import M01_MasterData_Alias as alias
from M02_MasterTemplete import M02_Hopyo_Ledger as ledger
from M02_MasterTemplete import M02_Hopyo_Store as hs
from M02_MasterTemplete import M02_Template_Layers as layers

# 올리기 길 시험의 `client` 마당(임시 storage · 마스터 층)을 같이 씀
# ruff: noqa: F811
from resources.tester.test_m02_hopyo_excel import client, master_hash  # noqa: F401

DIG = {"이름": "구조물터파기(인력100%)", "규격": "토사", "단위": "㎥"}
DIG_KEY = "구조물터파기(인력100%)|토사|m3"
FORM_KEY = "합판거푸집|4회|m2"
CUT_KEY = "토사절취||m3"
ROOT = Path(__file__).resolve().parents[2]


def _pair(code):
    return ledger.check(DIG_KEY, {"갈래": "짝", "짝": [{"로직": code, "배": 1}]})


def _seed():
    """회사 7(개인 42 · 43 = 같은 짝) · 회사 8(개인 50 = 다른 짝 · 회사 층 = 안 씀) · 내역맞춤."""
    for user, who in ((42, "갑"), (43, "을")):
        mine = layers.personal_dir(7, user)
        ledger.write(mine, {DIG_KEY: _pair("GF000101")}, {"id": user, "이름": who}, "가공사.xls")
    ledger.write(layers.personal_dir(8, 50), {DIG_KEY: _pair("GF000103")}, {"id": 50}, "나.xls")
    skip = ledger.check(FORM_KEY, {"갈래": "안 씀"})
    ledger.write(layers.company_dir(8), {FORM_KEY: skip}, {"id": 51, "이름": "병"}, "다.xls")
    live = {"확정": "earth:4", "이름": "절토(토사)"}
    hs.remember(layers.personal_dir(7, 42), {CUT_KEY: live, "a||": {lk.FORGOT: True}}, lk.MEMORY)
    hs.remember(layers.company_dir(8), {CUT_KEY: live}, lk.MEMORY)


def _one(got, kind, key, code=None):
    rows = [x for x in got["후보"] if (x["종류"], x["열쇠"]) == (kind, key)]
    if code:
        rows = [x for x in rows if x["짝"][0]["로직"] == code]
    assert len(rows) == 1, rows
    return rows[0]


def test_모음_묶기_익명_충돌(client):
    before = master_hash()
    _seed()
    got = alias.gather()
    assert got["모은곳"] == {"회사": 2, "사람": 5} and got["반려"] == 0
    assert len(got["후보"]) == 4  # 터파기 둘(충돌) · 거푸집 안 씀 · 토사절취 내역맞춤(같은 짝 하나)
    same = _one(got, alias.ALIAS, DIG_KEY, "GF000101")
    assert (same["회사"], same["사람"], same["충돌"], same["마스터"]) == (1, 2, True, None)
    assert _one(got, alias.ALIAS, DIG_KEY, "GF000103")["사람"] == 1
    form = _one(got, alias.ALIAS, FORM_KEY)
    assert (form["갈래"], form["충돌"], form["사람"]) == ("안 씀", False, 1)
    cut = _one(got, alias.BILL, CUT_KEY)
    assert (cut["회사"], cut["사람"], cut["확정"], cut["충돌"]) == (2, 2, "earth:4", False)
    text = json.dumps(got, ensure_ascii=False)
    for secret in ("갑", "을", "병", "가공사.xls", "누가", "처음파일", "주인"):
        assert secret not in text  # 익명 — 회사 수 · 사람 수만
    assert master_hash() == before


def test_승인_마스터_줄_반려_기억(client, tmp_path, monkeypatch):
    before = master_hash()
    _seed()
    newbie = layers.personal_dir(9, 99)  # 처음 쓰는 다른 회사 사용자
    assert ledger.recall(DIG, newbie) is None
    sig = _one(alias.gather(), alias.ALIAS, DIG_KEY, "GF000101")["서명"]
    got = alias.approve(alias.ALIAS, DIG_KEY, sig, {"id": 1, "이름": "관리자"})
    hit = ledger.recall(DIG, newbie)
    assert (hit["층"], hit["짝"][0]["로직"]) == ("system", "GF000101")  # 승인 → 모두가 씀
    master = ledger.read(client.system)
    assert master[DIG_KEY]["처음파일"] is None  # 마스터 줄에도 남의 파일 이름 없음
    other = _one(got, alias.ALIAS, DIG_KEY)  # 승인한 짝 = 마스터와 같음 → 후보에서 빠짐
    assert other["짝"][0]["로직"] == "GF000103"
    assert (other["마스터"], other["충돌"]) == ("다름", True)
    got = alias.reject(alias.ALIAS, DIG_KEY, other["서명"], {"id": 1})
    assert got["반려"] == 1 and not [x for x in got["후보"] if x["열쇠"] == DIG_KEY]
    assert alias.gather()["반려"] == 1  # 반려 기억 = 다시 모아도 안 보임
    with pytest.raises(alias.store.StoreError) as error:
        alias.approve(alias.ALIAS, DIG_KEY, other["서명"], {"id": 1})
    assert error.value.status == 404
    with pytest.raises(alias.store.StoreError):
        alias.approve("모름", DIG_KEY, sig, {"id": 1})
    # 내역맞춤 승인 → 마스터 기억 → 다른 회사 새 사용자 B09 잇기 기억에 나옴
    cut = _one(alias.gather(), alias.BILL, CUT_KEY)
    alias.approve(alias.BILL, CUT_KEY, cut["서명"], {"id": 1})
    assert [r["열쇠"] for r in alias.master_bill()] == [CUT_KEY]
    monkeypatch.setattr(route.access, "folder", lambda got, create=False: tmp_path / got.layer)
    user = {"company_id": 9, "user_id": 99, "role": "USER"}
    assert route.recalled(user)[CUT_KEY]["확정"] == "earth:4"
    assert alias.forget_bill(CUT_KEY) == [] and CUT_KEY not in route.recalled(user)
    assert master_hash() == before


def test_길_시스템_관리자만(client):
    _seed()
    app = FastAPI()
    app.include_router(alias.router, dependencies=[Depends(require_system_admin)])
    who = {"user_id": 1, "company_id": None, "role": "USER", "name": "갑"}
    app.dependency_overrides[verify_session] = lambda: who
    web = TestClient(app)
    for role in ("USER", "ADMIN"):
        who["role"] = role
        assert web.get("/api/m01/alias-ledger").status_code == 403
    who["role"] = "SYSTEM_ADMIN"
    got = web.get("/api/m01/alias-ledger").json()
    assert len(got["후보"]) == 4 and got["마스터내역"] == []
    form = _one(got, alias.ALIAS, FORM_KEY)
    pick = {"종류": alias.ALIAS, "열쇠": FORM_KEY, "서명": form["서명"]}
    res = web.post("/api/m01/alias-ledger/approve", json=pick)
    assert res.status_code == 200 and len(res.json()["후보"]) == 3
    assert ledger.read(client.system)[FORM_KEY]["갈래"] == "안 씀"
    assert web.post("/api/m01/alias-ledger/approve", json=pick).status_code == 404
    assert web.post("/api/m01/alias-ledger/reject", json={**pick, "서명": "x"}).status_code == 422
    gone = web.post("/api/m01/alias-ledger/master-bill/forget", json={"열쇠": CUT_KEY})
    assert gone.status_code == 404
    main = (ROOT / "main.py").read_text(encoding="utf-8")
    at = main.index("M01_MasterData.M01_MasterData_Alias")
    assert "dependencies=system_admin_only" in main[at : at + 120]