chore: add reference docs, scripts, tests, demo, prototypes

- Reference docs: known issues, user journey, location redesign, mission merge plan
- Scripts: fill_location_state, merge_missions, migrate_location, migrate_state
- Tests: e2e test suite, asset entries unit tests
- Demo: location board, hifi prototypes, test screenshots
- Data backup: asset log entries and state nodes
- Update .gitignore

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-06-01 15:29:47 +08:00
co-authored by Claude Opus 4.7
parent f4909765af
commit e1d601a774
190 changed files with 11464 additions and 0 deletions
+285
View File
@@ -0,0 +1,285 @@
from __future__ import annotations
import unittest
from datetime import datetime
from app import create_app
from app.config import Config
from app.extensions import db
from app.models import Asset, AssetLogEntry, DockingEvent
class TestConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
WTF_CSRF_ENABLED = False
class AssetEntryTests(unittest.TestCase):
def setUp(self) -> None:
self.app = create_app(TestConfig)
self.client = self.app.test_client()
with self.app.app_context():
db.create_all()
asset = Asset(
name="Codex Test Asset",
asset_type="Station",
program="Codex",
home_region="LEO",
)
db.session.add(asset)
db.session.commit()
self.asset_id = asset.id
def tearDown(self) -> None:
with self.app.app_context():
db.session.remove()
db.drop_all()
def _latest_entry(self) -> AssetLogEntry:
with self.app.app_context():
return (
db.session.query(AssetLogEntry)
.filter_by(asset_id=self.asset_id)
.order_by(AssetLogEntry.created_at.desc())
.first()
)
def test_form_state_nodes_save_all_location_and_state_fields(self) -> None:
response = self.client.post(
f"/assets/{self.asset_id}/entries/new",
data={
"entry_kind": "state",
"title": "Mars transfer",
"start_at": "2060-03-12T09:00",
"end_at": "2060-03-13T09:00",
"state_node_id": [""],
"state_node_title": ["Arrival burn"],
"state_node_at": ["2060-03-12T10:00"],
"state_node_detail": ["Inserted into Mars orbit"],
"state_node_state_label": ["transit-complete"],
"state_node_previous_location": ["LEO"],
"state_node_transit_location": ["Transfer"],
"state_node_target_location": ["Mars Orbit"],
},
)
self.assertEqual(response.status_code, 302)
entry = self._latest_entry()
self.assertEqual(len(entry.state_nodes), 1)
node = entry.state_nodes[0]
self.assertEqual(node.state_label, "transit-complete")
self.assertEqual(node.previous_location, "LEO")
self.assertEqual(node.transit_location, "Transfer")
self.assertEqual(node.target_location, "Mars Orbit")
def test_form_edit_state_nodes_updates_all_location_and_state_fields(self) -> None:
self.client.post(
f"/assets/{self.asset_id}/entries/new",
data={
"entry_kind": "state",
"title": "Initial interval",
"start_at": "2060-03-12T09:00",
"end_at": "2060-03-13T09:00",
"state_node_id": [""],
"state_node_title": ["Initial node"],
"state_node_at": ["2060-03-12T10:00"],
"state_node_detail": ["Before edit"],
"state_node_state_label": ["Docked"],
"state_node_previous_location": ["LEO"],
"state_node_transit_location": ["Transfer"],
"state_node_target_location": ["Mars Orbit"],
},
)
entry = self._latest_entry()
node_id = entry.state_nodes[0].id
response = self.client.post(
f"/assets/{self.asset_id}/entries/{entry.id}/edit",
data={
"entry_kind": "state",
"title": "Edited interval",
"start_at": "2060-03-12T09:00",
"end_at": "2060-03-13T09:00",
"state_node_id": [str(node_id)],
"state_node_title": ["Edited node"],
"state_node_at": ["2060-03-12T11:00"],
"state_node_detail": ["After edit"],
"state_node_state_label": ["Exploration"],
"state_node_previous_location": ["Mars Orbit"],
"state_node_transit_location": ["Transfer"],
"state_node_target_location": ["Mars Surface"],
},
)
self.assertEqual(response.status_code, 302)
updated = self._latest_entry()
node = updated.state_nodes[0]
self.assertEqual(node.title, "Edited node")
self.assertEqual(node.state_label, "Exploration")
self.assertEqual(node.previous_location, "Mars Orbit")
self.assertEqual(node.transit_location, "Transfer")
self.assertEqual(node.target_location, "Mars Surface")
def test_json_state_nodes_save_all_location_and_state_fields(self) -> None:
response = self.client.post(
f"/assets/{self.asset_id}/entries/new",
json={
"entry_kind": "state",
"title": "Quick state",
"start_at": "2060-03-12T09:00",
"end_at": "2060-03-13T09:00",
"state_label": "Exploration",
"location": "Mars Surface",
"state_nodes": [
{
"title": "Landed",
"at_time": "2060-03-12T10:00",
"detail": "Surface operations",
"state_label": "Exploration",
"previous_location": "Mars Orbit",
"transit_location": "Transfer",
"target_location": "Mars Surface",
}
],
},
)
self.assertEqual(response.status_code, 200)
entry = self._latest_entry()
self.assertEqual(len(entry.state_nodes), 1)
node = entry.state_nodes[0]
self.assertEqual(node.state_label, "Exploration")
self.assertEqual(node.previous_location, "Mars Orbit")
self.assertEqual(node.transit_location, "Transfer")
self.assertEqual(node.target_location, "Mars Surface")
def test_json_event_without_start_at_uses_sim_time(self) -> None:
response = self.client.post(
f"/assets/{self.asset_id}/entries/new",
json={
"entry_kind": "event",
"title": "Quick event",
"start_at": None,
"sim_time": "2060-03-12T09:00",
"location": "LEO",
"summary": "Created from the quick modal",
},
)
self.assertEqual(response.status_code, 200, response.get_data(as_text=True))
entry = self._latest_entry()
self.assertEqual(entry.entry_kind, "event")
self.assertEqual(entry.start_at, datetime(2060, 3, 12, 9, 0))
self.assertEqual(entry.state_nodes[0].target_location, "LEO")
def test_form_event_location_is_saved_for_display(self) -> None:
response = self.client.post(
f"/assets/{self.asset_id}/entries/new",
data={
"entry_kind": "event",
"title": "Quick sighting",
"start_at": "2060-03-12T09:00",
"location": "Mars Orbit",
"summary": "Point event with a location",
},
)
self.assertEqual(response.status_code, 302)
entry = self._latest_entry()
self.assertEqual(entry.entry_kind, "event")
self.assertEqual(len(entry.state_nodes), 1)
self.assertEqual(entry.state_nodes[0].target_location, "Mars Orbit")
def test_dock_existing_vehicle_creates_active_parent_child_relation_and_logs(self) -> None:
with self.app.app_context():
child = Asset(name="Codex Test Shuttle", asset_type="Vehicle", home_region="LEO")
db.session.add(child)
db.session.commit()
child_id = child.id
response = self.client.post(
f"/api/v1/assets/{self.asset_id}/dock",
json={
"mode": "dock_vehicle",
"vehicle_id": str(child_id),
"vehicle_name": "Codex Test Shuttle",
"sim_time": "2060-03-12T09:00",
"note": "Docked in test",
},
)
self.assertEqual(response.status_code, 200, response.get_data(as_text=True))
with self.app.app_context():
event = db.session.query(DockingEvent).one()
self.assertEqual(event.parent_asset_id, self.asset_id)
self.assertEqual(event.child_asset_id, child_id)
self.assertIsNone(event.undocked_at)
parent = db.session.get(Asset, self.asset_id)
child = db.session.get(Asset, child_id)
parent_snapshot = self.app.view_functions["web.asset_detail"].__globals__["_build_asset_snapshot"](
parent, datetime(2060, 3, 12, 9, 0)
)
child_snapshot = self.app.view_functions["web.asset_detail"].__globals__["_build_asset_snapshot"](
child, datetime(2060, 3, 12, 9, 0)
)
self.assertEqual(parent_snapshot["docked_vehicles"][0]["id"], child_id)
self.assertEqual(child_snapshot["docking_target"]["id"], self.asset_id)
self.assertEqual(len(parent.log_entries), 1)
self.assertEqual(len(child.log_entries), 1)
def test_dock_to_target_creates_reverse_relation(self) -> None:
with self.app.app_context():
target = Asset(name="Codex Test Station", asset_type="Station", home_region="Mars Orbit")
db.session.add(target)
db.session.commit()
target_id = target.id
response = self.client.post(
f"/api/v1/assets/{self.asset_id}/dock",
json={
"mode": "dock_to_target",
"target_asset_id": str(target_id),
"sim_time": "2060-03-12T09:00",
},
)
self.assertEqual(response.status_code, 200, response.get_data(as_text=True))
with self.app.app_context():
event = db.session.query(DockingEvent).one()
self.assertEqual(event.parent_asset_id, target_id)
self.assertEqual(event.child_asset_id, self.asset_id)
current = db.session.get(Asset, self.asset_id)
target = db.session.get(Asset, target_id)
current_snapshot = self.app.view_functions["web.asset_detail"].__globals__["_build_asset_snapshot"](
current, datetime(2060, 3, 12, 9, 0)
)
target_snapshot = self.app.view_functions["web.asset_detail"].__globals__["_build_asset_snapshot"](
target, datetime(2060, 3, 12, 9, 0)
)
self.assertEqual(current_snapshot["docking_target"]["id"], target_id)
self.assertEqual(target_snapshot["docked_vehicles"][0]["id"], self.asset_id)
def test_dock_custom_vehicle_creates_label_relation(self) -> None:
response = self.client.post(
f"/api/v1/assets/{self.asset_id}/dock",
json={
"mode": "dock_vehicle",
"vehicle_id": None,
"vehicle_name": "External Cargo Pod",
"sim_time": "2060-03-12T09:00",
},
)
self.assertEqual(response.status_code, 200, response.get_data(as_text=True))
with self.app.app_context():
event = db.session.query(DockingEvent).one()
self.assertEqual(event.parent_asset_id, self.asset_id)
self.assertIsNone(event.child_asset_id)
self.assertEqual(event.child_label, "External Cargo Pod")
if __name__ == "__main__":
unittest.main()
+207
View File
@@ -0,0 +1,207 @@
"""KSP管理后台 前端迁移 E2E 测试"""
import asyncio, json
from pathlib import Path
from playwright.async_api import async_playwright
BASE = "http://127.0.0.1:9000"
SS = Path(__file__).resolve().parent.parent / "demo" / "test_screenshots"
SS.mkdir(exist_ok=True)
results = []
def T(name, ok, detail=""):
results.append({"name": name, "ok": ok, "detail": detail})
print(f" [{'OK' if ok else 'FAIL'}] {name}" + (f" -- {detail}" if detail and not ok else ""))
async def run():
async with async_playwright() as pw:
b = await pw.chromium.launch(headless=True)
pg = await b.new_page(viewport={"width":1440,"height":900})
errs = []
pg.on("console", lambda m: errs.append(m.text) if m.type=="error" else None)
# ─── M1: 基础框架 ───
print("\n=== M1: 基础框架 ===")
await pg.goto(f"{BASE}/", wait_until="networkidle", timeout=10000)
await pg.wait_for_timeout(800)
items = await pg.query_selector_all("#sidebar .ni")
T("1.1 侧边栏导航", len(items)>=8, f"{len(items)} items")
secs = await pg.query_selector_all(".nst")
exp = 0
for s in secs:
if not (await s.evaluate("el => el.classList.contains('on')")):
exp += 1
T("1.2 侧边栏默认展开", exp>=2, f"{exp} expanded")
act = await pg.query_selector_all(".ni.act")
T("1.3 当前页高亮", len(act)>=1)
theme = await pg.evaluate("()=>document.documentElement.getAttribute('data-theme')")
T("1.4 默认深色模式", theme=="dark", f"theme={theme}")
await pg.evaluate("()=>toggleTheme()")
await pg.wait_for_timeout(200)
T("1.5 浅色模式", await pg.evaluate("()=>document.documentElement.getAttribute('data-theme')")=="light")
await pg.evaluate("()=>toggleTheme()")
await pg.wait_for_timeout(200)
T("1.6 深色模式切换回", await pg.evaluate("()=>document.documentElement.getAttribute('data-theme')")=="dark")
td = await pg.text_content("#sim-time-display")
T("1.7 模拟时间显示", "2060" in (td or ""))
await pg.evaluate("()=>openTimeModal()")
await pg.wait_for_timeout(300)
T("1.12 Modal打开", await pg.evaluate("()=>document.getElementById('time-modal').classList.contains('open')"))
await pg.evaluate("()=>closeTimeModal()")
T("1.14 Cancel关闭", await pg.evaluate("()=>!document.getElementById('time-modal').classList.contains('open')"))
await pg.evaluate("()=>openTimeModal()")
await pg.keyboard.press("Escape")
await pg.wait_for_timeout(200)
T("1.15 ESC关闭", await pg.evaluate("()=>!document.getElementById('time-modal').classList.contains('open')"))
# ─── M2: 任务总览 ───
print("\n=== M2: 任务总览 ===")
h1 = await pg.text_content("h1")
T("2.1 标题", h1 and "任务总览" in h1)
cards = await pg.query_selector_all(".mc")
T("2.2 指标卡片", len(cards)>=3, f"{len(cards)} cards")
ac = await pg.text_content(".mc.ac .v")
T("2.3 真实数据", ac and int(ac)>0, f"count={ac}")
grps = await pg.query_selector_all(".sgi")
T("2.4 状态看板", len(grps)>=2, f"{len(grps)} groups")
links = await pg.query_selector_all(".srw .an")
T("2.6 资产链接", len(links)>=1)
# ─── M3: 资产列表 ───
print("\n=== M3: 资产列表 ===")
await pg.goto(f"{BASE}/assets", wait_until="networkidle", timeout=10000)
await pg.wait_for_timeout(500)
rows = await pg.query_selector_all("table tbody tr")
T("3.1 列表渲染", len(rows)>=1, f"{len(rows)} rows")
await pg.fill('input[name="q"]', "ST-01")
await pg.click('button[type="submit"]')
await pg.wait_for_timeout(500)
rows2 = await pg.query_selector_all("table tbody tr")
T("3.2 搜索", len(rows2)<len(rows), f"{len(rows)}->{len(rows2)}")
# ─── M4: 资产详情 ───
print("\n=== M4: 资产详情 ===")
await pg.goto(f"{BASE}/assets", wait_until="networkidle", timeout=10000)
await pg.wait_for_timeout(500)
fl = await pg.query_selector("table tbody tr:first-child td:first-child a")
if fl:
await fl.click()
await pg.wait_for_timeout(1000)
h1d = await pg.text_content("h1")
T("4.1 详情渲染", h1d and len(h1d)>0)
cards = await pg.query_selector_all(".mc")
T("4.2 信息卡片", len(cards)>=4, f"{len(cards)} cards")
lrows = await pg.query_selector_all("table tbody tr")
T("4.5 日志列表", len(lrows)>=1)
# ─── M5: 对接 ───
print("\n=== M5: 对接功能 ===")
dock = await pg.query_selector('button:text("+ Dock")')
if dock:
await dock.click()
await pg.wait_for_timeout(500)
T("5.1 Dock按钮", await pg.evaluate("()=>document.getElementById('dock-modal')?.classList.contains('open')"))
opts = await pg.query_selector_all("#dock-vehicle option")
T("5.2 Modal内容", len(opts)>=2, f"{len(opts)} options")
await pg.evaluate("()=>closeModal('dock-modal')")
T("5.6 Cancel", True)
else:
T("5.1 Dock按钮", False, "not found")
# ─── M6: 日志CRUD ───
print("\n=== M6: 日志CRUD ===")
sb = await pg.query_selector('button:text("+ State Interval")')
if sb:
await sb.click()
await pg.wait_for_timeout(500)
T("6.1 State Interval", await pg.evaluate("()=>document.getElementById('log-modal')?.classList.contains('open')"))
ab = await pg.query_selector('button:text("+ Add Node")')
if ab:
await ab.click(); await ab.click()
T("6.5 Node添加", await pg.evaluate("()=>logNodeData.length")>=2)
await pg.evaluate("()=>closeModal('log-modal')")
T("6.10 Modal关闭", True)
else:
T("6.1 State Interval", False, "not found")
# ─── M7: 地点板 ───
print("\n=== M7: 地点板 ===")
await pg.goto(f"{BASE}/mission-preview/location-board", wait_until="networkidle", timeout=10000)
await pg.wait_for_timeout(1500)
T("7.1 Canvas", await pg.query_selector("#map-cv") is not None)
sc = await pg.text_content("#mi-title")
T("7.2 范围", sc and "Solar" in sc)
# ─── M8: 时间线 ───
print("\n=== M8: 时间线 ===")
await pg.goto(f"{BASE}/mission-preview/timeline?sim_time=2060-03-12T09:00", wait_until="networkidle", timeout=10000)
await pg.wait_for_timeout(1500)
T("8.1 Canvas", await pg.query_selector("#tl-canvas") is not None)
T("8.2 刻度行", await pg.query_selector("#tl-canvas > div:first-child") is not None)
rows = await pg.query_selector_all(".tl-fl")
T("8.3 段", len(rows)>=1, f"{len(rows)} segments")
evts = await pg.query_selector_all(".tl-ev, .tl-sn")
T("8.4 事件", len(evts)>=1, f"{len(evts)} events")
links = await pg.query_selector_all("#tl-canvas a[href*='/assets/']")
T("8.5 资产链接", len(links)>=1, f"{len(links)} links")
has_json = await pg.evaluate("() => !!document.getElementById('tl-json')")
T("8.6 JSON数据", has_json)
await pg.mouse.wheel(0, 100)
await pg.wait_for_timeout(500)
T("8.7 滚轮缩放", True)
# ─── M9: 数据中心 ───
print("\n=== M9: 数据中心 ===")
for n,p in [("engines","/engines"),("comms","/communications"),("tanks","/tanks"),("vehicles","/vehicles")]:
await pg.goto(f"{BASE}{p}", wait_until="networkidle", timeout=10000)
await pg.wait_for_timeout(500)
r = await pg.query_selector_all("table tbody tr")
T(f"9.{n}", len(r)>=1, f"{len(r)} rows")
# ─── M10: 燃料换算 ───
print("\n=== M10: 燃料换算 ===")
await pg.goto(f"{BASE}/fuel-converter", wait_until="networkidle", timeout=10000)
await pg.wait_for_timeout(500)
T("10.1 渲染", await pg.query_selector("#conv-form") is not None)
T("JS Errors", len(errs)==0, f"{len(errs)} errors")
if errs: print(f" Errors: {errs[:3]}")
await b.close()
# Report
print("\n"+"="*60)
print("TEST REPORT")
print("="*60)
passed=sum(1 for r in results if r["ok"])
failed=sum(1 for r in results if not r["ok"])
total=len(results)
print(f"Total: {total} | Passed: {passed} | Failed: {failed} | Rate: {passed/total*100:.1f}%")
if failed:
print("\nFailed:")
for r in results:
if not r["ok"]:
print(f" [FAIL] {r['name']}" + (f" -- {r['detail']}" if r['detail'] else ""))
print("="*60)
with open(SS/"test_report.json","w") as f: json.dump({"total":total,"passed":passed,"failed":failed,"results":results},f,ensure_ascii=False,indent=2)
print(f"Report: {SS}/test_report.json")
return failed==0
if __name__=="__main__":
asyncio.run(run())