eomsangdonandClaude Opus 5
018ebe7f40
계획서 0-8 의 남은 하나. 서버는 분석이 도는 동안 새 자료를 이미 막고 있었고 (`is_analysis_running`), 없던 것은 **막힌 까닭을 사람에게 보이는 것**이었음. 여럿이 한 프로젝트를 보면 「왜 안 올라가지」가 됨. - 분석을 띄우는 **경로 넷 모두** 시작한 사람을 `params.started_by` 로 담음. - `analysis_lock_owner` 가 그 id 로 이름을 붙여 냄 — 옛 자료는 id 가 없어 **이름 없이 잠금만** 돎(막는 것이 먼저). - `/upload-overview` 응답에 `analysis_lock` 을 실어 B03 화면이 띠로 보임. 화면 실측 — 「엄상돈 님이 올린 자료를 분석하는 중입니다」 띠가 경고색으로 뜸(응답을 잠깐 가로채 확인하고 곧바로 되돌림). 지금 도는 분석이 없어 실제 잠금 화면은 다음 업로드 때 볼 것. 시험 넷 추가(도는 중 아님·이름 붙음·옛 자료·경로 넷 대조), 전체 1211 통과·실패 0. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
"""업로드 잠금 표시 — **지금 누가 올리는 중인지** (2026-09-09, 계획서 0-8 남은 하나).
|
|
|
|
서버는 분석이 도는 동안 새 자료를 이미 막고 있었다(`is_analysis_running`). 없던 것은
|
|
**막힌 까닭을 화면에 보이는 것**이다 — 여럿이 한 프로젝트를 보면 「왜 안 올라가지」가 된다.
|
|
|
|
⇒ 시작한 사람을 `start_stage` 의 `params.started_by` 에 담고, `analysis_lock_owner` 가
|
|
그 id 로 이름을 붙여 낸다. 옛 자료는 id 가 없어 **이름 없이 잠금만** 돈다(막는 것이 먼저).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from common_util import common_util_workflow_state as state_mod # noqa: E402
|
|
|
|
|
|
class _Cursor:
|
|
"""질의 한 벌만 흉내 낸다 — 단계 행 하나와 사용자 한 명."""
|
|
|
|
def __init__(self, stage: dict[str, Any] | None, user: dict[str, Any] | None) -> None:
|
|
self.stage = stage
|
|
self.user = user
|
|
self._last = ""
|
|
|
|
async def execute(self, sql: str, args: tuple[Any, ...] = ()) -> None:
|
|
self._last = sql
|
|
|
|
async def fetchone(self) -> dict[str, Any] | None:
|
|
return self.user if "FROM users" in self._last else self.stage
|
|
|
|
|
|
@pytest.fixture
|
|
def _running(monkeypatch: pytest.MonkeyPatch):
|
|
async def _yes(cursor: Any, project_id: str) -> bool:
|
|
return True
|
|
|
|
monkeypatch.setattr(state_mod, "is_analysis_running", _yes)
|
|
|
|
|
|
def test_도는_중이_아니면_잠금이_없다(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
async def _no(cursor: Any, project_id: str) -> bool:
|
|
return False
|
|
|
|
monkeypatch.setattr(state_mod, "is_analysis_running", _no)
|
|
assert asyncio.run(state_mod.analysis_lock_owner(_Cursor(None, None), "p1")) is None
|
|
|
|
|
|
def test_시작한_사람의_이름이_붙는다(_running) -> None:
|
|
stage = {"params": json.dumps({"started_by": 7}), "started_at": None}
|
|
cursor = _Cursor(stage, {"name": "엄상돈", "email": "a@b.c"})
|
|
lock = asyncio.run(state_mod.analysis_lock_owner(cursor, "p1"))
|
|
assert lock is not None
|
|
assert lock["running"] is True and lock["name"] == "엄상돈" and lock["user_id"] == 7
|
|
|
|
|
|
def test_옛_자료는_이름_없이_잠금만_돈다(_running) -> None:
|
|
"""`started_by` 가 없던 시절 자료 — 막는 것이 먼저다. 이름이 없다고 잠금을 풀지 않는다."""
|
|
cursor = _Cursor({"params": json.dumps({"force": False}), "started_at": None}, None)
|
|
lock = asyncio.run(state_mod.analysis_lock_owner(cursor, "p1"))
|
|
assert lock is not None and lock["running"] is True and lock["name"] is None
|
|
|
|
|
|
def test_올리는_경로_넷이_모두_시작한_사람을_싣는다() -> None:
|
|
"""한 경로만 빠져도 그 경로로 올린 잠금은 이름이 안 뜬다 — 옛 사고와 같은 결."""
|
|
sources = [
|
|
ROOT / "B03_FileInput" / name
|
|
for name in (
|
|
"B03_FileInput_Router.py",
|
|
"B03_FileInput_Router_Chunks.py",
|
|
"B03_FileInput_Router_Temp.py",
|
|
)
|
|
]
|
|
calls = sum(
|
|
source.read_text(encoding="utf-8").count("trigger_wf1_analysis_and_email(")
|
|
for source in sources
|
|
)
|
|
carried = sum(
|
|
source.read_text(encoding="utf-8").count("started_by=session.get") for source in sources
|
|
)
|
|
assert calls == carried, f"분석을 띄우는 자리 {calls} 곳 중 {carried} 곳만 시작한 사람을 싣는다"
|