218 lines
8.8 KiB
Python
218 lines
8.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
import tools.validate_snapshot_admin_web_v1 as validator
|
|
from src.quant_engine.snapshot_admin_server_v1 import (
|
|
build_ui_state,
|
|
fetch_domain_rows,
|
|
fetch_table_rows,
|
|
list_browsable_tables,
|
|
render_collection_html,
|
|
render_index_html,
|
|
render_tables_html,
|
|
)
|
|
from src.quant_engine.snapshot_admin_store_v1 import import_seed_json
|
|
|
|
|
|
def _write_valid_seed(path: Path) -> None:
|
|
payload = {
|
|
"data": {
|
|
"settings": [
|
|
{"ordinal": 1, "key": "total_asset_krw", "value": 500000000, "note": "seed"},
|
|
{"ordinal": 2, "key": "settlement_cash_d2_krw", "value": 250000000, "note": "seed"},
|
|
],
|
|
"account_snapshot": [
|
|
{
|
|
"captured_at": "2026-06-22T11:15:47+09:00",
|
|
"account": "demo",
|
|
"account_type": "일반계좌",
|
|
"ticker": "005930",
|
|
"name": "삼성전자",
|
|
"holding_quantity": 10,
|
|
"average_cost": 70000,
|
|
"parse_status": "NOT_PROVIDED",
|
|
"position_type": "core",
|
|
}
|
|
],
|
|
}
|
|
}
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
class TestSnapshotAdminWebV1(unittest.TestCase):
|
|
|
|
def test_render_index_html_contains_spreadsheet_surface(self):
|
|
html = render_index_html()
|
|
self.assertIn("Snapshot Admin", html)
|
|
self.assertIn("contenteditable", html)
|
|
self.assertIn("/api/settings/save", html)
|
|
self.assertIn("/api/account_snapshot/save", html)
|
|
self.assertIn("Lock target", html)
|
|
self.assertIn("Lock row", html)
|
|
self.assertIn("Approve pending", html)
|
|
self.assertIn("Refresh diff", html)
|
|
self.assertIn("Export approval packet", html)
|
|
self.assertIn("Selection Inspector", html)
|
|
self.assertIn("Recent row history", html)
|
|
self.assertIn("Save view", html)
|
|
self.assertIn("Apply TSV to selection", html)
|
|
self.assertIn("Ctrl+S", html)
|
|
self.assertIn("KIS Collection", html)
|
|
self.assertIn("Recent collector snapshots", html)
|
|
self.assertIn("Collection detail", html)
|
|
self.assertIn("Filter runs / snapshots / errors", html)
|
|
self.assertIn("Filter change log", html)
|
|
self.assertIn("Timeline", html)
|
|
self.assertIn("/collection", html)
|
|
self.assertIn("Open collection dashboard", html)
|
|
|
|
def test_render_collection_html_contains_dashboard_surface(self):
|
|
html = render_collection_html()
|
|
self.assertIn("KIS Collection Dashboard", html)
|
|
self.assertIn("/api/state", html)
|
|
self.assertIn("Download raw JSON", html)
|
|
self.assertIn("Download CSV", html)
|
|
self.assertIn("Filter runs / snapshots / errors", html)
|
|
self.assertIn("Ticker quick search", html)
|
|
self.assertIn("Date quick search", html)
|
|
|
|
def test_build_ui_state_exposes_expected_columns(self):
|
|
import tempfile
|
|
import shutil
|
|
tmp_dir = tempfile.mkdtemp()
|
|
try:
|
|
db_path = Path(tmp_dir) / "snapshot_admin.db"
|
|
seed_path = Path(tmp_dir) / "valid_seed.json"
|
|
_write_valid_seed(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
state = build_ui_state(db_path)
|
|
self.assertTrue(state["summary"]["settings_rows"] > 0)
|
|
self.assertTrue(state["summary"]["account_snapshot_rows"] > 0)
|
|
self.assertEqual(state["summary"]["topology"]["mode"], "single_workspace_sqlite")
|
|
self.assertTrue(state["summary"]["topology"]["settings_and_snapshot_share_db"])
|
|
self.assertTrue(state["summary"]["topology"]["collector_separate_db"])
|
|
self.assertEqual(state["account_snapshot_columns"][0], "captured_at")
|
|
self.assertIn("settings", state["validation"])
|
|
self.assertTrue(state["version"]["app"])
|
|
self.assertIn("fingerprint", state["version"]["source"])
|
|
self.assertIn("collection", state)
|
|
self.assertIn("counts", state["collection"])
|
|
self.assertIn("latest_report", state["collection"])
|
|
self.assertEqual(state["summary"]["topology"]["mode"], "single_workspace_sqlite")
|
|
finally:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
def test_snapshot_admin_workflow_and_script_exist(self):
|
|
workflow = ROOT / ".gitea" / "workflows" / "snapshot_admin.yml"
|
|
package = json.loads((ROOT / "package.json").read_text(encoding="utf-8"))
|
|
self.assertTrue(workflow.exists())
|
|
self.assertIn("--reload", package["scripts"]["ops:snapshot-web"])
|
|
self.assertIn("ops:snapshot-validate", package["scripts"])
|
|
self.assertIn("ops:snapshot-web-validate", package["scripts"])
|
|
|
|
def test_render_tables_html_contains_tabler_grid_surface(self):
|
|
html = render_tables_html()
|
|
self.assertIn("tabler", html.lower())
|
|
self.assertIn("tableSelect", html)
|
|
self.assertIn("/api/tables", html)
|
|
self.assertIn("/api/table_rows", html)
|
|
self.assertIn("/api/domain_rows", html)
|
|
self.assertIn("saveCurrentTable", html)
|
|
self.assertIn("gridTable", html)
|
|
|
|
def test_list_browsable_tables_covers_all_three_databases(self):
|
|
import tempfile
|
|
import shutil
|
|
tmp_dir = tempfile.mkdtemp()
|
|
try:
|
|
db_path = Path(tmp_dir) / "snapshot_admin.db"
|
|
seed_path = Path(tmp_dir) / "valid_seed.json"
|
|
_write_valid_seed(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
tables = list_browsable_tables(db_path)
|
|
names = {row["table"] for row in tables}
|
|
self.assertTrue({"settings", "account_snapshot", "workspace_change_log"} <= names)
|
|
self.assertTrue({"collection_runs", "collection_snapshots", "collection_source_errors"} <= names)
|
|
self.assertTrue({"sell_strategy_results", "satellite_recommendations"} <= names)
|
|
|
|
settings_row = next(row for row in tables if row["table"] == "settings")
|
|
self.assertTrue(settings_row["exists"])
|
|
self.assertTrue(settings_row["row_count"] > 0)
|
|
finally:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
def test_fetch_table_rows_paginates_and_rejects_unknown_table(self):
|
|
import tempfile
|
|
import shutil
|
|
tmp_dir = tempfile.mkdtemp()
|
|
try:
|
|
db_path = Path(tmp_dir) / "snapshot_admin.db"
|
|
seed_path = Path(tmp_dir) / "valid_seed.json"
|
|
_write_valid_seed(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
page1 = fetch_table_rows("settings", db_path, limit=2, offset=0)
|
|
self.assertTrue(page1["columns"])
|
|
self.assertEqual(len(page1["rows"]), 2)
|
|
self.assertTrue(page1["total"] >= 2)
|
|
|
|
page2 = fetch_table_rows("settings", db_path, limit=2, offset=2)
|
|
self.assertNotEqual(page1["rows"], page2["rows"])
|
|
|
|
with self.assertRaises(ValueError):
|
|
fetch_table_rows("settings; DROP TABLE settings;--", db_path)
|
|
finally:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
def test_fetch_domain_rows_exposes_editable_tables(self):
|
|
import tempfile
|
|
import shutil
|
|
tmp_dir = tempfile.mkdtemp()
|
|
try:
|
|
db_path = Path(tmp_dir) / "snapshot_admin.db"
|
|
seed_path = Path(tmp_dir) / "valid_seed.json"
|
|
_write_valid_seed(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
settings = fetch_domain_rows("settings", db_path)
|
|
snapshot = fetch_domain_rows("account_snapshot", db_path)
|
|
self.assertEqual(settings["domain"], "settings")
|
|
self.assertTrue(settings["rows"])
|
|
self.assertEqual(snapshot["domain"], "account_snapshot")
|
|
self.assertTrue(snapshot["rows"])
|
|
|
|
with self.assertRaises(ValueError):
|
|
fetch_domain_rows("workspace_change_log", db_path)
|
|
finally:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
|
|
|
|
def test_snapshot_admin_web_validation_script_passes(self):
|
|
out = ROOT / "Temp" / "snapshot_admin_web_validation_v1.json"
|
|
if out.exists():
|
|
out.unlink()
|
|
|
|
rc = validator.main()
|
|
payload = json.loads(out.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(rc, 0)
|
|
self.assertEqual(payload["gate"], "PASS")
|
|
self.assertEqual(payload["formula_id"], "SNAPSHOT_ADMIN_WEB_VALIDATION_V1")
|
|
self.assertTrue(payload["settings_rows"] > 0)
|
|
self.assertTrue(payload["account_snapshot_rows"] > 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|