355 lines
14 KiB
Python
355 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
import unittest
|
|
from contextlib import contextmanager
|
|
|
|
from playwright.async_api import async_playwright
|
|
from werkzeug.serving import make_server
|
|
|
|
from topology_api.app import create_app
|
|
|
|
|
|
@contextmanager
|
|
def topology_server():
|
|
app = create_app()
|
|
app.config.update(TESTING=True)
|
|
server = make_server("127.0.0.1", 0, app, threaded=True)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
port = server.socket.getsockname()[1]
|
|
yield f"http://127.0.0.1:{port}"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=2)
|
|
|
|
|
|
class GuanghanTopologyE2ETests(unittest.TestCase):
|
|
def test_historical_graph_and_ghc05_rendering(self) -> None:
|
|
async def scenario(base_url: str) -> None:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(headless=True)
|
|
page = await browser.new_page(
|
|
viewport={"width": 1440, "height": 1000}
|
|
)
|
|
console_errors: list[str] = []
|
|
page.on(
|
|
"console",
|
|
lambda msg: (
|
|
console_errors.append(msg.text)
|
|
if msg.type == "error"
|
|
else None
|
|
),
|
|
)
|
|
|
|
async def open_mission(code: str) -> None:
|
|
await page.goto(
|
|
f"{base_url}/guanghan_topology_demo.html"
|
|
f"?mission={code}",
|
|
wait_until="networkidle",
|
|
)
|
|
await page.wait_for_function(
|
|
"""code => (
|
|
document.querySelector('#page-subtitle')
|
|
?.textContent.startsWith(code + ' ·')
|
|
)""",
|
|
arg=code,
|
|
)
|
|
|
|
await open_mission("BL-01")
|
|
self.assertEqual(
|
|
await page.locator(".topology-node").count(), 1
|
|
)
|
|
self.assertEqual(
|
|
await page.locator(".connection").count(), 0
|
|
)
|
|
self.assertEqual(await page.locator(".radiator").count(), 0)
|
|
|
|
await open_mission("GHC-01")
|
|
self.assertEqual(
|
|
await page.locator(".topology-node").count(), 2
|
|
)
|
|
self.assertEqual(
|
|
await page.locator(".connection").count(), 0
|
|
)
|
|
|
|
await open_mission("GHC-05")
|
|
self.assertEqual(
|
|
await page.locator(".topology-node").count(), 14
|
|
)
|
|
self.assertEqual(
|
|
await page.locator('.connection[data-type="pressurized_passage"]').count(), 2
|
|
)
|
|
self.assertEqual(await page.locator(".radiator").count(), 4)
|
|
strokes = await page.locator(
|
|
'.connection[data-type="pressurized_passage"]'
|
|
).evaluate_all(
|
|
"(els) => els.map((el) => getComputedStyle(el).stroke)"
|
|
)
|
|
self.assertEqual(
|
|
strokes,
|
|
["rgb(81, 232, 156)", "rgb(81, 232, 156)"],
|
|
)
|
|
|
|
overflows = await page.locator(
|
|
".topology-node"
|
|
).evaluate_all(
|
|
"""nodes => nodes.flatMap(node => {
|
|
const rect = node.querySelector('rect').getBBox();
|
|
return [...node.querySelectorAll('text')]
|
|
.map(text => ({text: text.textContent,
|
|
box: text.getBBox()}))
|
|
.filter(item =>
|
|
item.box.x < rect.x ||
|
|
item.box.x + item.box.width > rect.x + rect.width
|
|
);
|
|
})"""
|
|
)
|
|
self.assertEqual(overflows, [])
|
|
self.assertEqual(console_errors, [])
|
|
await browser.close()
|
|
|
|
with topology_server() as base_url:
|
|
asyncio.run(scenario(base_url))
|
|
|
|
def test_controls_navigation_and_modal_semantics(self) -> None:
|
|
async def scenario(base_url: str) -> None:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(headless=True)
|
|
page = await browser.new_page(
|
|
viewport={"width": 1440, "height": 1000}
|
|
)
|
|
await page.goto(
|
|
f"{base_url}/guanghan_topology_demo.html"
|
|
"?mission=GHC-05",
|
|
wait_until="networkidle",
|
|
)
|
|
await page.wait_for_function(
|
|
"""() => document.querySelectorAll(
|
|
'.topology-node'
|
|
).length === 14"""
|
|
)
|
|
|
|
options = await page.locator(
|
|
"#mission-select option"
|
|
).evaluate_all("(items) => items.map(item => item.value)")
|
|
self.assertEqual(
|
|
options,
|
|
[
|
|
"",
|
|
"BL-01",
|
|
"GHC-01",
|
|
"JC-01",
|
|
"BL-02",
|
|
"GHC-02",
|
|
"GHC-03",
|
|
"BL-03",
|
|
"GHC-04",
|
|
"GHC-05",
|
|
],
|
|
)
|
|
|
|
core = page.locator(
|
|
'.topology-node[data-node-key="core_cabin"]'
|
|
)
|
|
await core.click()
|
|
self.assertIn(
|
|
"GHC-01",
|
|
await page.locator("#detail-mission").inner_text(),
|
|
)
|
|
await page.locator(".modal-close").click()
|
|
self.assertEqual(
|
|
await page.evaluate(
|
|
"() => document.activeElement?.dataset.nodeKey"
|
|
),
|
|
"core_cabin",
|
|
)
|
|
|
|
await page.select_option("#mission-select", "GHC-01")
|
|
await page.fill(
|
|
"#time-input",
|
|
"2025-01-02T08:00:00+08:00",
|
|
)
|
|
await page.locator("#go-btn").click()
|
|
await page.wait_for_function(
|
|
"""() => document.querySelector('#snapshot-info')
|
|
?.textContent.includes('明确时间')"""
|
|
)
|
|
self.assertTrue(
|
|
(await page.locator("#page-subtitle").inner_text())
|
|
.startswith("GHC-01 ·")
|
|
)
|
|
self.assertIn("mission=GHC-01", page.url)
|
|
self.assertIn("at=", page.url)
|
|
|
|
await page.fill("#time-input", "not-a-time")
|
|
await page.locator("#go-btn").click()
|
|
self.assertIn(
|
|
"请输入带时区的 ISO 8601 时间",
|
|
await page.locator("#error-msg").inner_text(),
|
|
)
|
|
|
|
await page.fill("#time-input", "")
|
|
await page.select_option("#mission-select", "GHC-01")
|
|
await page.locator("#go-btn").click()
|
|
await page.wait_for_function(
|
|
"""() => document.querySelector('#page-subtitle')
|
|
?.textContent.startsWith('GHC-01 ·')"""
|
|
)
|
|
await page.select_option("#mission-select", "GHC-05")
|
|
await page.locator("#go-btn").click()
|
|
await page.wait_for_function(
|
|
"""() => document.querySelector('#page-subtitle')
|
|
?.textContent.startsWith('GHC-05 ·')"""
|
|
)
|
|
await page.go_back()
|
|
await page.wait_for_function(
|
|
"""() => document.querySelector('#page-subtitle')
|
|
?.textContent.startsWith('GHC-01 ·')"""
|
|
)
|
|
self.assertEqual(
|
|
await page.locator("#mission-select").input_value(),
|
|
"GHC-01",
|
|
)
|
|
await browser.close()
|
|
|
|
with topology_server() as base_url:
|
|
asyncio.run(scenario(base_url))
|
|
|
|
|
|
def test_database_edge_style_changes_browser_rendering(self) -> None:
|
|
async def scenario(base_url: str) -> None:
|
|
from topology_api.db import get_writer_connection
|
|
|
|
with get_writer_connection() as conn:
|
|
conn.execute("SET search_path TO topology, public")
|
|
with conn.cursor() as cur:
|
|
cur.execute("""
|
|
SELECT stroke_color
|
|
FROM diagram_edge_styles
|
|
WHERE connection_type = 'pressurized_passage'
|
|
""")
|
|
original_color = cur.fetchone()["stroke_color"]
|
|
cur.execute("""
|
|
UPDATE diagram_edge_styles
|
|
SET stroke_color = '#00ffaa'
|
|
WHERE connection_type = 'pressurized_passage'
|
|
""")
|
|
conn.commit()
|
|
try:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(headless=True)
|
|
page = await browser.new_page(
|
|
viewport={"width": 1440, "height": 1000}
|
|
)
|
|
await page.goto(
|
|
f"{base_url}/guanghan_topology_demo.html"
|
|
"?mission=GHC-05",
|
|
wait_until="networkidle",
|
|
)
|
|
await page.wait_for_function(
|
|
"""() => document.querySelectorAll(
|
|
'.connection[data-type="pressurized_passage"]'
|
|
).length === 2"""
|
|
)
|
|
strokes = await page.locator(
|
|
'.connection[data-type="pressurized_passage"]'
|
|
).evaluate_all(
|
|
"(els) => els.map((el) => getComputedStyle(el).stroke)"
|
|
)
|
|
self.assertEqual(
|
|
strokes,
|
|
["rgb(0, 255, 170)", "rgb(0, 255, 170)"],
|
|
)
|
|
await browser.close()
|
|
finally:
|
|
with conn.cursor() as cur:
|
|
cur.execute("""
|
|
UPDATE diagram_edge_styles
|
|
SET stroke_color = %(original)s
|
|
WHERE connection_type = 'pressurized_passage'
|
|
""", {"original": original_color})
|
|
conn.commit()
|
|
|
|
with topology_server() as base_url:
|
|
asyncio.run(scenario(base_url))
|
|
|
|
def test_database_node_style_changes_browser_rendering(self) -> None:
|
|
async def scenario(base_url: str) -> None:
|
|
from topology_api.db import get_writer_connection
|
|
|
|
with get_writer_connection() as conn:
|
|
conn.execute("SET search_path TO topology, public")
|
|
with conn.cursor() as cur:
|
|
cur.execute("""
|
|
SELECT fill_color
|
|
FROM diagram_node_styles
|
|
WHERE style_key = 'core-card'
|
|
""")
|
|
original_fill = cur.fetchone()["fill_color"]
|
|
cur.execute("""
|
|
UPDATE diagram_node_styles
|
|
SET fill_color = '#123456'
|
|
WHERE style_key = 'core-card'
|
|
""")
|
|
conn.commit()
|
|
try:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(headless=True)
|
|
page = await browser.new_page(
|
|
viewport={"width": 1440, "height": 1000}
|
|
)
|
|
await page.goto(
|
|
f"{base_url}/guanghan_topology_demo.html"
|
|
"?mission=GHC-05",
|
|
wait_until="networkidle",
|
|
)
|
|
core_fill = await page.locator(
|
|
'.topology-node[data-node-key="core_cabin"] rect'
|
|
).evaluate(
|
|
"(el) => getComputedStyle(el).fill"
|
|
)
|
|
self.assertEqual(core_fill, "rgb(18, 52, 86)")
|
|
await browser.close()
|
|
finally:
|
|
with conn.cursor() as cur:
|
|
cur.execute("""
|
|
UPDATE diagram_node_styles
|
|
SET fill_color = %(original)s
|
|
WHERE style_key = 'core-card'
|
|
""", {"original": original_fill})
|
|
conn.commit()
|
|
|
|
with topology_server() as base_url:
|
|
asyncio.run(scenario(base_url))
|
|
|
|
def test_legend_renders_from_database_presentation(self) -> None:
|
|
async def scenario(base_url: str) -> None:
|
|
async with async_playwright() as pw:
|
|
browser = await pw.chromium.launch(headless=True)
|
|
page = await browser.new_page(
|
|
viewport={"width": 1440, "height": 1000}
|
|
)
|
|
await page.goto(
|
|
f"{base_url}/guanghan_topology_demo.html?mission=GHC-05",
|
|
wait_until="networkidle",
|
|
)
|
|
labels = await page.locator("#legend .legend-item").evaluate_all(
|
|
"(items) => items.map(item => item.textContent.trim())"
|
|
)
|
|
self.assertEqual(
|
|
labels,
|
|
["结构直连", "加压走道", "燃料 / 供电", "大型折叠散热面板"],
|
|
)
|
|
await browser.close()
|
|
|
|
with topology_server() as base_url:
|
|
asyncio.run(scenario(base_url))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|