test(db): add schema migration tests

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_schema_migration.py

@@ -0,0 +1,376 @@
+# -*- 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()