390 lines
17 KiB
Python
390 lines
17 KiB
Python
"""PLAN 508-2 화면 — 자기 포트 Playwright 머리 없는 브라우저 · 서버가 떠 있을 때만 · 검증_sub4.
|
|
|
|
① 판 D3 = 서버 · 저장 3D 모형 열쇠를 낡게(머리 hash 만 바꾼 사본 · 끝나면 바이트 되돌림)
|
|
→ 들어오면 옛 모형이 섬 · [3D 업데이트] 「늦음」 · 3D 창을 열어도 안 지음
|
|
· [3D 업데이트] = /calc corridor.mesh 한 번 → 새 모형 · 늦음 꺼짐 · 한 번 더 = /calc 0(같은 입력)
|
|
② 추천 노선 창 — 저장본 있는 갈래 = [재계산] 숨김 · 저장본 없음(409) = 안내 줄 옆 [재계산]
|
|
· 제약 칸 바꿈 = 종단 반영 오른쪽 [재계산] · [재계산] · [적용]은 안 누름
|
|
③ 노선 편집 — 교각점을 범위(저장 노선 띠 300 + 여유 300 m) 밖으로 끌어 놓으면 잠잠 뒤 bbox 로
|
|
잘린 도엽 다시 받음 · 받는 양 · 밝음 · 어두움 한 장씩(`tmp/5082/`) · 측점 스냅 끔은 세션만
|
|
- 저장 안 함(세션 초안만) · 판 = 파일 바이트 되돌림 · 시험 브라우저 표식 시스템 로그 행만 지움
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_508_2_browser.py -q -s
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
ROOT = HERE.parents[1]
|
|
if str(HERE) not in sys.path:
|
|
sys.path.insert(0, str(HERE))
|
|
|
|
from helper_verify_project import own_ports # noqa: E402
|
|
from test_505_3_cross_card_screen import JS_NODES # noqa: E402
|
|
from test_511_a_calc_sites_browser import _UP, _started # noqa: E402
|
|
|
|
API, BASE = own_ports()
|
|
BOARD = ROOT / "resources" / "master_template" / "calc_sites.json"
|
|
PROJECT = "c725d747-50ae-46b0-87ce-4d4e56a53cf3" # 검증_sub4(OWNERS.md)
|
|
MARK = "AisloTest-508-2"
|
|
SHOTS = ROOT / "tmp" / "5082"
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
_UP is None or _UP != _started(API), reason="이 폴더 서버 없음 · vite 가 이 폴더 백엔드로 안 감"
|
|
)
|
|
|
|
JS_CORRIDOR = r"""() => {
|
|
const s = window.__corridorScene;
|
|
const g = s && s.scene.getObjectByName('corridor');
|
|
if (!g) return null;
|
|
g.userData.mark ??= Math.random().toString(36).slice(2);
|
|
return { mark: g.userData.mark, children: g.children.length };
|
|
}"""
|
|
JS_PENDING = r"""() => {
|
|
const b = [...document.querySelectorAll('button')]
|
|
.find((x) => /3D 업데이트|짓는 중|갱신 중/.test(x.textContent));
|
|
return b ? { text: b.textContent.trim(), pending: b.classList.contains('is-pending') } : null;
|
|
}"""
|
|
JS_BUILT = (
|
|
"() => ![...document.querySelectorAll('button')].some((b) => /짓는 중/.test(b.textContent))"
|
|
)
|
|
JS_IDLE = (
|
|
"() => !/셈하는 중/.test(document.querySelector('.b05-routerec [data-state]').textContent)"
|
|
)
|
|
JS_SHOWN = (
|
|
"() => /미리 보기/.test(document.querySelector('.b05-routerec [data-state]').textContent)"
|
|
)
|
|
JS_REC = r"""() => {
|
|
const m = document.querySelector('.b05-routerec');
|
|
const r = m?.querySelector('[data-act="rerun"]');
|
|
return m && {
|
|
state: m.querySelector('[data-state]').textContent,
|
|
shown: !!r && !r.hidden,
|
|
beside: !!r && r.previousElementSibling?.matches('[data-state]'),
|
|
inRow: !!r && !!r.closest('.b05-routerec__row'),
|
|
};
|
|
}"""
|
|
|
|
|
|
def _project_root() -> Path:
|
|
import asyncio
|
|
|
|
from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
|
|
from common_util.common_util_storage import resolve_stored_project_path
|
|
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:
|
|
return await get_project_storage_relative_path(connection, PROJECT)
|
|
finally:
|
|
await close_db_pool()
|
|
|
|
return Path(resolve_stored_project_path(asyncio.run(run())))
|
|
|
|
|
|
# 플레이라이트 고리 밖에서 한 번(그 안에서는 asyncio.run 못 씀)
|
|
PROJECT_ROOT = _project_root() if _UP and _UP == _started(API) else None
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def ctx():
|
|
sync = pytest.importorskip("playwright.sync_api")
|
|
committed = BOARD.read_bytes()
|
|
cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
|
|
SHOTS.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
with sync.sync_playwright() as play:
|
|
chrome = play.chromium.launch(channel="msedge", headless=True)
|
|
context = chrome.new_context(viewport={"width": 1600, "height": 1000}, user_agent=MARK)
|
|
login = {k: cred[k] for k in ("email", "password")}
|
|
r = context.request.post(f"{BASE}/api/auth/login/request", data=login, timeout=60000)
|
|
assert r.ok, f"로그인 실패 {r.status}" # 비밀번호는 안 찍음
|
|
try:
|
|
yield context
|
|
finally:
|
|
chrome.close()
|
|
finally:
|
|
BOARD.write_bytes(committed)
|
|
_drop_logs()
|
|
|
|
|
|
def _drop_logs() -> None:
|
|
import asyncio
|
|
|
|
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())
|
|
|
|
|
|
# 계정 화면 취향(서버 user_ui_prefs) — 새 문맥이어도 남음 · 끌기 시험이 건드리는 것
|
|
VIEW_PREFS = (f"aislo:pref:routeedit-view:{PROJECT}", "aislo:pref:profile-collapsed")
|
|
|
|
|
|
def _prefs(email: str, values: dict[str, str | None] | None = None) -> dict[str, str | None]:
|
|
"""`VIEW_PREFS` 값 읽기 · `values` 있으면 그 키만 되씀(None = 지움 · 다른 키 그대로)."""
|
|
import asyncio
|
|
import concurrent.futures
|
|
|
|
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("SELECT id FROM users WHERE email = %s", (email,))
|
|
user = (await cursor.fetchone())[0]
|
|
for key, value in (values or {}).items():
|
|
path = f'$."{key}"'
|
|
if value is None:
|
|
sql, args = "JSON_REMOVE(prefs, %s)", (path,)
|
|
else:
|
|
sql, args = "JSON_SET(prefs, %s, %s)", (path, value)
|
|
await cursor.execute(
|
|
f"UPDATE user_ui_prefs SET prefs = {sql} WHERE user_id = %s", (*args, user)
|
|
)
|
|
await connection.commit()
|
|
await cursor.execute("SELECT prefs FROM user_ui_prefs WHERE user_id = %s", (user,))
|
|
row = await cursor.fetchone()
|
|
prefs = json.loads(row[0]) if row else {}
|
|
return {key: prefs.get(key) for key in VIEW_PREFS}
|
|
finally:
|
|
await close_db_pool()
|
|
|
|
# 플레이라이트 고리 안 — asyncio.run 은 딴 실에서
|
|
with concurrent.futures.ThreadPoolExecutor(1) as pool:
|
|
return pool.submit(asyncio.run, run()).result()
|
|
|
|
|
|
def _board(corridor: str | None) -> None:
|
|
board = json.loads(BOARD.read_text("utf-8"))
|
|
board["sites"]["corridor_mesh"] = corridor or "both"
|
|
BOARD.write_text(json.dumps(board, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def _open(context, on_response=None):
|
|
page = context.new_page()
|
|
page.add_init_script(f"localStorage.setItem('frd_current_project_id', '{PROJECT}');")
|
|
if on_response:
|
|
page.on("response", on_response)
|
|
page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
|
|
page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
|
|
page.wait_for_timeout(9000)
|
|
return page
|
|
|
|
|
|
def test_3D_서버_낡음_옛_모형_단추로_바꿔_끼움(ctx):
|
|
root = PROJECT_ROOT
|
|
route_id = max(
|
|
int(p.stem.split("_")[1])
|
|
for p in (root / "B05_Profile" / "corridor").glob("corridor_*.json")
|
|
)
|
|
stored = root / "B05_Profile" / "corridor" / f"corridor_{route_id:04d}.json"
|
|
original = stored.read_bytes()
|
|
head = original[:4096]
|
|
found = re.search(rb'"hash"\s*:\s*"([0-9a-fA-F]+)"', head)
|
|
assert found
|
|
stale = original.replace(found.group(0), b'"hash":"0badbeef"', 1)
|
|
calls: list[str] = []
|
|
|
|
def on_response(res):
|
|
if res.url.endswith("/calc") and res.request.method == "POST":
|
|
body = json.loads(res.request.post_data)
|
|
calls.append("+".join(body["run"]))
|
|
|
|
_board("server")
|
|
stored.write_bytes(stale)
|
|
try:
|
|
page = _open(ctx, on_response)
|
|
page.get_by_role("button", name="3D", exact=True).first.click()
|
|
page.wait_for_timeout(6000)
|
|
entry = {"model": page.evaluate(JS_CORRIDOR), "button": page.evaluate(JS_PENDING)}
|
|
opened_calls = list(calls)
|
|
page.screenshot(path=str(SHOTS / "d3_stale_entry.png"))
|
|
button = page.locator("button", has_text="3D 업데이트").first
|
|
button.click()
|
|
page.wait_for_timeout(300)
|
|
during = page.evaluate(JS_PENDING)
|
|
page.wait_for_function(JS_BUILT, timeout=120000)
|
|
page.wait_for_timeout(1000)
|
|
after = {
|
|
"model": page.evaluate(JS_CORRIDOR),
|
|
"button": page.evaluate(JS_PENDING),
|
|
"source": page.evaluate("() => window.__corridorSource?.source ?? null"),
|
|
}
|
|
page.screenshot(path=str(SHOTS / "d3_rebuilt.png"))
|
|
built_calls = list(calls)
|
|
page.locator("button", has_text="3D 업데이트").first.click()
|
|
page.wait_for_timeout(4000)
|
|
again_calls = list(calls)
|
|
page.close()
|
|
finally:
|
|
stored.write_bytes(original)
|
|
_board(None)
|
|
print("entry", entry, "during", during, "after", after)
|
|
print("calls", opened_calls, built_calls, again_calls)
|
|
assert entry["model"] and entry["model"]["children"] > 0 # 옛 모형(빈 3D 아님)
|
|
assert entry["button"]["pending"] is True
|
|
assert "corridor.mesh" not in opened_calls # 창 열 때 안 지음
|
|
assert during["text"] == "서버에서 짓는 중…"
|
|
assert built_calls.count("corridor.mesh") == 1
|
|
assert after["model"]["mark"] != entry["model"]["mark"] # 바꿔 끼움
|
|
assert after["button"]["pending"] is False and after["source"] == "built"
|
|
assert again_calls.count("corridor.mesh") == 1 # 같은 입력 = 다시 안 지음
|
|
assert stored.read_bytes() == original
|
|
|
|
|
|
def test_추천_창_재계산_409_때만(ctx):
|
|
page = _open(ctx)
|
|
page.get_by_text("노선 추천", exact=True).first.click()
|
|
page.wait_for_timeout(800)
|
|
page.locator("button", has_text="노선 추천").last.click()
|
|
page.wait_for_selector(".b05-routerec", timeout=30000)
|
|
seen = []
|
|
for logic in ("follow", "gradeline", "earthwork", "gentle", "weighted"):
|
|
page.click(f'.b05-routerec [data-logic="{logic}"]')
|
|
page.wait_for_function(JS_IDLE, timeout=120000)
|
|
seen.append({"logic": logic, **page.evaluate(JS_REC)})
|
|
terrain = page.locator('.b05-routerec [data-act="terrain"]')
|
|
if terrain.is_visible():
|
|
other = "sheet" if terrain.input_value() == "las" else "las"
|
|
terrain.select_option(other)
|
|
page.wait_for_function(JS_IDLE, timeout=120000)
|
|
seen.append({"logic": f"terrain:{other}", **page.evaluate(JS_REC)})
|
|
print(json.dumps(seen, ensure_ascii=False, indent=1))
|
|
missing = [s for s in seen if "저장본이 없습니다" in s["state"]]
|
|
if missing:
|
|
page.screenshot(path=str(SHOTS / "rec_409_rerun.png"))
|
|
# 제약 칸 바꿈(저장본 있는 갈래로 돌아가서) — 종단 반영 오른쪽
|
|
stored = [s for s in seen if s["state"].startswith("미리 보기")]
|
|
if stored:
|
|
if terrain.is_visible():
|
|
terrain.select_option("las") # 지형 되돌림 — 마지막 갈래(비중) 라이다 저장본
|
|
else:
|
|
page.click(f'.b05-routerec [data-logic="{stored[0]["logic"]}"]')
|
|
page.wait_for_function(JS_SHOWN, timeout=120000)
|
|
field = page.locator(".b05-routerec__criteria input").first
|
|
field.fill("7")
|
|
page.wait_for_timeout(300)
|
|
dirty = page.evaluate(JS_REC)
|
|
page.screenshot(path=str(SHOTS / "rec_dirty_rerun.png"))
|
|
page.close()
|
|
print(json.dumps(seen, ensure_ascii=False, indent=1))
|
|
for s in seen:
|
|
if s["state"].startswith("미리 보기"):
|
|
assert s["shown"] is False, s
|
|
if "저장본이 없습니다" in s["state"]:
|
|
assert s["shown"] and s["beside"] and not s["inRow"], s
|
|
assert stored, "저장본 있는 갈래 없음"
|
|
assert dirty["shown"] and dirty["inRow"] and not dirty["beside"], dirty
|
|
if not missing:
|
|
pytest.skip("저장본 없는 갈래 없음 — 409 줄은 단위 시험 몫")
|
|
|
|
|
|
def test_노선_크게_끌어_놓기_등고선_다시_받음(ctx):
|
|
got: list[dict] = []
|
|
|
|
def on_response(res):
|
|
if "/geojson?" in res.url and "scope=route" in res.url:
|
|
got.append({"bbox": "bbox=" in res.url, "status": res.status, "bytes": len(res.body())})
|
|
|
|
# 새 문맥 — 앞 시험 페이지가 남긴 보던 화면 · 보관함 없이(축척 · 받기가 늘 같게)
|
|
fresh_ctx = ctx.browser.new_context(viewport={"width": 1600, "height": 1000}, user_agent=MARK)
|
|
cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
|
|
# 보던 화면은 계정(서버)에도 남음 — 지난 판 끌기 · 줄임이 쌓여 시작 화면이 매번 달랐음(514-4)
|
|
kept = _prefs(cred["email"])
|
|
_prefs(cred["email"], {VIEW_PREFS[0]: None})
|
|
try:
|
|
_drag_and_fetch(fresh_ctx, cred, got, on_response)
|
|
finally:
|
|
fresh_ctx.close()
|
|
_prefs(cred["email"], kept) # 계정 화면 취향 되돌림
|
|
|
|
|
|
def _drag_and_fetch(fresh_ctx, cred, got, on_response) -> None:
|
|
login = {k: cred[k] for k in ("email", "password")}
|
|
assert fresh_ctx.request.post(f"{BASE}/api/auth/login/request", data=login, timeout=60000).ok
|
|
page = _open(fresh_ctx, on_response)
|
|
handle = page.locator(".ui-workflow-overlay__handle--bottom", has_text="종단").first
|
|
for _ in range(6): # 종단 접음 — 평면 넓게(배치가 늦게 되살아나면 다시 접음)
|
|
if handle.bounding_box()["y"] > 900:
|
|
break
|
|
handle.click()
|
|
page.wait_for_timeout(2500)
|
|
page.mouse.move(960, 500)
|
|
for _ in range(3): # 줄여 봄 — 오른쪽 끝 교각점을 범위 밖까지 끌 자리
|
|
page.mouse.wheel(0, -300)
|
|
page.wait_for_timeout(300)
|
|
first = [g for g in got if not g["bbox"]] # 앞 시험이 담아 두었으면 보관함 — 0 일 수 있음
|
|
nodes = []
|
|
for _ in range(20): # 접은 뒤 다시 그릴 때까지(사진 한 장 = 그림 한 번 내보냄)
|
|
page.screenshot(path=str(SHOTS / "sheet_follow_before.png"))
|
|
nodes = page.evaluate(JS_NODES)
|
|
if nodes:
|
|
break
|
|
page.wait_for_timeout(500)
|
|
page.screenshot(path=str(SHOTS / "sheet_follow_before.png"))
|
|
assert nodes, "교각점 못 찾음"
|
|
page.get_by_role("button", name="측점 스냅", exact=True).click() # 끔(세션만) — 큰 끌기 막음 풂
|
|
page.wait_for_timeout(500)
|
|
during = None
|
|
drops = 0
|
|
for _ in range(4): # 끌어 놓기 · 화면을 왼쪽으로 옮김 — 범위 밖까지
|
|
node = max(nodes, key=lambda n: n["x"]) # 오른쪽 끝 교각점 — 오른쪽 바깥으로
|
|
page.mouse.move(node["x"], node["y"])
|
|
page.mouse.down()
|
|
page.mouse.move(node["x"] + 150, node["y"], steps=8)
|
|
page.wait_for_timeout(600)
|
|
if during is None:
|
|
during = sum(g["bbox"] for g in got) # 끄는 동안(잠잠 시간보다 오래 쥠)
|
|
page.mouse.move(1590, node["y"], steps=8)
|
|
page.mouse.up()
|
|
drops += 1
|
|
page.wait_for_timeout(4000)
|
|
if any(g["bbox"] for g in got):
|
|
# 둘째 도엽(수 MB)은 늦게 옴 — 바쁜 서버(싱크 때 시험 묶음)면 4 초 안에 하나만(514-4)
|
|
for _ in range(60):
|
|
if sum(g["bbox"] for g in got) >= 2:
|
|
break
|
|
page.wait_for_timeout(500)
|
|
page.wait_for_timeout(1500) # 더 오는 것(다시 받기 두 번)도 잡히게
|
|
break
|
|
page.mouse.move(1500, 900)
|
|
page.mouse.down(button="middle")
|
|
page.mouse.move(900, 900, steps=10)
|
|
page.mouse.up(button="middle")
|
|
page.wait_for_timeout(800)
|
|
page.screenshot(path=str(SHOTS / "sheet_follow_before.png"))
|
|
nodes = page.evaluate(JS_NODES)
|
|
assert nodes, "옮긴 뒤 교각점 못 찾음"
|
|
status = page.locator(".b05-routeedit__status").first.text_content()
|
|
after = list(got)
|
|
page.screenshot(path=str(SHOTS / "sheet_follow_light.png"))
|
|
page.evaluate("() => document.documentElement.setAttribute('data-theme', 'dark')")
|
|
page.wait_for_timeout(1500)
|
|
page.screenshot(path=str(SHOTS / "sheet_follow_dark.png"))
|
|
page.evaluate("() => document.documentElement.setAttribute('data-theme', 'light')")
|
|
page.close()
|
|
fresh = [g for g in after if g["bbox"]]
|
|
print("first", first, "fresh", fresh, "drops", drops, "status", status)
|
|
assert during == 0 # 끄는 동안 = 안 받음
|
|
assert len(fresh) == 2 and all(g["bbox"] and g["status"] == 200 for g in fresh)
|
|
assert all(g["bytes"] < 5_000_000 for g in fresh) # 통째 사본(36.8 MB) 아님
|