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" 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)