"""업로드 잠금 표시 — **지금 누가 올리는 중인지** (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} 곳만 시작한 사람을 싣는다"