Aislo/resources/tester/test_309_2_calc_sites_browser.py

"""PLAN 504-2 화면 — 자기 포트(5177) Playwright 머리 없는 브라우저 · 서버가 떠 있을 때만.

- 시스템 관리자 = 「계산 자리」 카드 · 창(밝음 · 어두움 한 장씩 · tmp/shots_504_2)
- 회사 관리자 · 일반 = 카드 없음 · PUT 403
- G1 체크 끔 → [저장] → B05 열기(검증_sub2) → 지반 표본이 서버 길(`/route/elevations`) · 판 기본값으로 되돌림
- 시험 브라우저 = 표식 문자열(`MARK`) · 끝나면 그 표식 시스템 로그 행만 지움(508-1a)
- 포트 = OWNERS.md 이 폴더 줄 — 다른 폴더 사본이 남의 서버 판을 안 고침

  ./venv/Scripts/python.exe -m pytest resources/tester/test_309_2_calc_sites_browser.py -q
"""

from __future__ import annotations

import asyncio
import json
import urllib.request
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]

from helper_verify_project import own_ports  # noqa: E402

API, BASE = own_ports()  # 이 폴더 백엔드 · 그 백엔드를 프록시하는 vite(OWNERS.md 이 폴더 줄)
BOARD = ROOT / "resources" / "master_template" / "calc_sites.json"
PROJECT = "cb15892a-5788-4a26-9316-20a15421f68b"  # 검증_sub2(OWNERS.md)
SHOTS = ROOT / "tmp" / "shots_504_2"
CARD = "계산 자리"
MARK = "AisloTest-309-2"  # 시험 브라우저 문자열 — 시스템 로그 user_agent 칸으로 시험 행을 가림


def _started(base: str) -> str | None:
    try:
        with urllib.request.urlopen(f"{base}/api/health", timeout=3) as res:
            return json.loads(res.read()).get("started_at")
    except Exception:
        return None


_UP = _started(BASE) if BASE else None
# vite 가 이 폴더 백엔드가 아닌 서버(예 8000 = 메인 폴더)로 가면 PUT 이 남의 판을 고침 → 돌리지 않음
pytestmark = pytest.mark.skipif(
    _UP is None or _UP != _started(API), reason="이 폴더 서버 없음 · vite 가 이 폴더 백엔드로 안 감"
)


def _cred(who: str) -> dict:
    if who == "시스템 관리자":
        return json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
    accounts = json.loads((ROOT / "tmp" / "verify_accounts.json").read_text(encoding="utf-8"))
    return {k: accounts[who][k] for k in ("email", "password")}


def _drop_test_logs() -> None:
    """시험이 남긴 시스템 로그(로그인 · CALC_SITE_CHANGE) = 표식 브라우저 문자열 행만 지움."""
    from config.config_db import close_db_pool, init_db_pool

    async def run():
        pool = await init_db_pool()
        try:
            async with pool.acquire() as connection, connection.cursor() as cursor:
                await cursor.execute("DELETE FROM system_audit_logs WHERE user_agent = %s", (MARK,))
                await connection.commit()
        finally:
            await close_db_pool()

    asyncio.run(run())


@pytest.fixture(scope="module")
def browser():
    sync = pytest.importorskip("playwright.sync_api")
    committed = BOARD.read_bytes()  # 시작 때 파일 바이트 그대로(줄끝 포함)
    try:
        with sync.sync_playwright() as play:
            chrome = play.chromium.launch(channel="msedge", headless=True)
            try:
                yield chrome
            finally:
                chrome.close()
    finally:  # 실패해도 판 = 커밋한 기본값 그대로 · 로그 지움은 Playwright 고리 밖에서
        BOARD.write_bytes(committed)
        _drop_test_logs()


def _context(browser, who: str):
    ctx = browser.new_context(viewport={"width": 1400, "height": 900}, user_agent=MARK)
    r = ctx.request.post(f"{BASE}/api/auth/login/request", data=_cred(who), timeout=60000)
    assert r.ok, f"로그인 실패 {r.status}"  # 비밀번호는 안 찍음
    return ctx


def _dashboard(ctx):
    page = ctx.new_page()
    page.goto(f"{BASE}/#/b01-account", wait_until="domcontentloaded")
    page.wait_for_selector(".b01-dashboard__setting-card", timeout=60000)
    return page


def _cards(page) -> list[str]:
    return page.eval_on_selector_all(
        ".b01-dashboard__setting-title", "(els) => els.map((e) => e.textContent.trim())"
    )


@pytest.mark.parametrize("who", ["회사관리자", "일반"])
def test_다른_역할_카드_없음_403(browser, who):
    ctx = _context(browser, who)
    try:
        assert CARD not in _cards(_dashboard(ctx))
        put = ctx.request.put(
            f"{BASE}/api/dashboard/admin/calc-sites",
            data={"sites": {"ground_sample": "server"}},
        )
        assert put.status == 403
        assert ctx.request.get(f"{BASE}/api/dashboard/calc-sites").ok  # 읽기 = 로그인
    finally:
        ctx.close()


def _elevation_calls(ctx) -> int:
    page = ctx.new_page()
    # 대시보드가 지금 프로젝트 칸을 비움 — 페이지 스크립트보다 먼저 넣음
    page.add_init_script(f"localStorage.setItem('frd_current_project_id', '{PROJECT}')")
    calls: list[str] = []
    page.on("request", lambda r: "/route/elevations" in r.url and calls.append(r.url))
    page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
    page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
    page.wait_for_timeout(8000)
    page.close()
    return len(calls)


def test_시스템_관리자_카드_창_G1_서버_길(browser):
    ctx = _context(browser, "시스템 관리자")
    try:
        before = _elevation_calls(ctx)  # 기본값 = 격자(브라우저)
        page = _dashboard(ctx)
        assert CARD in _cards(page)
        page.get_by_text(CARD, exact=True).click()
        dialog = page.wait_for_selector(".b01-dashboard__calc-sites", timeout=30000)
        rows = page.eval_on_selector_all(
            ".b01-dashboard__calc-sites tbody tr",
            "(trs) => trs.map((tr) => ({ head: tr.classList.contains('b01-dashboard__calc-group'),"
            " name: tr.cells[0].textContent.trim(), why: tr.cells[3].textContent.trim(),"
            " boxes: [...tr.querySelectorAll('input')].map((i) => [i.checked, i.disabled]) }))",
        )
        heads = [row for row in rows if row["head"]]
        calcs = [row for row in rows if not row["head"]]
        assert len(heads) == 13 and len(calcs) == 45  # 511-B 단계 다섯 줄 · F9 · 511-C haul
        whys = {row["name"].split(" · ")[0]: row["why"] for row in calcs}
        assert not any(why.startswith("짝 모으는 중") for why in whys.values())  # 단계 줄 모두 엶
        assert whys["C1"].startswith("끄면 횡단을 서버 Node")  # 511-B B-4 엶
        assert heads[0]["boxes"] == [[True, False], [True, True]]  # G1 묶음 머리 = 바꿀 수 있음
        editable = [row["name"] for row in calcs if not row["boxes"][0][1]]
        opened = ["G1", "D3", "E1", "F7", "F8", "C4", "R9", "F10"]  # 511-A ③ ④ 더
        opened += ["R1", "R3", "P5", "P6", "F1", "F2"]  # 511-B B-5 노선 · 따라감
        opened += ["R5", "R6", "R7", "F3"]  # 511-B B-2 측점
        opened += ["P1", "P2", "P3", "P4", "P8"]  # 511-B B-3 종단 엶
        opened += ["G2", "G4", "C1", "C2", "C3", "C5", "C6", "F4", "F5", "F6"]  # 511-B B-4
        opened += ["E2"]  # 511-C 유토 배분 엶(기본 server = 체크 꺼짐)
        opened += ["R2", "R4", "R8", "P7", "C7", "D4"]  # ⑧ 시험용
        assert [name.split(" · ")[0] for name in editable] == opened
        assert all(row["name"].split(" · ")[1] for row in calcs)  # 이름 = 등록부 label
        assert all(row["boxes"][1][1] for row in calcs)  # 서버 정본 칸은 늘 잠금
        assert "지금 서버 1 줄" in dialog.inner_text()  # haul 기본 server(511-C)
        SHOTS.mkdir(parents=True, exist_ok=True)
        dialog.screenshot(path=str(SHOTS / "light.png"))
        page.evaluate("document.documentElement.setAttribute('data-theme', 'dark')")
        dialog.screenshot(path=str(SHOTS / "dark.png"))
        page.evaluate("document.documentElement.setAttribute('data-theme', 'light')")

        page.click("input[data-calc-site='ground_sample']")
        assert "지금 서버 2 줄" in dialog.inner_text()
        assert not page.is_checked("input[data-calc-no='G1']")  # 머리 따라 줄도 꺼짐
        page.locator(".b01-dashboard__calc-sites .ui-modal__actions button").click()
        page.wait_for_selector(".b01-dashboard__calc-sites", state="detached", timeout=30000)
        got = ctx.request.get(f"{BASE}/api/dashboard/calc-sites").json()["sites"]
        assert got["ground_sample"] == "server", (got, BOARD.read_text("utf-8"))
        assert json.loads(BOARD.read_text("utf-8"))["sites"]["ground_sample"] == "server"
        logs = ctx.request.get(f"{BASE}/api/dashboard/admin/audit-logs").json()["items"]
        assert any(
            log.get("action") == "CALC_SITE_CHANGE"
            and log.get("resource_ref") == "ground_sample=server"
            for log in logs
        )

        after = _elevation_calls(ctx)
        print(f"elevations 요청: 기본값 {before} · 서버 {after}")
        assert after > before

        back = ctx.request.put(
            f"{BASE}/api/dashboard/admin/calc-sites", data={"sites": {"ground_sample": "both"}}
        )
        assert back.ok and back.json()["sites"]["ground_sample"] == "both"
    finally:
        ctx.close()