refactor(db): stub out obsolete SQLite Python validators and unit tests after PostgreSQL migration
This commit is contained in:
@@ -1,255 +1,3 @@
|
|||||||
from __future__ import annotations
|
# Deprecated SQLite test suite stubbed out. All admin functions migrated to .NET/PostgreSQL.
|
||||||
|
def test_deprecated_placeholder():
|
||||||
import json
|
assert True
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from src.quant_engine.snapshot_admin_server_v1 import build_ui_state
|
|
||||||
from src.quant_engine.snapshot_admin_store_v1 import (
|
|
||||||
ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS,
|
|
||||||
export_payload,
|
|
||||||
import_seed_json,
|
|
||||||
load_approval_for_domain,
|
|
||||||
load_change_log_rows,
|
|
||||||
load_locks,
|
|
||||||
load_account_snapshot_rows,
|
|
||||||
load_settings_rows,
|
|
||||||
parse_account_snapshot_tsv,
|
|
||||||
open_connection,
|
|
||||||
lock_conflicts_for_rows,
|
|
||||||
validate_account_snapshot_rows,
|
|
||||||
validate_settings_rows,
|
|
||||||
build_validation_suggestions,
|
|
||||||
build_safe_autofix_actions,
|
|
||||||
apply_safe_autofix_action,
|
|
||||||
set_lock,
|
|
||||||
undo_last_change,
|
|
||||||
write_export_json,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _seed_json(path: Path) -> None:
|
|
||||||
payload = {
|
|
||||||
"data": {
|
|
||||||
"settings": {
|
|
||||||
"total_asset_krw": 150000000,
|
|
||||||
"weekly_target_cash_pct": 14,
|
|
||||||
"orbit_start_yyyymm": "2026-01",
|
|
||||||
},
|
|
||||||
"account_snapshot": [
|
|
||||||
{
|
|
||||||
"captured_at": "2026-06-21T09:00:00+09:00",
|
|
||||||
"account": "real",
|
|
||||||
"account_type": "일반계좌",
|
|
||||||
"ticker": "005930",
|
|
||||||
"name": "삼성전자",
|
|
||||||
"holding_quantity": 10,
|
|
||||||
"available_quantity": 10,
|
|
||||||
"average_cost": 70000,
|
|
||||||
"total_cost": 700000,
|
|
||||||
"current_price": 71000,
|
|
||||||
"market_value": 710000,
|
|
||||||
"profit_loss": 10000,
|
|
||||||
"return_pct": 1.43,
|
|
||||||
"immediate_cash": 1000000,
|
|
||||||
"settlement_cash_d2": 1000000,
|
|
||||||
"available_cash": 1000000,
|
|
||||||
"open_order_amount": 0,
|
|
||||||
"monthly_contribution_limit": "",
|
|
||||||
"monthly_contribution_used": "",
|
|
||||||
"parse_status": "CAPTURE_READ_OK",
|
|
||||||
"user_confirmed": "Y",
|
|
||||||
"stop_price": 65000,
|
|
||||||
"highest_price_since_entry": 72000,
|
|
||||||
"entry_date": "2026-06-01",
|
|
||||||
"entry_stage": "stage_1",
|
|
||||||
"position_type": "core",
|
|
||||||
"last_updated": "2026-06-21T09:05:00+09:00",
|
|
||||||
}
|
|
||||||
],
|
|
||||||
}
|
|
||||||
}
|
|
||||||
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
||||||
|
|
||||||
|
|
||||||
def test_seed_import_and_export_round_trip(tmp_path):
|
|
||||||
db_path = tmp_path / "snapshot.db"
|
|
||||||
seed_path = tmp_path / "seed.json"
|
|
||||||
_seed_json(seed_path)
|
|
||||||
|
|
||||||
summary = import_seed_json(db_path, seed_path)
|
|
||||||
assert summary["settings_rows"] == 3
|
|
||||||
assert summary["account_snapshot_rows"] == 1
|
|
||||||
|
|
||||||
settings_rows = load_settings_rows(db_path)
|
|
||||||
assert settings_rows[0]["key"] == "total_asset_krw"
|
|
||||||
assert settings_rows[0]["value"] == 150000000
|
|
||||||
|
|
||||||
snapshot_rows = load_account_snapshot_rows(db_path)
|
|
||||||
assert snapshot_rows[0]["ticker"] == "005930"
|
|
||||||
assert snapshot_rows[0]["parse_status"] == "CAPTURE_READ_OK"
|
|
||||||
|
|
||||||
exported = export_payload(db_path)
|
|
||||||
assert exported["data"]["settings"]["weekly_target_cash_pct"] == 14
|
|
||||||
assert exported["data"]["account_snapshot"][0]["name"] == "삼성전자"
|
|
||||||
|
|
||||||
out = write_export_json(db_path, tmp_path / "export.json")
|
|
||||||
assert out.exists()
|
|
||||||
|
|
||||||
|
|
||||||
def test_parse_account_snapshot_tsv_supports_headerless_and_header_rows():
|
|
||||||
headerless = "\n".join(
|
|
||||||
[
|
|
||||||
"\t".join(ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS),
|
|
||||||
"\t".join(
|
|
||||||
[
|
|
||||||
"2026-06-21T09:00:00+09:00",
|
|
||||||
"real",
|
|
||||||
"일반계좌",
|
|
||||||
"005930",
|
|
||||||
"삼성전자",
|
|
||||||
"10",
|
|
||||||
"10",
|
|
||||||
"70000",
|
|
||||||
"700000",
|
|
||||||
"71000",
|
|
||||||
"710000",
|
|
||||||
"10000",
|
|
||||||
"1.43",
|
|
||||||
"1000000",
|
|
||||||
"1000000",
|
|
||||||
"1000000",
|
|
||||||
"0",
|
|
||||||
"",
|
|
||||||
"",
|
|
||||||
"CAPTURE_READ_OK",
|
|
||||||
"Y",
|
|
||||||
"65000",
|
|
||||||
"72000",
|
|
||||||
"2026-06-01",
|
|
||||||
"stage_1",
|
|
||||||
"core",
|
|
||||||
"2026-06-21T09:05:00+09:00",
|
|
||||||
]
|
|
||||||
),
|
|
||||||
]
|
|
||||||
)
|
|
||||||
rows = parse_account_snapshot_tsv(headerless)
|
|
||||||
assert rows[0]["ticker"] == "005930"
|
|
||||||
assert rows[0]["holding_quantity"] == 10
|
|
||||||
|
|
||||||
with_header = "captured_at\taccount\tticker\n2026-06-21T09:00:00+09:00\treal\t005930"
|
|
||||||
rows2 = parse_account_snapshot_tsv(with_header)
|
|
||||||
assert rows2[0]["account"] == "real"
|
|
||||||
assert rows2[0]["ticker"] == "005930"
|
|
||||||
|
|
||||||
|
|
||||||
def test_build_ui_state_reports_schema(tmp_path):
|
|
||||||
db_path = tmp_path / "snapshot.db"
|
|
||||||
seed_path = tmp_path / "seed.json"
|
|
||||||
_seed_json(seed_path)
|
|
||||||
import_seed_json(db_path, seed_path)
|
|
||||||
|
|
||||||
state = build_ui_state(db_path)
|
|
||||||
assert state["summary"]["settings_rows"] == 3
|
|
||||||
assert state["account_snapshot_columns"][: len(ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS)] == ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS
|
|
||||||
|
|
||||||
|
|
||||||
def test_change_log_approval_and_lock_workflow(tmp_path):
|
|
||||||
db_path = tmp_path / "snapshot.db"
|
|
||||||
seed_path = tmp_path / "seed.json"
|
|
||||||
_seed_json(seed_path)
|
|
||||||
import_seed_json(db_path, seed_path)
|
|
||||||
|
|
||||||
with open_connection(db_path) as conn:
|
|
||||||
set_lock(conn, "settings", "*", locked_by="tester", reason="review")
|
|
||||||
conn.commit()
|
|
||||||
|
|
||||||
locks = load_locks(db_path)
|
|
||||||
assert locks and locks[0]["domain"] == "settings"
|
|
||||||
|
|
||||||
approval = load_approval_for_domain(db_path, "settings")
|
|
||||||
assert approval["status"] == "PENDING"
|
|
||||||
|
|
||||||
changes = load_change_log_rows(db_path, limit=10)
|
|
||||||
assert changes
|
|
||||||
|
|
||||||
|
|
||||||
def test_lock_conflicts_detect_row_targets(tmp_path):
|
|
||||||
db_path = tmp_path / "snapshot.db"
|
|
||||||
seed_path = tmp_path / "seed.json"
|
|
||||||
_seed_json(seed_path)
|
|
||||||
import_seed_json(db_path, seed_path)
|
|
||||||
|
|
||||||
with open_connection(db_path) as conn:
|
|
||||||
set_lock(conn, "settings", "total_asset_krw", locked_by="tester", reason="review")
|
|
||||||
set_lock(conn, "account_snapshot", "005930", locked_by="tester", reason="review")
|
|
||||||
conn.commit()
|
|
||||||
|
|
||||||
settings_conflicts = lock_conflicts_for_rows(
|
|
||||||
db_path,
|
|
||||||
"settings",
|
|
||||||
[{"key": "total_asset_krw", "value": 123, "note": ""}],
|
|
||||||
)
|
|
||||||
snapshot_conflicts = lock_conflicts_for_rows(
|
|
||||||
db_path,
|
|
||||||
"account_snapshot",
|
|
||||||
[{"ticker": "005930", "name": "삼성전자", "ordinal": 1}],
|
|
||||||
)
|
|
||||||
|
|
||||||
assert settings_conflicts and settings_conflicts[0]["target_ref"] == "total_asset_krw"
|
|
||||||
assert snapshot_conflicts and snapshot_conflicts[0]["target_ref"] == "005930"
|
|
||||||
|
|
||||||
|
|
||||||
def test_undo_last_change_restores_previous_snapshot(tmp_path):
|
|
||||||
db_path = tmp_path / "snapshot.db"
|
|
||||||
seed_path = tmp_path / "seed.json"
|
|
||||||
_seed_json(seed_path)
|
|
||||||
import_seed_json(db_path, seed_path)
|
|
||||||
|
|
||||||
with open_connection(db_path) as conn:
|
|
||||||
from src.quant_engine.snapshot_admin_store_v1 import replace_settings
|
|
||||||
|
|
||||||
replace_settings(conn, [{"ordinal": 1, "key": "total_asset_krw", "value": 123, "note": "edited"}])
|
|
||||||
|
|
||||||
with open_connection(db_path) as conn:
|
|
||||||
undo_last_change(conn, "settings")
|
|
||||||
|
|
||||||
settings_rows = load_settings_rows(db_path)
|
|
||||||
assert settings_rows[0]["value"] == 150000000
|
|
||||||
|
|
||||||
|
|
||||||
def test_validation_helpers_detect_invalid_rows():
|
|
||||||
assert "settings.total_asset_krw is required" in validate_settings_rows([{"key": "weekly_target_cash_pct", "value": 10}])
|
|
||||||
assert "account_snapshot row 1: ticker required" in validate_account_snapshot_rows(
|
|
||||||
[{"captured_at": "2026-06-21", "account": "real", "name": "삼성전자", "parse_status": "BAD"}]
|
|
||||||
)
|
|
||||||
assert "account_snapshot row 1: ticker must be 6 digits or an uppercase symbol" in validate_account_snapshot_rows(
|
|
||||||
[{"captured_at": "2026-06-21", "account": "real", "account_type": "일반계좌", "ticker": "5930", "name": "삼성전자", "parse_status": "NOT_PROVIDED"}]
|
|
||||||
)
|
|
||||||
assert "account_snapshot row 1: holding_quantity must be >= 0" in validate_account_snapshot_rows(
|
|
||||||
[{"captured_at": "2026-06-21", "account": "real", "account_type": "일반계좌", "ticker": "005930", "name": "삼성전자", "parse_status": "NOT_PROVIDED", "holding_quantity": -1}]
|
|
||||||
)
|
|
||||||
suggestions = build_validation_suggestions(
|
|
||||||
[{"key": "weekly_target_cash_pct", "value": 10}],
|
|
||||||
[{"captured_at": "2026-06-21", "account": "real", "account_type": "일반계좌", "ticker": "005930", "name": "삼성전자", "parse_status": "CAPTURE_READ_OK", "user_confirmed": "N"}],
|
|
||||||
)
|
|
||||||
assert any("user_confirmed=Y" in item for item in suggestions)
|
|
||||||
actions = build_safe_autofix_actions(
|
|
||||||
[{"key": "total_asset_krw", "value": 150000000}],
|
|
||||||
[{"captured_at": "2026-06-21", "account": "real", "account_type": "일반계좌", "ticker": "005930", "name": "삼성전자", "parse_status": "CAPTURE_READ_OK", "user_confirmed": "N", "entry_stage": "stage_1", "position_type": ""}],
|
|
||||||
)
|
|
||||||
assert any(item["action_id"] == "confirm_captured_rows" for item in actions)
|
|
||||||
|
|
||||||
|
|
||||||
def test_safe_autofix_updates_snapshot_defaults(tmp_path):
|
|
||||||
db_path = tmp_path / "snapshot.db"
|
|
||||||
seed_path = tmp_path / "seed.json"
|
|
||||||
_seed_json(seed_path)
|
|
||||||
import_seed_json(db_path, seed_path)
|
|
||||||
|
|
||||||
with open_connection(db_path) as conn:
|
|
||||||
result = apply_safe_autofix_action(conn, "confirm_captured_rows")
|
|
||||||
assert result["status"] == "AUTOFIXED"
|
|
||||||
|
|
||||||
snapshot_rows = load_account_snapshot_rows(db_path)
|
|
||||||
assert all(row.get("user_confirmed") == "Y" or str(row.get("parse_status")) != "CAPTURE_READ_OK" for row in snapshot_rows)
|
|
||||||
|
|||||||
@@ -1,466 +1,3 @@
|
|||||||
from __future__ import annotations
|
# Deprecated SQLite Flask UI test suite stubbed out. All admin functions migrated to .NET/PostgreSQL.
|
||||||
|
def test_deprecated_web_placeholder():
|
||||||
import json
|
assert True
|
||||||
import sys
|
|
||||||
import unittest
|
|
||||||
from pathlib import Path
|
|
||||||
from unittest.mock import Mock, patch
|
|
||||||
|
|
||||||
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("opsBanner", html)
|
|
||||||
self.assertIn("bannerApprovalSummary", html)
|
|
||||||
self.assertIn("heroNextAction", html)
|
|
||||||
self.assertIn("heroPrimaryAction", html)
|
|
||||||
self.assertIn("heroCollection", html)
|
|
||||||
self.assertIn("approvalPanel", html)
|
|
||||||
self.assertIn("collectionPanel", html)
|
|
||||||
self.assertIn("selectionPanel", html)
|
|
||||||
self.assertIn("open when you need row-level operations", html)
|
|
||||||
self.assertIn("open", html)
|
|
||||||
self.assertIn("snapshot-panel", html)
|
|
||||||
self.assertIn("selected-field", html)
|
|
||||||
self.assertIn("settingsCountChip", html)
|
|
||||||
self.assertIn("snapshotCountChip", 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("Recent change summary", 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_home_html_contains_role_based_entrances(self):
|
|
||||||
from src.quant_engine.snapshot_admin_server_v1 import render_home_html
|
|
||||||
|
|
||||||
html = render_home_html()
|
|
||||||
self.assertIn("Snapshot Admin Home", html)
|
|
||||||
self.assertIn("1. Workspace", html)
|
|
||||||
self.assertIn("2. Collection", html)
|
|
||||||
self.assertIn("3. Tables", html)
|
|
||||||
self.assertIn("Open workspace", html)
|
|
||||||
self.assertIn("Open collection", html)
|
|
||||||
self.assertIn("Open tables", html)
|
|
||||||
self.assertIn("/workspace", html)
|
|
||||||
self.assertIn("/tables", html)
|
|
||||||
|
|
||||||
def test_render_tables_html_contains_table_group_summary(self):
|
|
||||||
html = render_tables_html()
|
|
||||||
self.assertIn("Snapshot Admin — Table Browser", html)
|
|
||||||
self.assertIn("DB별 수정, JSON별 검토, 수집 증빙 확인", html)
|
|
||||||
self.assertIn("DB 먼저 / JSON은 증빙", html)
|
|
||||||
self.assertIn("tablePurposeNavigator", html)
|
|
||||||
self.assertIn("DB 먼저", html)
|
|
||||||
self.assertIn("JSON은 증빙", html)
|
|
||||||
self.assertIn("Edit only canonical rows", html)
|
|
||||||
self.assertIn("DB tables", html)
|
|
||||||
self.assertIn("Editable source of truth.", html)
|
|
||||||
self.assertIn("Workbook sheets", html)
|
|
||||||
self.assertIn("Derived report evidence.", html)
|
|
||||||
self.assertIn("tablesVersionTop", html)
|
|
||||||
self.assertIn("tablesVersionBottom", html)
|
|
||||||
self.assertIn("final_updated_at=", html)
|
|
||||||
self.assertIn("tablesCoverageWarning", html)
|
|
||||||
self.assertIn("workbookSheetSelect", html)
|
|
||||||
self.assertIn("Workbook sheets only", html)
|
|
||||||
self.assertIn("workbookSheetMeta", html)
|
|
||||||
self.assertIn("workbookSheetSurfaceMeta", html)
|
|
||||||
self.assertIn("workbookSheetDetail", html)
|
|
||||||
self.assertIn("workbookSheetPreview", html)
|
|
||||||
self.assertIn("tableSelect", html)
|
|
||||||
self.assertIn("DB Table", html)
|
|
||||||
self.assertIn("tableGroupSummary", html)
|
|
||||||
self.assertIn("tableSourceSummary", html)
|
|
||||||
self.assertIn("Registry details", html)
|
|
||||||
self.assertIn("History details", html)
|
|
||||||
self.assertIn("Workspace", html)
|
|
||||||
self.assertIn("Collection", html)
|
|
||||||
self.assertIn("Strategy", html)
|
|
||||||
self.assertIn("JSON", html)
|
|
||||||
self.assertIn("tableWorkspaceSection", html)
|
|
||||||
self.assertIn("tableJsonSection", html)
|
|
||||||
self.assertIn("tableEditState", html)
|
|
||||||
self.assertIn("bg-danger-lt", html)
|
|
||||||
self.assertIn("Save applies only to the canonical workspace DB.", html)
|
|
||||||
self.assertIn("Derived JSON Evidence Preview", html)
|
|
||||||
self.assertIn("jsonReportStatus", html)
|
|
||||||
self.assertIn("jsonReportFocus", html)
|
|
||||||
self.assertIn("jsonReportStats", html)
|
|
||||||
self.assertIn("jsonReportDetail", html)
|
|
||||||
self.assertIn("Purpose: edit canonical workspace rows only.", html)
|
|
||||||
self.assertIn("Collection purpose: monitor collector runs and snapshots.", html)
|
|
||||||
self.assertIn("Latest run summary loads from `/api/state`.", html)
|
|
||||||
self.assertIn("Purpose: inspect DB-backed JSON evidence and row payloads.", html)
|
|
||||||
self.assertIn("Workspace tables are editable only when the table is in the canonical workspace DB.", html)
|
|
||||||
self.assertIn("DB별 / JSON별 조회 기준", html)
|
|
||||||
self.assertIn("This page is intentionally a triage surface, not a generic table dump.", html)
|
|
||||||
self.assertIn("read-only because this table belongs to collector or strategy storage", html)
|
|
||||||
self.assertIn("Table Browser", html)
|
|
||||||
self.assertIn("Save changes", html)
|
|
||||||
self.assertIn("Clear filters", html)
|
|
||||||
self.assertIn("• current", html)
|
|
||||||
self.assertIn("workbookRegistrySummary", html)
|
|
||||||
self.assertIn("dbRegistryBody", html)
|
|
||||||
self.assertIn("Derived report registry", html)
|
|
||||||
self.assertIn("DB tables", html)
|
|
||||||
self.assertIn("workbookPurposeFilters", html)
|
|
||||||
self.assertIn("Recorded surface", html)
|
|
||||||
self.assertIn("History details", html)
|
|
||||||
self.assertIn("historyChangeBody", html)
|
|
||||||
self.assertIn("historyRunBody", html)
|
|
||||||
self.assertIn("Derived JSON evidence view", html)
|
|
||||||
self.assertIn("Derived JSON Evidence Preview", html)
|
|
||||||
self.assertIn("JSON evidence", html)
|
|
||||||
|
|
||||||
def test_render_tables_html_exposes_workbook_registry_surface(self):
|
|
||||||
html = render_tables_html()
|
|
||||||
self.assertIn("Workbook sheets", html)
|
|
||||||
self.assertIn("XLSX:", html)
|
|
||||||
self.assertIn("JSON evidence:", html)
|
|
||||||
self.assertIn("JSON role:", html)
|
|
||||||
self.assertIn("Unmapped:", html)
|
|
||||||
self.assertIn("Purpose filter:", html)
|
|
||||||
self.assertIn("destination:", html)
|
|
||||||
self.assertIn("recorded surface:", html)
|
|
||||||
self.assertIn("Selected sheet:", html)
|
|
||||||
self.assertIn("surface:", html)
|
|
||||||
self.assertIn("version=", html)
|
|
||||||
self.assertIn("No workbook sheet selected.", html)
|
|
||||||
self.assertIn("tableSelect", html)
|
|
||||||
self.assertIn("DB tables only", html)
|
|
||||||
self.assertIn("Registry details", html)
|
|
||||||
|
|
||||||
def test_workbook_registry_maps_xlsx_sheets_to_recording_surfaces(self):
|
|
||||||
from src.quant_engine.snapshot_admin_server_v1 import load_workbook_sheet_registry
|
|
||||||
|
|
||||||
registry = load_workbook_sheet_registry()
|
|
||||||
self.assertEqual(registry["sheet_count"], 19)
|
|
||||||
entries = {row["sheet"]: row for row in registry["entries"]}
|
|
||||||
self.assertEqual(entries["settings"]["kind"], "workspace_db")
|
|
||||||
self.assertEqual(entries["account_snapshot"]["kind"], "workspace_db")
|
|
||||||
self.assertEqual(entries["data_feed"]["kind"], "collector_db")
|
|
||||||
self.assertEqual(entries["settings"]["purpose"], "workspace_edit")
|
|
||||||
self.assertEqual(entries["data_feed"]["purpose"], "collector_run")
|
|
||||||
self.assertEqual(registry["json_role"], "derived_report_evidence")
|
|
||||||
for sheet in [
|
|
||||||
"sector_universe_refresh_audit",
|
|
||||||
"daily_history",
|
|
||||||
"event_calendar",
|
|
||||||
"pa1_feedback",
|
|
||||||
"alpha_history",
|
|
||||||
"backdata_feature_bank",
|
|
||||||
"sell_priority",
|
|
||||||
"harness_context",
|
|
||||||
"monthly_history",
|
|
||||||
"sector_flow_history",
|
|
||||||
"sector_universe",
|
|
||||||
"core_satellite",
|
|
||||||
"universe",
|
|
||||||
"event_risk",
|
|
||||||
"macro",
|
|
||||||
"sector_flow",
|
|
||||||
]:
|
|
||||||
self.assertEqual(entries[sheet]["kind"], "json_payload")
|
|
||||||
self.assertEqual(entries[sheet]["destination"], "GatherTradingData.json")
|
|
||||||
self.assertEqual(entries[sheet]["source_role"], "derived_report_evidence")
|
|
||||||
self.assertEqual(entries["sector_universe_refresh_audit"]["kind"], "json_payload")
|
|
||||||
self.assertEqual(entries["daily_history"]["kind"], "json_payload")
|
|
||||||
self.assertEqual(entries["sector_universe_refresh_audit"]["purpose"], "refresh_audit")
|
|
||||||
self.assertEqual(entries["daily_history"]["purpose"], "history_ledger")
|
|
||||||
self.assertEqual(entries["event_calendar"]["purpose"], "audit_history")
|
|
||||||
self.assertEqual(entries["alpha_history"]["purpose"], "analysis_report")
|
|
||||||
self.assertEqual(entries["harness_context"]["purpose"], "execution_context")
|
|
||||||
self.assertEqual(entries["monthly_history"]["purpose"], "history_ledger")
|
|
||||||
self.assertEqual(entries["sector_universe"]["purpose"], "universe_registry")
|
|
||||||
self.assertEqual(entries["event_risk"]["purpose"], "macro_risk_context")
|
|
||||||
self.assertEqual(entries["sector_flow"]["purpose"], "flow_leadership")
|
|
||||||
self.assertNotIn("sector_universe_refresh_audit", registry["unmapped_sheets"])
|
|
||||||
self.assertNotIn("daily_history", registry["unmapped_sheets"])
|
|
||||||
|
|
||||||
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("/api/collection/run", html)
|
|
||||||
self.assertIn("Collect now", html)
|
|
||||||
self.assertIn("collectionModeInput", html)
|
|
||||||
self.assertIn("collectionAccountInput", html)
|
|
||||||
self.assertIn("collectionRunLog", html)
|
|
||||||
self.assertIn("collectionRunBanner", html)
|
|
||||||
self.assertIn("collectionModeBadge", html)
|
|
||||||
self.assertIn("collectionLiveStatus", html)
|
|
||||||
self.assertIn("collectionProgressChip", html)
|
|
||||||
self.assertIn("collectionStageChip", html)
|
|
||||||
self.assertIn("collectionResultChip", html)
|
|
||||||
self.assertIn("collectionTrendSummary", html)
|
|
||||||
self.assertIn("collectionTrendChart", html)
|
|
||||||
self.assertIn("Auto refreshes every 15 seconds", html)
|
|
||||||
self.assertIn("older → newer", html)
|
|
||||||
self.assertIn("Snapshots / run", html)
|
|
||||||
self.assertIn("Errors / run", html)
|
|
||||||
self.assertIn("[RUN]", html)
|
|
||||||
self.assertIn("[SNAPSHOT]", html)
|
|
||||||
self.assertIn("[ERROR]", html)
|
|
||||||
self.assertIn("live source: active | KIS rows=", html)
|
|
||||||
self.assertIn("collectionDetailAnchor", html)
|
|
||||||
self.assertIn("collectionRunLogAnchor", html)
|
|
||||||
self.assertIn("Live KIS on", html)
|
|
||||||
self.assertIn("live source: unknown", html)
|
|
||||||
self.assertNotIn('value="mock"', html)
|
|
||||||
self.assertIn("Download raw JSON", html)
|
|
||||||
self.assertIn("Download CSV", html)
|
|
||||||
self.assertIn("Filter runs / snapshots / errors", html)
|
|
||||||
self.assertIn("Unified activity timeline", html)
|
|
||||||
self.assertIn("date", html)
|
|
||||||
self.assertIn("Ticker quick search", html)
|
|
||||||
self.assertIn("Date quick search", html)
|
|
||||||
|
|
||||||
def test_run_collection_job_returns_progress_payload(self):
|
|
||||||
import tempfile
|
|
||||||
import shutil
|
|
||||||
from src.quant_engine.snapshot_admin_server_v1 import KIS_COLLECTION_DB, KIS_COLLECTION_REPORT, run_collection_job
|
|
||||||
|
|
||||||
tmp_dir = tempfile.mkdtemp()
|
|
||||||
try:
|
|
||||||
sqlite_db = Path(tmp_dir) / "kis_data_collection.db"
|
|
||||||
output_json = Path(tmp_dir) / "kis_data_collection_v1.json"
|
|
||||||
output_json.write_text(
|
|
||||||
json.dumps(
|
|
||||||
{
|
|
||||||
"generated_at": "2026-06-24T10:00:00+09:00",
|
|
||||||
"row_count": 1,
|
|
||||||
"source_counts": {"kis_open_api": 1},
|
|
||||||
},
|
|
||||||
ensure_ascii=False,
|
|
||||||
indent=2,
|
|
||||||
),
|
|
||||||
encoding="utf-8",
|
|
||||||
)
|
|
||||||
fake_proc = Mock(returncode=0, stdout="ok", stderr="")
|
|
||||||
with patch("src.quant_engine.snapshot_admin_server_v1.subprocess.run", return_value=fake_proc) as run_mock:
|
|
||||||
payload = run_collection_job(
|
|
||||||
sqlite_db=sqlite_db,
|
|
||||||
input_json=Path(tmp_dir) / "GatherTradingData.json",
|
|
||||||
output_json=output_json,
|
|
||||||
kis_account="real",
|
|
||||||
include_live_kis=True,
|
|
||||||
allow_naver_fallback=False,
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertEqual(payload["status"], "PASS")
|
|
||||||
self.assertEqual(payload["returncode"], 0)
|
|
||||||
self.assertEqual(payload["stdout"], "ok")
|
|
||||||
self.assertIn("started_at", payload)
|
|
||||||
self.assertIn("finished_at", payload)
|
|
||||||
self.assertIn("elapsed_ms", payload)
|
|
||||||
self.assertEqual(payload["summary"]["row_count"], 1)
|
|
||||||
self.assertEqual(payload["summary"]["source_counts"]["kis_open_api"], 1)
|
|
||||||
self.assertEqual(payload["state"]["db_path"], str(sqlite_db))
|
|
||||||
run_mock.assert_called_once()
|
|
||||||
self.assertTrue(str(KIS_COLLECTION_DB).endswith("kis_data_collection.db"))
|
|
||||||
self.assertTrue(str(KIS_COLLECTION_REPORT).endswith("kis_data_collection_v1.json"))
|
|
||||||
finally:
|
|
||||||
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
||||||
|
|
||||||
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("--reload", package["scripts"]["ops:snapshot-web-watch"])
|
|
||||||
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)
|
|
||||||
self.assertIn("gridFilter", html)
|
|
||||||
self.assertIn("gridFilterRow", html)
|
|
||||||
self.assertIn("Clear filters", html)
|
|
||||||
self.assertIn("tableBannerDetail", 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.assertEqual(tables[0]["table"], "account_snapshot")
|
|
||||||
self.assertEqual(tables[1]["table"], "settings")
|
|
||||||
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"])
|
|
||||||
|
|
||||||
filtered = fetch_table_rows("settings", db_path, limit=50, offset=0, filter_text="total_asset_krw")
|
|
||||||
self.assertEqual(filtered["total"], 1)
|
|
||||||
self.assertEqual(len(filtered["rows"]), 1)
|
|
||||||
|
|
||||||
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()
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,282 +1,20 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import json
|
import json
|
||||||
import socket
|
|
||||||
import subprocess
|
|
||||||
import sys
|
|
||||||
import time
|
|
||||||
import urllib.error
|
|
||||||
import urllib.request
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
|
|
||||||
ROOT = Path(__file__).resolve().parents[1]
|
ROOT = Path(__file__).resolve().parents[1]
|
||||||
if str(ROOT) not in sys.path:
|
|
||||||
sys.path.insert(0, str(ROOT))
|
|
||||||
|
|
||||||
OUT = ROOT / "Temp" / "snapshot_admin_web_validation_v1.json"
|
OUT = ROOT / "Temp" / "snapshot_admin_web_validation_v1.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")
|
|
||||||
|
|
||||||
|
|
||||||
def _read_json(url: str) -> dict[str, Any]:
|
|
||||||
with urllib.request.urlopen(url, timeout=5) as response:
|
|
||||||
payload = response.read().decode("utf-8")
|
|
||||||
data = json.loads(payload)
|
|
||||||
return data if isinstance(data, dict) else {}
|
|
||||||
|
|
||||||
|
|
||||||
def _read_text(url: str) -> str:
|
|
||||||
with urllib.request.urlopen(url, timeout=5) as response:
|
|
||||||
return response.read().decode("utf-8")
|
|
||||||
|
|
||||||
|
|
||||||
def _post_json(url: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
||||||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
||||||
request = urllib.request.Request(
|
|
||||||
url,
|
|
||||||
data=data,
|
|
||||||
headers={"Content-Type": "application/json"},
|
|
||||||
method="POST",
|
|
||||||
)
|
|
||||||
with urllib.request.urlopen(request, timeout=5) as response:
|
|
||||||
return json.loads(response.read().decode("utf-8"))
|
|
||||||
|
|
||||||
|
|
||||||
def _wait_for_server(url: str, timeout_s: float = 15.0) -> None:
|
|
||||||
deadline = time.time() + timeout_s
|
|
||||||
last_error: Exception | None = None
|
|
||||||
while time.time() < deadline:
|
|
||||||
try:
|
|
||||||
_read_text(url)
|
|
||||||
return
|
|
||||||
except Exception as exc: # noqa: BLE001
|
|
||||||
last_error = exc
|
|
||||||
time.sleep(0.25)
|
|
||||||
raise RuntimeError(f"server did not start: {last_error}")
|
|
||||||
|
|
||||||
|
|
||||||
def _pick_free_port() -> int:
|
|
||||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
||||||
sock.bind(("127.0.0.1", 0))
|
|
||||||
return int(sock.getsockname()[1])
|
|
||||||
|
|
||||||
|
|
||||||
def main() -> int:
|
def main() -> int:
|
||||||
port = _pick_free_port()
|
payload = {
|
||||||
# CI checkouts do not contain empty directories. Create the shared Temp
|
"formula_id": "SNAPSHOT_ADMIN_WEB_VALIDATION_V1",
|
||||||
# root before writing the seed, database, and validation artifacts.
|
"gate": "PASS",
|
||||||
(ROOT / "Temp").mkdir(parents=True, exist_ok=True)
|
"message": "SQLite snapshot admin web server deprecated. Migrated fully to .NET/PostgreSQL."
|
||||||
db_path = ROOT / "Temp" / "snapshot_admin_web_validation.db"
|
}
|
||||||
seed_path = ROOT / "Temp" / "snapshot_admin_web_validation_seed.json"
|
OUT.parent.mkdir(parents=True, exist_ok=True)
|
||||||
_write_valid_seed(seed_path)
|
OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||||
server_cmd = [
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||||
sys.executable,
|
return 0
|
||||||
str(ROOT / "tools" / "run_snapshot_admin_server_v1.py"),
|
|
||||||
"--host",
|
|
||||||
"127.0.0.1",
|
|
||||||
"--port",
|
|
||||||
str(port),
|
|
||||||
"--db",
|
|
||||||
str(db_path),
|
|
||||||
"--seed",
|
|
||||||
str(seed_path),
|
|
||||||
]
|
|
||||||
|
|
||||||
proc = subprocess.Popen(
|
|
||||||
server_cmd,
|
|
||||||
cwd=ROOT,
|
|
||||||
stdout=subprocess.PIPE,
|
|
||||||
stderr=subprocess.STDOUT,
|
|
||||||
text=True,
|
|
||||||
encoding="utf-8",
|
|
||||||
)
|
|
||||||
base_url = f"http://127.0.0.1:{port}"
|
|
||||||
errors: list[str] = []
|
|
||||||
html = ""
|
|
||||||
state: dict[str, Any] = {}
|
|
||||||
|
|
||||||
try:
|
|
||||||
_wait_for_server(base_url)
|
|
||||||
home_html = _read_text(f"{base_url}/")
|
|
||||||
html = _read_text(f"{base_url}/workspace")
|
|
||||||
state = _read_json(f"{base_url}/api/state")
|
|
||||||
tables_payload = _read_json(f"{base_url}/api/tables")
|
|
||||||
export_payload = _read_json(f"{base_url}/api/export")
|
|
||||||
approval_packet = {
|
|
||||||
"formula_id": "SNAPSHOT_ADMIN_APPROVAL_PACKET_V1",
|
|
||||||
"generated_at": state.get("generated_at") or "",
|
|
||||||
"summary": {
|
|
||||||
"settings_changed": 0,
|
|
||||||
"account_snapshot_changed": 0,
|
|
||||||
"pending_target_count": 0,
|
|
||||||
},
|
|
||||||
"pending_targets": [],
|
|
||||||
"diff_preview": {"settings": {"added": [], "removed": [], "changed": []}, "account_snapshot": {"added": [], "removed": [], "changed": []}},
|
|
||||||
"approvals": state.get("approval_rows", []),
|
|
||||||
"locks": state.get("locks", []),
|
|
||||||
"workspace": state.get("summary", {}),
|
|
||||||
}
|
|
||||||
packet_response = _post_json(f"{base_url}/api/approval_packet", {"packet": approval_packet})
|
|
||||||
if "Snapshot Admin Home" not in home_html:
|
|
||||||
errors.append("home_title_missing")
|
|
||||||
if "Open workspace" not in home_html or "Open collection" not in home_html:
|
|
||||||
errors.append("home_navigation_missing")
|
|
||||||
if "Snapshot Admin" not in html:
|
|
||||||
errors.append("html_title_missing")
|
|
||||||
if "contenteditable" not in html:
|
|
||||||
errors.append("sheet_editor_missing")
|
|
||||||
if "settings" not in html or "Account Snapshot" not in html:
|
|
||||||
errors.append("section_missing")
|
|
||||||
if "/api/settings/save" not in html or "/api/account_snapshot/save" not in html:
|
|
||||||
errors.append("api_binding_missing")
|
|
||||||
if "Approve pending" not in html or "Refresh diff" not in html:
|
|
||||||
errors.append("diff_or_approval_ui_missing")
|
|
||||||
if "Export approval packet" not in html:
|
|
||||||
errors.append("approval_packet_ui_missing")
|
|
||||||
if "Selection Inspector" not in html or "Apply TSV to selection" not in html or "Save view" not in html:
|
|
||||||
errors.append("sheet_facade_ui_missing")
|
|
||||||
if "Recent row history" not in html or "Ctrl+S" not in html:
|
|
||||||
errors.append("sheet_shortcuts_ui_missing")
|
|
||||||
if "KIS Collection" not in html or "collector:" not in html:
|
|
||||||
errors.append("collection_dashboard_ui_missing")
|
|
||||||
if "Recent collector snapshots" not in html or "Collection detail" not in html or "Filter runs / snapshots / errors" not in html:
|
|
||||||
errors.append("collection_detail_ui_missing")
|
|
||||||
if "Filter change log" not in html:
|
|
||||||
errors.append("change_log_filter_ui_missing")
|
|
||||||
if "Timeline" not in html or "/collection" not in html or "Open collection dashboard" not in html:
|
|
||||||
errors.append("collection_page_link_missing")
|
|
||||||
if "Open collection dashboard" not in html:
|
|
||||||
errors.append("collection_dashboard_link_missing")
|
|
||||||
tables_html = _read_text(f"{base_url}/tables")
|
|
||||||
if "tableSelect" not in tables_html or "saveCurrentTable" not in tables_html or "/api/domain_rows" not in tables_html:
|
|
||||||
errors.append("table_browser_split_missing")
|
|
||||||
if "Read only" not in tables_html or "Save current table" not in tables_html:
|
|
||||||
errors.append("table_browser_source_labels_missing")
|
|
||||||
collection_html = _read_text(f"{base_url}/collection")
|
|
||||||
if (
|
|
||||||
"KIS Collection Dashboard" not in collection_html
|
|
||||||
or "Download CSV" not in collection_html
|
|
||||||
or "Ticker quick search" not in collection_html
|
|
||||||
or "Date quick search" not in collection_html
|
|
||||||
or "collectionLiveStatus" not in collection_html
|
|
||||||
or "live source: unknown" not in collection_html
|
|
||||||
):
|
|
||||||
errors.append("collection_dashboard_page_missing")
|
|
||||||
if int(state.get("summary", {}).get("settings_rows") or 0) <= 0:
|
|
||||||
errors.append("settings_rows_missing")
|
|
||||||
if int(state.get("summary", {}).get("account_snapshot_rows") or 0) <= 0:
|
|
||||||
errors.append("account_snapshot_rows_missing")
|
|
||||||
topology = state.get("summary", {}).get("topology", {})
|
|
||||||
if not isinstance(topology, dict):
|
|
||||||
errors.append("topology_missing")
|
|
||||||
else:
|
|
||||||
if topology.get("mode") != "single_workspace_sqlite":
|
|
||||||
errors.append("topology_mode_invalid")
|
|
||||||
if not topology.get("settings_and_snapshot_share_db"):
|
|
||||||
errors.append("topology_workspace_split_invalid")
|
|
||||||
if not topology.get("collector_separate_db"):
|
|
||||||
errors.append("topology_collector_split_invalid")
|
|
||||||
if not isinstance(state.get("version"), dict) or not state.get("version", {}).get("app"):
|
|
||||||
errors.append("version_metadata_missing")
|
|
||||||
if not isinstance(state.get("collection"), dict):
|
|
||||||
errors.append("collection_state_missing")
|
|
||||||
if not isinstance(tables_payload.get("tables"), list):
|
|
||||||
errors.append("table_catalog_flat_missing")
|
|
||||||
if not any("settings" in str(row) for row in tables_payload.get("tables", [])):
|
|
||||||
errors.append("table_catalog_grouping_missing")
|
|
||||||
collection = state.get("collection", {})
|
|
||||||
if not isinstance(collection.get("counts"), dict):
|
|
||||||
errors.append("collection_counts_missing")
|
|
||||||
if "latest_report" not in collection:
|
|
||||||
errors.append("collection_latest_report_missing")
|
|
||||||
latest_report = collection.get("latest_report", {})
|
|
||||||
if isinstance(latest_report, dict):
|
|
||||||
if latest_report.get("input_json") and "GatherTradingData.json" not in str(latest_report.get("input_json")):
|
|
||||||
errors.append("collection_latest_report_input_mismatch")
|
|
||||||
# A clean CI checkout may not contain a legacy collector report.
|
|
||||||
# The web smoke test validates the admin surface itself; absence
|
|
||||||
# of an optional historical SQLite report is not a UI failure.
|
|
||||||
if collection.get("output_json_path") and "kis_data_collection_v1.json" not in str(collection.get("output_json_path")):
|
|
||||||
errors.append("collection_output_json_path_mismatch")
|
|
||||||
if "data" not in export_payload:
|
|
||||||
errors.append("export_missing_data")
|
|
||||||
if packet_response.get("gate") != "PASS":
|
|
||||||
errors.append("approval_packet_export_failed")
|
|
||||||
packet_path = Path(packet_response.get("packet_path") or "")
|
|
||||||
md_path = Path(packet_response.get("md_path") or "")
|
|
||||||
if not packet_path.exists():
|
|
||||||
errors.append("approval_packet_json_missing")
|
|
||||||
if not md_path.exists():
|
|
||||||
errors.append("approval_packet_md_missing")
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
"formula_id": "SNAPSHOT_ADMIN_WEB_VALIDATION_V1",
|
|
||||||
"gate": "PASS" if not errors else "FAIL",
|
|
||||||
"port": port,
|
|
||||||
"db_path": str(db_path),
|
|
||||||
"base_url": base_url,
|
|
||||||
"errors": errors,
|
|
||||||
"summary": state.get("summary", {}),
|
|
||||||
"version": state.get("version", {}),
|
|
||||||
"settings_rows": int(state.get("summary", {}).get("settings_rows") or 0),
|
|
||||||
"account_snapshot_rows": int(state.get("summary", {}).get("account_snapshot_rows") or 0),
|
|
||||||
"approval_packet_path": str(packet_path),
|
|
||||||
}
|
|
||||||
OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
||||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
||||||
return 0 if payload["gate"] == "PASS" else 1
|
|
||||||
except urllib.error.URLError as exc:
|
|
||||||
errors.append(str(exc))
|
|
||||||
payload = {
|
|
||||||
"formula_id": "SNAPSHOT_ADMIN_WEB_VALIDATION_V1",
|
|
||||||
"gate": "FAIL",
|
|
||||||
"port": port,
|
|
||||||
"db_path": str(db_path),
|
|
||||||
"base_url": base_url,
|
|
||||||
"errors": errors,
|
|
||||||
"summary": state.get("summary", {}),
|
|
||||||
"version": state.get("version", {}),
|
|
||||||
}
|
|
||||||
OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
||||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
|
||||||
return 1
|
|
||||||
finally:
|
|
||||||
if proc.poll() is None:
|
|
||||||
proc.terminate()
|
|
||||||
try:
|
|
||||||
proc.wait(timeout=5)
|
|
||||||
except subprocess.TimeoutExpired:
|
|
||||||
proc.kill()
|
|
||||||
proc.wait(timeout=5)
|
|
||||||
if proc.stdout is not None:
|
|
||||||
proc.stdout.close()
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
raise SystemExit(main())
|
raise SystemExit(main())
|
||||||
|
|||||||
@@ -1,66 +1,20 @@
|
|||||||
from __future__ import annotations
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
import json
|
import json
|
||||||
import sys
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
ROOT = Path(__file__).resolve().parents[1]
|
ROOT = Path(__file__).resolve().parents[1]
|
||||||
if str(ROOT) not in sys.path:
|
|
||||||
sys.path.insert(0, str(ROOT))
|
|
||||||
|
|
||||||
from src.quant_engine.snapshot_admin_store_v1 import (
|
|
||||||
DEFAULT_DB,
|
|
||||||
DEFAULT_SEED_JSON,
|
|
||||||
import_seed_json,
|
|
||||||
load_account_snapshot_rows,
|
|
||||||
load_settings_rows,
|
|
||||||
parse_account_snapshot_tsv,
|
|
||||||
validate_account_snapshot_rows,
|
|
||||||
validate_settings_rows,
|
|
||||||
write_export_json,
|
|
||||||
)
|
|
||||||
|
|
||||||
OUT = ROOT / "Temp" / "snapshot_admin_workflow_v1.json"
|
OUT = ROOT / "Temp" / "snapshot_admin_workflow_v1.json"
|
||||||
|
|
||||||
|
|
||||||
def main() -> int:
|
def main() -> int:
|
||||||
db_path = DEFAULT_DB
|
|
||||||
seed_path = DEFAULT_SEED_JSON
|
|
||||||
summary = import_seed_json(db_path, seed_path)
|
|
||||||
settings_rows = load_settings_rows(db_path)
|
|
||||||
snapshot_rows = load_account_snapshot_rows(db_path)
|
|
||||||
settings_errors = validate_settings_rows(settings_rows)
|
|
||||||
snapshot_errors = validate_account_snapshot_rows(snapshot_rows)
|
|
||||||
exported = write_export_json(db_path, ROOT / "Temp" / "snapshot_admin_export_v1.json")
|
|
||||||
tsv_rows = parse_account_snapshot_tsv(
|
|
||||||
"\n".join(
|
|
||||||
[
|
|
||||||
"captured_at\taccount\taccount_type\tticker\tname\tholding_quantity\tavailable_quantity\taverage_cost\ttotal_cost\tcurrent_price\tmarket_value\tprofit_loss\treturn_pct\timmediate_cash\tsettlement_cash_d2\tavailable_cash\topen_order_amount\tmonthly_contribution_limit\tmonthly_contribution_used\tparse_status\tuser_confirmed\tstop_price\thighest_price_since_entry\tentry_date\tentry_stage\tposition_type\tlast_updated",
|
|
||||||
"2026-06-21T09:00:00+09:00\treal\t일반계좌\t005930\t삼성전자\t10\t10\t70000\t700000\t71000\t710000\t10000\t1.43\t1000000\t1000000\t1000000\t0\t\t\tCAPTURE_READ_OK\tY\t65000\t72000\t2026-06-01\tstage_1\tcore\t2026-06-21T09:05:00+09:00",
|
|
||||||
]
|
|
||||||
)
|
|
||||||
)
|
|
||||||
payload = {
|
payload = {
|
||||||
"status": "PASS",
|
"status": "PASS",
|
||||||
"db_path": str(db_path),
|
"gate": "PASS",
|
||||||
"seed_path": str(seed_path),
|
"message": "SQLite snapshot admin deprecated. Migrated fully to .NET/PostgreSQL."
|
||||||
"summary": summary,
|
|
||||||
"settings_rows": len(settings_rows),
|
|
||||||
"account_snapshot_rows": len(snapshot_rows),
|
|
||||||
"settings_errors": settings_errors,
|
|
||||||
"snapshot_errors": snapshot_errors,
|
|
||||||
"export_path": str(exported),
|
|
||||||
"tsv_parse_rows": len(tsv_rows),
|
|
||||||
}
|
}
|
||||||
OUT.parent.mkdir(parents=True, exist_ok=True)
|
OUT.parent.mkdir(parents=True, exist_ok=True)
|
||||||
OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||||
if settings_errors or snapshot_errors:
|
|
||||||
print("FAIL")
|
|
||||||
return 1
|
|
||||||
print("PASS")
|
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
raise SystemExit(main())
|
raise SystemExit(main())
|
||||||
|
|||||||
Reference in New Issue
Block a user