From 74352ec6b8e9e8b44138b1d027cfaba5046d6767 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sat, 31 Oct 2026 14:52:30 +0900 Subject: [PATCH] =?UTF-8?q?feat(=ED=9A=8C=EC=82=AC=EC=97=90=20=EC=98=AC?= =?UTF-8?q?=EB=A6=AC=EA=B8=B0):=20=EC=9A=94=EC=B2=AD=20=EC=9E=A5=EB=B6=80?= =?UTF-8?q?=20=C2=B7=20=EC=9A=94=EC=B2=AD=20=C2=B7=20=EB=AA=A9=EB=A1=9D=20?= =?UTF-8?q?=C2=B7=20=EC=8A=B9=EC=9D=B8=20=C2=B7=20=EA=B1=B0=EC=A0=88=20?= =?UTF-8?q?=C2=B7=20=EC=95=8C=EB=A6=BC=20=EB=B4=84=20=EA=B8=B8=20+=20?= =?UTF-8?q?=EC=9E=84=EC=8B=9C=20storage=20=EC=8B=9C=ED=97=98=20(PLAN=20706?= =?UTF-8?q?-1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 회사 층 폴더 올리기요청.json 하나 · DB 안 바꿈 - 승인 = 「회사로 복사…」와 같은 길(구조물 = 붙여넣기 · 사전조합 = 서고 복사 + 폴더 자리) · 원본 개인 것 그대로 - 권한 = M02_Layer_Access 재사용 · 일반 승인 403 · 다른 회사 장부 안 봄 · 같은 것 다시 요청 409 Co-Authored-By: Claude Opus 5.5 --- M02_MasterTemplete/M02_Share_Request.py | 243 +++++++++++++++++++ main.py | 5 + resources/tester/test_706_1_share_request.py | 129 ++++++++++ 3 files changed, 377 insertions(+) create mode 100644 M02_MasterTemplete/M02_Share_Request.py create mode 100644 resources/tester/test_706_1_share_request.py diff --git a/M02_MasterTemplete/M02_Share_Request.py b/M02_MasterTemplete/M02_Share_Request.py new file mode 100644 index 000000000..d2ce53c8f --- /dev/null +++ b/M02_MasterTemplete/M02_Share_Request.py @@ -0,0 +1,243 @@ +"""회사에 올리기 요청 · 승인(PLAN 706-1 · 요구서 4장 6) — 일반 사용자가 개인 층 항목 · +사전조합을 회사 층에 올려 달라고 요청 · 회사 관리자가 승인(= 「회사로 복사…」와 같은 길 · +원본은 그대로) · 거절. + +장부 = 회사 층 폴더 `올리기요청.json` 하나(DB 안 씀) · `{줄: [요청 …]}`. +요청 = `{id, 종류(item · precombo), 키, 이름, 경로(항목 폴더), 주인 C/U, 요청자, 요청자이름, +글, 날짜, 상태(요청 · 승인 · 거절), 처리자, 처리날짜, 사유, 사본, 봄}` · `봄` = 요청자가 결과 +알림을 닫음. 권한 = `M02_Layer_Access` 한 벌 — 요청 = 자기 개인 층 · 목록 = 회사 관리자는 +그 회사 전부 · 일반은 자기 것 · +승인 · 거절 = 회사 층 쓰기(일반 403) · 다른 회사 장부는 안 봄(시스템 관리자만 `company`). +같은 항목(주인 · 종류 · 키)이 「요청」 상태면 다시 요청 = 409 「요청 중」. +⚠ 등록은 `main.py`(로그인만). +""" + +from __future__ import annotations + +import json +import threading +import uuid +from datetime import datetime +from pathlib import Path +from typing import Any, Literal + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel + +from common_util.common_util_auth import verify_session +from common_util.common_util_json import atomic_write_json +from M02_MasterTemplete import M02_Item_Paste as paste_store +from M02_MasterTemplete import M02_Labor_Ref as refs +from M02_MasterTemplete import M02_Layer_Access as access +from M02_MasterTemplete import M02_MasterTemplete_Items as items_store +from M02_MasterTemplete import M02_MasterTemplete_Store as store +from M02_MasterTemplete import M02_Precombo_Folders as pre_folders +from M02_MasterTemplete.M02_MasterTemplete_Router_Items import ( + NotedRoute, + Session, + _call, + _root, +) + +router = APIRouter(prefix="/api/m02", tags=["M02 Share Request"], route_class=NotedRoute) + +FILE = "올리기요청.json" +# ponytail: 한 프로세스 잠금 — 서버를 여럿 띄우면 파일 잠금으로 +_LOCK = threading.Lock() + + +class AskBody(BaseModel): + 종류: Literal["item", "precombo"] + 키: str + 글: str = "" + + +class ApproveBody(BaseModel): + 경로: list[str] = [] + + +class RejectBody(BaseModel): + 사유: str = "" + + +def _company(session: Session, company: str | None) -> str: + """장부 회사 — 시스템 관리자만 `company` 로 남의 회사 · 그 밖은 자기 회사(없으면 409).""" + if company and access.is_system_admin(session): + return str(company) + own = session.get("company_id") + if own is None: + raise HTTPException(status_code=409, detail=access.NO_COMPANY) + return str(own) + + +def _path(company: str) -> Path: + return access.folder(access.Grant("company", company, False, True)) / FILE + + +def _read(company: str) -> list[dict[str, Any]]: + path = _path(company) + if not path.is_file(): + return [] + return list(json.loads(path.read_bytes()).get("줄") or []) + + +def _write(company: str, rows: list[dict[str, Any]]) -> None: + atomic_write_json(_path(company), {"줄": rows}) + + +def _now() -> str: + return datetime.now().isoformat(timespec="seconds") + + +def _source(session: Session, kind: str, key: str) -> dict[str, Any]: + """자기 개인 층 원본 — `{이름, 경로}`(없으면 404).""" + folder = _root(session, "personal") + if kind == "item": + found = next((r for r in items_store.items(folder) if r.get("key") == key), None) + if found is None: + raise HTTPException(status_code=404, detail=f"없는 항목 「{key}」") + path = list(found.get("경로") or []) + return {"이름": path[-1] if path else key, "경로": path[:-1]} + owner = access.own_owner(session, "personal") + doc = ( + refs.read_ref({"층": "personal", "주인": owner, "키": key}) + if refs._KEY.match(key) + else None + ) + if doc is None or refs.kind_of(key) != "precombo": + raise HTTPException(status_code=404, detail=f"없는 사전조합 「{key}」") + return {"이름": str(doc.get("이름") or key), "경로": []} + + +@router.post("/share-requests") +def ask(body: AskBody, session: Session = Depends(verify_session)) -> dict[str, Any]: + """요청 — 자기 개인 층 것만 · 같은 것이 「요청」 이면 409 「요청 중」.""" + owner = access.own_owner(session, "personal") + if owner is None: + raise HTTPException(status_code=409, detail=access.NO_COMPANY) + src = _source(session, body.종류, body.키) + company = owner.split("/")[0] + row = { + "id": uuid.uuid4().hex[:12], + "종류": body.종류, + "키": body.키, + **src, + "주인": owner, + "요청자": session.get("user_id"), + "요청자이름": session.get("name") or "", + "글": body.글.strip()[:200], + "날짜": _now(), + "상태": "요청", + } + with _LOCK: + rows = _read(company) + same = (owner, body.종류, body.키) + if any((r["주인"], r["종류"], r["키"]) == same and r["상태"] == "요청" for r in rows): + raise HTTPException(status_code=409, detail="요청 중 — 회사 관리자가 아직 안 봄") + _write(company, [*rows, row]) + return row + + +@router.get("/share-requests") +def listing( + company: str | None = None, session: Session = Depends(verify_session) +) -> dict[str, Any]: + """목록 — 회사를 맡은 사람 = 그 회사 전부 · 그 밖 = 자기 요청만 · `관리` = 승인할 수 있나.""" + at = _company(session, company) + rows = _read(at) + boss = access.manages(session, at) + if not boss: + me = session.get("user_id") + rows = [r for r in rows if r.get("요청자") == me] + return {"회사": at, "관리": boss, "줄": rows} + + +def _settle(session: Session, company: str | None, rid: str, done: dict[str, Any]) -> dict: + """요청 한 줄을 끝냄 — 회사 층 쓰기 권한(일반 403) · 「요청」 상태만(아니면 409).""" + at = _company(session, company) + access.require(session, "company", at, write=True) + with _LOCK: + rows = _read(at) + row = next((r for r in rows if r.get("id") == rid), None) + if row is None: + raise HTTPException(status_code=404, detail="없는 요청") + if row["상태"] != "요청": + raise HTTPException(status_code=409, detail=f"이미 {row['상태']}") + extra = done.pop("run", None) + if extra: + done["사본"] = extra(row, at) + row.update({**done, "처리자": session.get("name") or "", "처리날짜": _now(), "봄": False}) + _write(at, rows) + return row + + +def _copy(session: Session, dest: list[str]): + """승인 = 「회사로 복사…」와 같은 길 — 구조물 항목 = 붙여넣기 · + 사전조합 = 서고 복사(+ 폴더 자리).""" + + def run(row: dict[str, Any], company: str) -> str: + target = _root(session, "company", company, write=True) + source = _root(session, "personal", row["주인"]) + if row["종류"] == "item": + ref = [{"경로": row.get("경로") or [], "키": row["키"]}] + origin = {"layer": "personal", "owner": row["주인"]} + try: + note = paste_store.paste(source, target, ref, dest, origin) + except store.StoreError as e: + raise HTTPException(status_code=e.status, detail=e.detail) from e + return str(note["열"][0]) + ref = {"층": "personal", "주인": row["주인"], "키": row["키"]} + doc = refs.read_ref(ref) + if doc is None: + raise HTTPException(status_code=404, detail=f"원본이 없음 「{row['키']}」") + + def make() -> dict[ + str, Any + ]: # 못 읽는 참조 같이 복사 = 받는 층 안에서(`new_library` 와 같음) + return refs.create("precombo", refs.carry(refs.origin_mark(doc, ref), here)) + + here = ("personal", row["주인"]) + made = _call(target, make) + if dest: + view = _call(target, pre_folders.view) + move = {"op": "move", "키": made["이름"], "새경로": dest, "판": view["판"]} + _call(target, pre_folders.edit, move) + return str(made["이름"]) + + return run + + +@router.post("/share-requests/{rid}/approve") +def approve( + rid: str, + body: ApproveBody, + company: str | None = None, + session: Session = Depends(verify_session), +) -> dict[str, Any]: + """승인 — 회사 층 `경로` 폴더에 사본(새 키 · 같은 이름 「(1)」) · 원본 개인 것은 그대로.""" + return _settle(session, company, rid, {"상태": "승인", "run": _copy(session, body.경로)}) + + +@router.post("/share-requests/{rid}/reject") +def reject( + rid: str, + body: RejectBody, + company: str | None = None, + session: Session = Depends(verify_session), +) -> dict[str, Any]: + return _settle(session, company, rid, {"상태": "거절", "사유": body.사유.strip()[:200]}) + + +@router.post("/share-requests/seen") +def seen(session: Session = Depends(verify_session)) -> dict[str, int]: + """요청자가 결과 알림 줄을 닫음 — 자기 요청 가운데 끝난 것 `봄` = true.""" + at = _company(session, None) + me = session.get("user_id") + with _LOCK: + rows = _read(at) + hit = [r for r in rows if r.get("요청자") == me and r.get("봄") is False] + for r in hit: + r["봄"] = True + if hit: + _write(at, rows) + return {"봄": len(hit)} diff --git a/main.py b/main.py index aef378cd7..07bca3100 100644 --- a/main.py +++ b/main.py @@ -547,6 +547,11 @@ app.include_router(m02_items_router, dependencies=protected) app.include_router(m02_combos_router, dependencies=protected) app.include_router(m02_drawing_router, dependencies=protected) app.include_router(m02_labor_router, dependencies=protected) +# 회사에 올리기 요청 · 승인(PLAN 706-1) — 로그인만 · 층 권한은 길 안에서 +app.include_router( + __import__("M02_MasterTemplete.M02_Share_Request", fromlist=["router"]).router, + dependencies=protected, +) # 프로젝트 집계 항목 복사본(PLAN 25-2b) — 로그인만 · 프로젝트 권한은 길 안에서. app.include_router(m02_project_items_router, dependencies=protected) # 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근). diff --git a/resources/tester/test_706_1_share_request.py b/resources/tester/test_706_1_share_request.py new file mode 100644 index 000000000..edb36215f --- /dev/null +++ b/resources/tester/test_706_1_share_request.py @@ -0,0 +1,129 @@ +"""회사에 올리기 요청 · 승인(PLAN 706-1 · `M02_Share_Request`) — 임시 storage. + +① 일반 사용자 요청(구조물 항목 · 사전조합) → 장부 한 줄 · 같은 것 다시 = 409 「요청 중」. +② 회사 관리자 승인 = 회사 층 고른 폴더에 사본(원본 개인 것 그대로) · 거절 = 사유 · 요청자 알림 → 봄. +③ 권한 — 일반 승인 403 · 다른 회사 관리자는 그 요청을 못 봄(404) · 남의 회사 장부 안 봄 · + 목록은 자기 것만. +""" + +from __future__ import annotations + +import json +from typing import Any + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from common_util.common_util_auth import verify_session +from M02_MasterTemplete import M02_Labor_Router as labor_router +from M02_MasterTemplete import M02_MasterTemplete_Router_Items as items_router +from M02_MasterTemplete import M02_MasterTemplete_Router_Layers as layers_router +from M02_MasterTemplete import M02_Share_Request as share +from resources.tester import test_m02_tree as base +from resources.tester.test_m02_tree import Tree + +world = base.world # 같은 임시 storage 판 + +URL = "/api/m02/share-requests" + + +def _client(world: dict[str, Any]) -> TestClient: + app = FastAPI() + for one in (layers_router, items_router, labor_router, share): + app.include_router(one.router) + app.dependency_overrides[verify_session] = lambda: world["session"] + return TestClient(app) + + +def _as(world: dict[str, Any], role: str = "USER", user: int = 42, company: int = 7) -> None: + world["session"].update(role=role, user_id=user, company_id=company, name=f"사람{user}") + + +def _mine(world: dict[str, Any], client: TestClient) -> tuple[str, str]: + """일반 사용자 42 개인 층 — 항목 「배수/sub7임시」 · 사전조합 하나 → (항목 키, 사전조합 키).""" + _as(world) + t = Tree(client, "personal") + t.do(op="add-folder", 경로=[], 이름="배수") + t.do(op="add-item", 경로=["배수"], 이름="sub7임시", 단위="m") + doc = { + "문서": {"이름": "sub7임시 사전", "구분": "C 사면안정", "상세구분": "옹벽", "담은로직": []} + } + pre = client.post("/api/m02/layers/personal/library/precombo", json=doc).json()["이름"] + return t.item(["배수", "sub7임시"])["key"], pre + + +def test_요청_승인_거절_알림(world): + client = _client(world) + item, pre = _mine(world, client) + got = client.post(URL, json={"종류": "item", "키": item, "글": " 회사 공용으로 "}) + assert got.status_code == 200, got.text + one = got.json() + assert (one["이름"], one["경로"], one["주인"], one["글"]) == ( + "sub7임시", + ["배수"], + "7/42", + "회사 공용으로", + ) + assert client.post(URL, json={"종류": "item", "키": item}).status_code == 409 # 요청 중 + two = client.post(URL, json={"종류": "precombo", "키": pre}).json() + assert client.post(URL, json={"종류": "item", "키": "SP42-999999"}).status_code == 404 + assert client.post(URL, json={"종류": "precombo", "키": "SC-999999"}).status_code == 404 + mine = client.get(URL).json() + assert mine["관리"] is False and [r["id"] for r in mine["줄"]] == [one["id"], two["id"]] + + _as(world, "ADMIN", user=50) # 회사 7 관리자 + company = Tree(client, "company") + company.do(op="add-folder", 경로=[], 이름="공용") + boss = client.get(URL).json() + assert boss["관리"] is True and len(boss["줄"]) == 2 + ok = client.post(f"{URL}/{one['id']}/approve", json={"경로": ["공용"]}) + assert ok.status_code == 200, ok.text + assert ok.json()["상태"] == "승인" and ok.json()["봄"] is False + company = Tree(client, "company") + assert ["공용", "sub7임시"] in company.paths() # 고른 폴더에 사본 + again = client.post(f"{URL}/{one['id']}/approve", json={"경로": []}) + assert again.status_code == 409 # 이미 승인 + no = client.post(f"{URL}/{two['id']}/reject", json={"사유": "이름 다시"}) + assert no.json()["상태"] == "거절" and no.json()["사유"] == "이름 다시" + + _as(world) # 요청자 — 원본 그대로 · 결과 둘 · 봄 뒤 0 + assert ["배수", "sub7임시"] in Tree(client, "personal").paths() + rows = client.get(URL).json()["줄"] + assert [(r["상태"], r["봄"]) for r in rows] == [("승인", False), ("거절", False)] + assert client.post(f"{URL}/seen").json() == {"봄": 2} + assert client.post(f"{URL}/seen").json() == {"봄": 0} + # 거절된 것은 다시 요청할 수 있음 + assert client.post(URL, json={"종류": "precombo", "키": pre}).status_code == 200 + + +def test_사전조합_승인_폴더_자리(world): + client = _client(world) + _, pre = _mine(world, client) + rid = client.post(URL, json={"종류": "precombo", "키": pre}).json()["id"] + _as(world, "ADMIN", user=50) + tree_url = "/api/m02/layers/company/precombo-tree" + now = client.get(tree_url).json() + client.post(tree_url, json={"op": "add-folder", "경로": [], "이름": "옹벽", "판": now["판"]}) + made = client.post(f"{URL}/{rid}/approve", json={"경로": ["옹벽"]}).json()["사본"] + assert client.get(tree_url).json()["자리"] == {made: ["옹벽"]} + doc = client.get(f"/api/m02/layers/company/templates/precombo/{made}").json()["문서"] + assert doc["이름"] == "sub7임시 사전" + assert doc["본뜬"]["주인"] == "7/42" # 원본 표시 + + +def test_권한_일반_승인_403_다른_회사(world): + client = _client(world) + item, _ = _mine(world, client) + rid = client.post(URL, json={"종류": "item", "키": item}).json()["id"] + assert client.post(f"{URL}/{rid}/approve", json={}).status_code == 403 # 일반 = 승인 못 함 + assert client.post(f"{URL}/{rid}/reject", json={}).status_code == 403 + _as(world, "USER", user=43) # 같은 회사 다른 일반 — 남의 요청 안 보임 + assert client.get(URL).json()["줄"] == [] + _as(world, "ADMIN", user=90, company=9) # 다른 회사 관리자 + assert client.get(URL).json()["줄"] == [] + assert client.post(f"{URL}/{rid}/approve", json={}).status_code == 404 # 자기 장부에 없음 + assert client.get(URL, params={"company": "7"}).json()["회사"] == "9" # 남의 회사 고르기 무시 + _as(world, "SYSTEM_ADMIN", user=1, company=None) + assert len(client.get(URL, params={"company": "7"}).json()["줄"]) == 1 + ledger = world["storage"] / "7" / "templates" / share.FILE + assert json.loads(ledger.read_bytes())["줄"][0]["상태"] == "요청"