From 06aed3966128f50568ef58145e4af2bc18df36ec Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Mon, 11 May 2026 09:53:49 +0800 Subject: [PATCH] test(db): add schema migration tests Co-authored-by: Cursor --- tests/test_schema_migration.py | 376 +++++++++++++++++++++++++++++++++ 1 file changed, 376 insertions(+) create mode 100644 tests/test_schema_migration.py diff --git a/tests/test_schema_migration.py b/tests/test_schema_migration.py new file mode 100644 index 0000000..26a55be --- /dev/null +++ b/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()