# -*- coding: utf-8 -*-
"""Tests for v2 schema DDL and migrate_v1_to_v2 / ensure_schema."""
import os
import sqlite3
import sys
import tempfile
import time

import pytest

_scripts = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts")
if _scripts not in sys.path:
    sys.path.insert(0, _scripts)

from util.constants import SKILL_SLUG  # noqa: E402


# Minimal v1-like DDL (no auth_strategy / session_persistent / device_fingerprint /
# platforms.default_auth_strategy; credentials without account_id).
_V1_DDL = """
CREATE TABLE platforms (
    platform_key TEXT PRIMARY KEY,
    display_name TEXT NOT NULL,
    domain TEXT NOT NULL,
    provider_code TEXT,
    default_url TEXT,
    aliases_json TEXT NOT NULL DEFAULT '[]',
    capabilities_json TEXT NOT NULL DEFAULT '{}',
    enabled INTEGER NOT NULL DEFAULT 1,
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL
);
CREATE TABLE accounts (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    platform_key TEXT NOT NULL,
    account_label TEXT NOT NULL,
    login_id TEXT,
    login_id_type TEXT NOT NULL DEFAULT 'unknown',
    tenant_id TEXT,
    environment TEXT NOT NULL DEFAULT 'production',
    role TEXT NOT NULL DEFAULT 'default',
    provider_code TEXT,
    profile_dir TEXT,
    url TEXT,
    status TEXT NOT NULL DEFAULT 'active',
    session_status TEXT NOT NULL DEFAULT 'unknown',
    last_used_at INTEGER,
    last_login_check_at INTEGER,
    last_error_code TEXT,
    last_error_message TEXT,
    extra_json TEXT NOT NULL DEFAULT '{}',
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL
);
CREATE TABLE credentials (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    platform_key TEXT NOT NULL,
    credential_label TEXT NOT NULL,
    credential_type TEXT NOT NULL,
    secret_storage TEXT NOT NULL,
    secret_ref TEXT,
    secret_mask TEXT,
    expires_at INTEGER,
    status TEXT NOT NULL DEFAULT 'active',
    last_used_at INTEGER,
    extra_json TEXT NOT NULL DEFAULT '{}',
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL
);
CREATE TABLE account_leases (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    account_id INTEGER NOT NULL,
    lease_token TEXT NOT NULL UNIQUE,
    holder TEXT NOT NULL,
    purpose TEXT,
    expires_at INTEGER NOT NULL,
    heartbeat_at INTEGER,
    released_at INTEGER,
    status TEXT NOT NULL DEFAULT 'active',
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL
);
"""


def _db_path_for_env(root: str, uid: str) -> str:
    return os.path.join(root, uid, SKILL_SLUG, "account-manager.db")


def _prep_logging():
    from util.logging_config import setup_skill_logging

    setup_skill_logging(SKILL_SLUG, "openclaw.skill.account_manager_migration_test")


def _column_names(conn: sqlite3.Connection, table: str) -> list[str]:
    cur = conn.cursor()
    cur.execute(f"PRAGMA table_info({table})")
    return [row[1] for row in cur.fetchall()]


@pytest.fixture()
def iso_env(monkeypatch):
    with tempfile.TemporaryDirectory(prefix="acc_mgr_schema_", ignore_cleanup_errors=True) as tmp:
        data_root = os.path.join(tmp, "data-root")
        os.makedirs(data_root)
        uid = "migration_uid"
        monkeypatch.setenv("JIANGCHANG_DATA_ROOT", data_root)
        monkeypatch.setenv("JIANGCHANG_USER_ID", uid)
        db_path = _db_path_for_env(data_root, uid)
        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        _prep_logging()
        yield {"tmp": tmp, "data_root": data_root, "uid": uid, "db_path": db_path}


class TestFreshSchema:
    def test_fresh_schema_has_v2_columns_and_meta(self, iso_env):
        from db.connection import init_db, get_conn

        if os.path.isfile(iso_env["db_path"]):
            os.remove(iso_env["db_path"])
        init_db()
        conn = get_conn()
        try:
            cols_accounts = _column_names(conn, "accounts")
            assert "auth_strategy" in cols_accounts
            assert "session_persistent" in cols_accounts
            assert "device_fingerprint" in cols_accounts
            cols_plat = _column_names(conn, "platforms")
            assert "default_auth_strategy" in cols_plat
            cols_cred = _column_names(conn, "credentials")
            assert "account_id" in cols_cred
            cur = conn.cursor()
            cur.execute(
                "SELECT value FROM _schema_meta WHERE key = ?",
                ("schema_version",),
            )
            row = cur.fetchone()
            assert row is not None and row[0] == "2"
        finally:
            conn.close()


class TestMigrateFromV1:
    def test_v1_to_v2_migrates_and_preserves_rows(self, iso_env, capsys):
        db_path = iso_env["db_path"]
        if os.path.isfile(db_path):
            os.remove(db_path)
        os.makedirs(os.path.dirname(db_path), exist_ok=True)

        raw = sqlite3.connect(db_path)
        try:
            raw.executescript(_V1_DDL)
            now = int(time.time())
            raw.execute(
                """
                INSERT INTO platforms (platform_key, display_name, domain, provider_code,
                    default_url, aliases_json, capabilities_json, enabled, created_at, updated_at)
                VALUES ('keep_me', 'Keep', 'generic', NULL, '', '[]', '{}', 1, ?, ?)
                """,
                (now, now),
            )
            raw.execute(
                """
                INSERT INTO accounts (platform_key, account_label, login_id, login_id_type,
                    tenant_id, environment, role, provider_code, profile_dir, url,
                    status, session_status, extra_json, created_at, updated_at)
                VALUES ('keep_me', 'acc', 'u@test.com', 'email', NULL, 'production',
                    'booking', NULL, NULL, '', 'active', 'unknown', '{}', ?, ?)
                """,
                (now, now),
            )
            raw.commit()
        finally:
            raw.close()

        from db.connection import init_db, get_conn

        init_db()
        err = capsys.readouterr().err
        assert "[migration] account-manager v1→v2:" in err

        conn = get_conn()
        try:
            cur = conn.cursor()
            cur.execute("SELECT COUNT(*) FROM accounts WHERE login_id = ?", ("u@test.com",))
            assert int(cur.fetchone()[0]) == 1
            cur.execute(
                "SELECT auth_strategy, session_persistent FROM accounts WHERE login_id = ?",
                ("u@test.com",),
            )
            strat, sp = cur.fetchone()
            assert strat == "password_auto"
            assert int(sp) == 1
        finally:
            conn.close()

    def test_migration_idempotent(self, iso_env, capsys):
        db_path = iso_env["db_path"]
        if os.path.isfile(db_path):
            os.remove(db_path)
        os.makedirs(os.path.dirname(db_path), exist_ok=True)

        raw = sqlite3.connect(db_path)
        try:
            raw.executescript(_V1_DDL)
            raw.commit()
        finally:
            raw.close()

        from db.connection import init_db

        init_db()
        capsys.readouterr()
        init_db()
        err = capsys.readouterr().err
        assert "v1→v2" not in err

        conn = sqlite3.connect(db_path)
        try:
            cols = _column_names(conn, "accounts")
            assert cols.count("auth_strategy") == 1
        finally:
            conn.close()

    def test_platform_default_mapping_douyin(self, iso_env):
        db_path = iso_env["db_path"]
        if os.path.isfile(db_path):
            os.remove(db_path)
        os.makedirs(os.path.dirname(db_path), exist_ok=True)

        raw = sqlite3.connect(db_path)
        now = int(time.time())
        try:
            raw.executescript(_V1_DDL)
            raw.execute(
                """
                INSERT INTO platforms (platform_key, display_name, domain, provider_code,
                    default_url, aliases_json, capabilities_json, enabled, created_at, updated_at)
                VALUES ('douyin', 'Douyin', 'content', NULL, '', '[]', '{}', 1, ?, ?)
                """,
                (now, now),
            )
            raw.commit()
        finally:
            raw.close()

        from db.connection import init_db, get_conn

        init_db()
        conn = get_conn()
        try:
            cur = conn.cursor()
            cur.execute(
                "SELECT default_auth_strategy FROM platforms WHERE platform_key = ?",
                ("douyin",),
            )
            assert cur.fetchone()[0] == "qr_code_manual"
        finally:
            conn.close()

    def test_accounts_auth_strategy_priority(self, iso_env):
        db_path = iso_env["db_path"]
        if os.path.isfile(db_path):
            os.remove(db_path)
        os.makedirs(os.path.dirname(db_path), exist_ok=True)

        raw = sqlite3.connect(db_path)
        now = int(time.time())
        try:
            raw.executescript(_V1_DDL)
            for pk, dn in (
                ("maersk", "Maersk"),
                ("unknown_platform", "Unknown"),
            ):
                raw.execute(
                    """
                    INSERT INTO platforms (platform_key, display_name, domain, provider_code,
                        default_url, aliases_json, capabilities_json, enabled, created_at, updated_at)
                    VALUES (?, ?, 'generic', NULL, '', '[]', '{}', 1, ?, ?)
                    """,
                    (pk, dn, now, now),
                )
            raw.execute(
                """
                INSERT INTO accounts (platform_key, account_label, login_id, login_id_type,
                    tenant_id, environment, role, provider_code, profile_dir, url,
                    status, session_status, extra_json, created_at, updated_at)
                VALUES ('maersk', 'm', NULL, 'unknown', NULL, 'production',
                    'booking', NULL, NULL, '', 'active', 'unknown', '{}', ?, ?)
                """,
                (now, now),
            )
            raw.execute(
                """
                INSERT INTO accounts (platform_key, account_label, login_id, login_id_type,
                    tenant_id, environment, role, provider_code, profile_dir, url,
                    status, session_status, extra_json, created_at, updated_at)
                VALUES ('unknown_platform', 'u', NULL, 'unknown', NULL, 'production',
                    'booking', NULL, NULL, '', 'active', 'unknown', '{}', ?, ?)
                """,
                (now, now),
            )
            raw.commit()
        finally:
            raw.close()

        from db.connection import init_db, get_conn

        init_db()
        conn = get_conn()
        try:
            cur = conn.cursor()
            cur.execute(
                "SELECT auth_strategy FROM accounts WHERE platform_key = ? ORDER BY id",
                ("maersk",),
            )
            assert cur.fetchone()[0] == "qr_code_manual"
            cur.execute(
                "SELECT auth_strategy FROM accounts WHERE platform_key = ? ORDER BY id",
                ("unknown_platform",),
            )
            assert cur.fetchone()[0] == "password_auto"
        finally:
            conn.close()


class TestInsertAccountValidation:
    def test_insert_account_rejects_invalid_strategy(self, iso_env):
        from db.connection import init_db, get_conn
        from db.accounts_repo import insert_account
        from db.auth_strategy import AUTH_STRATEGY_QR_CODE_MANUAL

        if os.path.isfile(iso_env["db_path"]):
            os.remove(iso_env["db_path"])
        init_db()
        conn = get_conn()
        try:
            now = int(time.time())
            with pytest.raises(ValueError, match="invalid auth_strategy"):
                insert_account(
                    conn,
                    platform_key="x",
                    account_label="l",
                    login_id=None,
                    login_id_type="unknown",
                    tenant_id=None,
                    environment="production",
                    role="booking",
                    provider_code=None,
                    profile_dir=None,
                    url=None,
                    extra_json="{}",
                    now=now,
                    auth_strategy="not_a_strategy",
                )
            aid = insert_account(
                conn,
                platform_key="x",
                account_label="l",
                login_id=None,
                login_id_type="unknown",
                tenant_id=None,
                environment="production",
                role="booking",
                provider_code=None,
                profile_dir=None,
                url=None,
                extra_json="{}",
                now=now,
                auth_strategy=AUTH_STRATEGY_QR_CODE_MANUAL,
            )
            assert isinstance(aid, int) and aid > 0
            conn.commit()
        finally:
            conn.close()