arc/tests/mock_server/gen_mock_data.py

#!/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")
    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 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}")

        # Also write a fetch_transactions scenario (empty body, for manual use)
        ft_fname = f"{dev_short}_{bank}_fetch_transactions.json"
        ft_task = {
            "bank_name": bank,
            "material_id": row["material_id"],
            "type": "FETCH_TRANSACTIONS",
            "body": {},
        }
        (scenarios_dir / ft_fname).write_text(
            json.dumps(ft_task, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
        )

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

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