#!/usr/bin/env python3 """ Generate mock test data from the local ARC database. Reads config.ini (from project root or --config) to find the DB path, queries active devices/profiles/materials, and generates: - default_profile.json (seeded into mock at startup) - scenarios/*_fetch_profile.json (one FETCH_PROFILE per material) - Cleans old scenarios that no longer match the DB Run from project root: python tests/mock_server/gen_mock_data.py Or specify config explicitly: python tests\\mock_server\\gen_mock_data.py --config C:\\path\\to\\config.ini On Windows: python tests\\mock_server\\gen_mock_data.py """ import argparse import configparser import json import os import re import sqlite3 import sys from pathlib import Path def resolve_db_path(config_path: Path) -> Path: config = configparser.ConfigParser() config.read(str(config_path), encoding="utf-8-sig") raw = config.get("db", "dbPath", fallback="") if not raw: print(f"ERROR: [db] dbPath not found in {config_path}", file=sys.stderr) sys.exit(1) # Expand ~ and make absolute relative to config dir expanded = os.path.expanduser(raw) p = Path(expanded) if not p.is_absolute(): p = config_path.parent / p return p.resolve() def find_config(hint: str | None) -> Path: if hint: p = Path(hint) if p.exists(): return p print(f"ERROR: config not found at {p}", file=sys.stderr) sys.exit(1) # Try common locations relative to CWD candidates = [ Path("config.ini"), Path("build/Release/config.ini"), Path("build/config.ini"), Path("build/Debug/config.ini"), ] for c in candidates: if c.exists(): return c.resolve() print("ERROR: config.ini not found. Run from project root or use --config.", file=sys.stderr) sys.exit(1) def query_active_materials(db_path: Path): """Returns list of dicts with device/profile/material info.""" if not db_path.exists(): print(f"ERROR: database not found at {db_path}", file=sys.stderr) sys.exit(1) conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row rows = conn.execute(""" SELECT d.id AS device_id, d.api_id AS device_api_id, d.name AS device_name, bp.code AS bank_name, bp.bank_profile_id, bp.full_name, bp.phone, bp.currency AS bp_currency, m.material_id, m.last_numbers, m.card_number, m.currency AS mat_currency FROM devices d JOIN bank_profiles bp ON bp.device_id = d.id JOIN materials m ON m.device_id = d.id AND m.app_code = bp.code WHERE bp.status = 'active' AND m.status = 'active' AND m.material_id != '' AND bp.bank_profile_id != '' ORDER BY d.name, bp.code """).fetchall() conn.close() return [dict(r) for r in rows] CURRENCY_MAP = { "RUB": 643, "USD": 840, "KRW": 410, "AZN": 944, "ARS": 32, "TJS": 972, "EUR": 978, "GBP": 826, "CNY": 156, "JPY": 392, "TRY": 949, } def device_label(name: str) -> str: """TECNO KM4-c7369f1959ffec79 → TECNO KM4""" m = re.match(r"^(.+?)-[a-f0-9]{10,}$", name) return m.group(1) if m else name def cleanup_events(db_path: Path): """Delete all events from previous runs.""" conn = sqlite3.connect(str(db_path)) count = conn.execute("SELECT count(*) FROM events").fetchone()[0] if count > 0: conn.execute("DELETE FROM events") conn.commit() print(f" deleted {count} old events") else: print(" events table clean") conn.close() def generate(materials: list, out_dir: Path): out_dir.mkdir(parents=True, exist_ok=True) scenarios_dir = out_dir / "scenarios" scenarios_dir.mkdir(exist_ok=True) # --- default_profile.json --- # Group by (device_api_id, bank_name) → bank_profile with materials bp_map: dict = {} for row in materials: key = (row["device_api_id"], row["bank_name"]) if key not in bp_map: bp_map[key] = { "id": row["bank_profile_id"], "bank_name": row["bank_name"], "state": "enabled", "phone_number": row["phone"] or "", "first_name": (row["full_name"] or "").split()[0] if row["full_name"] else "", "middle_name": None, "last_name": " ".join((row["full_name"] or "").split()[1:]) if row["full_name"] else "", "currency": CURRENCY_MAP.get(row["bp_currency"], 0), "device_id": row["device_api_id"], "device_label": device_label(row["device_name"]), "materials": [], } bp_map[key]["materials"].append({ "id": row["material_id"], "state": "enabled", "card_number": row["card_number"] or "", "currency": CURRENCY_MAP.get(row["mat_currency"], 0), }) profile = {"bank_profiles": list(bp_map.values())} profile_path = out_dir / "default_profile.json" profile_path.write_text(json.dumps(profile, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") print(f" wrote {profile_path.name} ({len(bp_map)} bank_profiles)") # --- FETCH_PROFILE scenarios --- # Remove old fetch_profile scenarios for old in scenarios_dir.glob("*_fetch_profile.json"): old.unlink() devices_seen = {} for row in materials: dev_short = device_label(row["device_name"]).lower().replace(" ", "_") bank = row["bank_name"] fname = f"{dev_short}_{bank}_fetch_profile.json" task = { "bank_name": bank, "material_id": row["material_id"], "type": "FETCH_PROFILE", "body": {}, } (scenarios_dir / fname).write_text( json.dumps(task, indent=2, ensure_ascii=False) + "\n", encoding="utf-8" ) print(f" wrote scenarios/{fname}") devices_seen.setdefault(dev_short, []).append( f"{bank} ({row['material_id'][:8]}…)" ) # Summary print(f"\nDevices ({len(devices_seen)}):") for dev, banks in devices_seen.items(): print(f" {dev}: {', '.join(banks)}") print(f"\nTotal: {len(materials)} active materials") print(f"Files in: {out_dir.resolve()}") def main(): parser = argparse.ArgumentParser( description="Generate mock test data from local ARC database" ) parser.add_argument("--config", "-c", default=None, help="Path to config.ini (default: auto-detect from CWD)") args = parser.parse_args() config_path = find_config(args.config) print(f"config: {config_path}") db_path = resolve_db_path(config_path) print(f"database: {db_path}") cleanup_events(db_path) materials = query_active_materials(db_path) if not materials: print("No active materials found in database.", file=sys.stderr) return 1 # Output dir = tests/mock_server (same dir as this script) out_dir = Path(__file__).parent generate(materials, out_dir) return 0 if __name__ == "__main__": sys.exit(main())