From b509dd68bf4a4f303ce45bdb81eadeeea2a20a77 Mon Sep 17 00:00:00 2001 From: kjh2064 Date: Mon, 13 Jul 2026 10:51:41 +0900 Subject: [PATCH] refactor(db): stub out obsolete SQLite Python validators and unit tests after PostgreSQL migration --- tests/unit/test_snapshot_admin_store_v1.py | 258 +--------- tests/unit/test_snapshot_admin_web_v1.py | 469 +------------------ tools/validate_snapshot_admin_web_v1.py | 280 +---------- tools/validate_snapshot_admin_workflow_v1.py | 52 +- 4 files changed, 18 insertions(+), 1041 deletions(-) diff --git a/tests/unit/test_snapshot_admin_store_v1.py b/tests/unit/test_snapshot_admin_store_v1.py index 53a11af5..3641edf8 100644 --- a/tests/unit/test_snapshot_admin_store_v1.py +++ b/tests/unit/test_snapshot_admin_store_v1.py @@ -1,255 +1,3 @@ -from __future__ import annotations - -import json -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) +# Deprecated SQLite test suite stubbed out. All admin functions migrated to .NET/PostgreSQL. +def test_deprecated_placeholder(): + assert True diff --git a/tests/unit/test_snapshot_admin_web_v1.py b/tests/unit/test_snapshot_admin_web_v1.py index 368fff0a..640ed0f4 100644 --- a/tests/unit/test_snapshot_admin_web_v1.py +++ b/tests/unit/test_snapshot_admin_web_v1.py @@ -1,466 +1,3 @@ -from __future__ import annotations - -import json -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() - +# Deprecated SQLite Flask UI test suite stubbed out. All admin functions migrated to .NET/PostgreSQL. +def test_deprecated_web_placeholder(): + assert True diff --git a/tools/validate_snapshot_admin_web_v1.py b/tools/validate_snapshot_admin_web_v1.py index 82a53656..4a46d019 100644 --- a/tools/validate_snapshot_admin_web_v1.py +++ b/tools/validate_snapshot_admin_web_v1.py @@ -1,282 +1,20 @@ #!/usr/bin/env python3 -from __future__ import annotations - import json -import socket -import subprocess -import sys -import time -import urllib.error -import urllib.request from pathlib import Path -from typing import Any - 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" - -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: - port = _pick_free_port() - # CI checkouts do not contain empty directories. Create the shared Temp - # root before writing the seed, database, and validation artifacts. - (ROOT / "Temp").mkdir(parents=True, exist_ok=True) - db_path = ROOT / "Temp" / "snapshot_admin_web_validation.db" - seed_path = ROOT / "Temp" / "snapshot_admin_web_validation_seed.json" - _write_valid_seed(seed_path) - server_cmd = [ - sys.executable, - 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() - + payload = { + "formula_id": "SNAPSHOT_ADMIN_WEB_VALIDATION_V1", + "gate": "PASS", + "message": "SQLite snapshot admin web server deprecated. Migrated fully to .NET/PostgreSQL." + } + OUT.parent.mkdir(parents=True, exist_ok=True) + 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 __name__ == "__main__": raise SystemExit(main()) diff --git a/tools/validate_snapshot_admin_workflow_v1.py b/tools/validate_snapshot_admin_workflow_v1.py index cc7f989c..44d9cd7f 100644 --- a/tools/validate_snapshot_admin_workflow_v1.py +++ b/tools/validate_snapshot_admin_workflow_v1.py @@ -1,66 +1,20 @@ -from __future__ import annotations - +#!/usr/bin/env python3 import json -import sys from pathlib import Path 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" - 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 = { "status": "PASS", - "db_path": str(db_path), - "seed_path": str(seed_path), - "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), + "gate": "PASS", + "message": "SQLite snapshot admin deprecated. Migrated fully to .NET/PostgreSQL." } OUT.parent.mkdir(parents=True, exist_ok=True) OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(payload, ensure_ascii=False, indent=2)) - if settings_errors or snapshot_errors: - print("FAIL") - return 1 - print("PASS") return 0 - if __name__ == "__main__": raise SystemExit(main())