sql5 3.5.0

"""
sql5 Python 客戶端整合測試

本測試模組驗證 sql5 Python 客戶端能透過 JSON over stdin/stdout
與 Rust sql5 伺服器通訊。

架構：
  Python 客戶端 (sql5.connect())
       ↓
  subprocess.Popen(sql5 --server)
       ↓
  JSON over stdin/stdout
       ↓
  Rust sql5 伺服器
"""

import os
import sys
import tempfile
import pytest

# 取得測試腳本所在目錄
script_dir = os.path.dirname(os.path.abspath(__file__))
project_dir = os.path.dirname(script_dir)

# 測試前先找到本地的 sql5 二進位檔
def get_local_binary():
    """尋找本地的 sql5 二進位檔用於測試"""
    # 依序檢查可能的路徑
    paths_to_check = [
        os.path.join(project_dir, "target", "debug", "sql5"),
        os.path.join(project_dir, "..", "target", "debug", "sql5"),
        os.path.join(os.path.dirname(project_dir), "target", "debug", "sql5"),
    ]
    for p in paths_to_check:
        if os.path.exists(p):
            return p
    return None

# 匯入 sql5 前先設定 SQL5_BINARY
local_binary = get_local_binary()
if local_binary:
    os.environ["SQL5_BINARY"] = local_binary
    print(f"使用本地二進位檔：{local_binary}", file=sys.stderr)

# 將專案目錄加入 Python 路徑
sys.path.insert(0, project_dir)

import sql5
from sql5 import connect, Error


@pytest.fixture
def db():
    """測試用的資料庫連線 fixture"""
    connection = connect()  # 記憶體模式
    yield connection
    connection.close()


# ============================================================================
# 基本操作測試
# ============================================================================

class TestBasicOperations:
    """基本 SQL 操作測試"""

    def test_basic_select(self, db):
        """測試 1：基本 SELECT 查詢"""
        cursor = db.execute("SELECT 1 AS a, 2 AS b, 3 AS c")
        row = cursor.fetchone()
        assert row == [1, 2, 3]

    def test_select_multiple_rows(self, db):
        """測試：多列 UNION 查詢"""
        cursor = db.execute("SELECT 1 UNION SELECT 2 UNION SELECT 3")
        rows = cursor.fetchall()
        assert len(rows) == 3

    def test_select_with_alias(self, db):
        """測試：帶欄位別名的 SELECT"""
        cursor = db.execute("SELECT 42 AS answer, 'hello' AS greeting")
        row = cursor.fetchone()
        assert row == [42, "hello"]


# ============================================================================
# 建立表格與插入測試
# ============================================================================

class TestCreateAndInsert:
    """CREATE TABLE 和 INSERT 測試"""

    def test_create_table(self, db):
        """測試：CREATE TABLE 建立表格"""
        db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
        cursor = db.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = cursor.fetchall()
        table_names = [t[0] for t in tables]
        assert "users" in table_names

    def test_insert_single_row(self, db):
        """測試：INSERT 單列"""
        db.execute("CREATE TABLE test_insert (id INTEGER, val TEXT)")
        db.execute("INSERT INTO test_insert VALUES (1, 'hello')")
        cursor = db.execute("SELECT * FROM test_insert")
        row = cursor.fetchone()
        assert row == [1, "hello"]

    def test_insert_multiple_rows(self, db):
        """測試：INSERT 多列"""
        db.execute("CREATE TABLE test_multi (id INTEGER, name TEXT)")
        db.execute("INSERT INTO test_multi VALUES (1, 'Alice')")
        db.execute("INSERT INTO test_multi VALUES (2, 'Bob')")
        db.execute("INSERT INTO test_multi VALUES (3, 'Charlie')")
        cursor = db.execute("SELECT * FROM test_multi ORDER BY id")
        rows = cursor.fetchall()
        assert len(rows) == 3
        assert rows[0] == [1, "Alice"]
        assert rows[1] == [2, "Bob"]
        assert rows[2] == [3, "Charlie"]

    def test_update_row(self, db):
        """測試：UPDATE 更新資料"""
        db.execute("CREATE TABLE test_update (id INTEGER, val TEXT)")
        db.execute("INSERT INTO test_update VALUES (1, 'old')")
        db.execute("UPDATE test_update SET val = 'new' WHERE id = 1")
        cursor = db.execute("SELECT val FROM test_update WHERE id = 1")
        row = cursor.fetchone()
        assert row == ["new"]

    def test_delete_row(self, db):
        """測試：DELETE 刪除資料"""
        db.execute("CREATE TABLE test_delete (id INTEGER, val TEXT)")
        db.execute("INSERT INTO test_delete VALUES (1, 'keep')")
        db.execute("INSERT INTO test_delete VALUES (2, 'delete')")
        db.execute("DELETE FROM test_delete WHERE id = 2")
        cursor = db.execute("SELECT COUNT(*) FROM test_delete")
        count = cursor.fetchone()[0]
        assert count == 1


# ============================================================================
# 全文檢索測試
# ============================================================================

class TestFullTextSearch:
    """FTS5 全文檢索測試"""

    @pytest.mark.skip(reason="FTS5 可能尚未完全支援")
    def test_fts5_create(self, db):
        """測試：CREATE VIRTUAL TABLE 建立 FTS5 表格"""
        db.execute("CREATE VIRTUAL TABLE articles USING fts5(title, content)")
        cursor = db.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = cursor.fetchall()
        assert any("articles" in t[0] for t in tables)

    def test_fts5_insert_and_search(self, db):
        """測試：FTS5 插入與搜尋"""
        db.execute("CREATE VIRTUAL TABLE docs USING fts5(content)")
        db.execute("INSERT INTO docs VALUES ('Python is great')")
        db.execute("INSERT INTO docs VALUES ('Rust is fast')")
        db.execute("INSERT INTO docs VALUES ('SQL is powerful')")

        cursor = db.execute("SELECT * FROM docs WHERE docs MATCH 'Python'")
        rows = cursor.fetchall()
        assert len(rows) >= 1

    @pytest.mark.skip(reason="FTS5 中文可能尚未完全支援")
    def test_fts5_chinese(self, db):
        """測試：FTS5 中文全文檢索"""
        db.execute("CREATE VIRTUAL TABLE articles USING fts5(title, body)")
        db.execute("INSERT INTO articles VALUES ('Hello World', 'The quick brown fox')")
        db.execute("INSERT INTO articles VALUES ('Rust Guide', 'Memory safety without GC')")
        db.execute("INSERT INTO articles VALUES ('資料庫', '繁體中文全文檢索')")

        cursor = db.execute("SELECT * FROM articles WHERE articles MATCH 'rust'")
        rows = cursor.fetchall()
        assert len(rows) >= 1

        cursor = db.execute("SELECT * FROM articles WHERE articles MATCH '資料庫'")
        rows = cursor.fetchall()
        assert len(rows) >= 1


# ============================================================================
# 參數化查詢測試
# ============================================================================

class TestParameterizedQueries:
    """參數化查詢測試（防止 SQL 注入）"""

    def test_parameterized_query_positional(self, db):
        """測試：使用 ? 佔位符的參數化查詢"""
        db.execute("CREATE TABLE t (id INTEGER, val TEXT)")
        db.execute("INSERT INTO t VALUES (?, ?)", (1, "hello"))
        db.execute("INSERT INTO t VALUES (?, ?)", (2, "world"))

        cursor = db.execute("SELECT * FROM t WHERE id = ?", (1,))
        row = cursor.fetchone()
        assert row == [1, "hello"]

    def test_parameterized_multiple_rows(self, db):
        """測試：參數化查詢返回多列"""
        db.execute("CREATE TABLE params (id INTEGER, name TEXT)")
        db.execute("INSERT INTO params VALUES (1, 'Alice')")
        db.execute("INSERT INTO params VALUES (2, 'Bob')")
        db.execute("INSERT INTO params VALUES (3, 'Charlie')")

        cursor = db.execute("SELECT * FROM params WHERE id > ?", (1,))
        rows = cursor.fetchall()
        assert len(rows) == 2

    def test_parameterized_update(self, db):
        """測試：參數化 UPDATE"""
        db.execute("CREATE TABLE uparams (id INTEGER, val TEXT)")
        db.execute("INSERT INTO uparams VALUES (1, 'old')")
        db.execute("UPDATE uparams SET val = ? WHERE id = ?", ("new", 1))
        cursor = db.execute("SELECT val FROM uparams WHERE id = 1")
        row = cursor.fetchone()
        assert row == ["new"]


# ============================================================================
# Cursor 操作測試
# ============================================================================

class TestCursorOperations:
    """Cursor 指標操作測試"""

    def test_cursor_fetchone(self, db):
        """測試：fetchone 取一列"""
        cursor = db.execute("SELECT 1 AS id")
        row = cursor.fetchone()
        assert row == [1]

    def test_cursor_fetchall(self, db):
        """測試：fetchall 取所有列"""
        db.execute("CREATE TABLE fetchall (id INTEGER)")
        db.execute("INSERT INTO fetchall VALUES (1)")
        db.execute("INSERT INTO fetchall VALUES (2)")
        db.execute("INSERT INTO fetchall VALUES (3)")
        cursor = db.execute("SELECT * FROM fetchall")
        rows = cursor.fetchall()
        assert len(rows) == 3

    def test_cursor_iteration(self, db):
        """測試：使用 for 迴圈迭代 cursor"""
        db.execute("CREATE TABLE nums (n INTEGER)")
        for i in range(5):
            db.execute("INSERT INTO nums VALUES (?)", (i,))

        cursor = db.execute("SELECT * FROM nums")
        count = 0
        for row in cursor:
            assert isinstance(row, list)
            count += 1
        assert count == 5


# ============================================================================
# 上下文管理器測試
# ============================================================================

class TestContextManager:
    """上下文管理器（with 語句）測試"""

    def test_context_manager(self):
        """測試：使用 with 語句自動關閉連線"""
        with connect() as db:
            db.execute("CREATE TABLE ctx_test (id INTEGER)")
            db.execute("INSERT INTO ctx_test VALUES (42)")
            cursor = db.execute("SELECT * FROM ctx_test")
            row = cursor.fetchone()
            assert row == [42]

    def test_context_manager_rollback(self):
        """測試：上下文管理器中的例外處理"""
        with pytest.raises(Exception):
            with connect() as db:
                db.execute("CREATE TABLE ctx_rollback (id INTEGER)")
                raise Exception("模擬錯誤")


# ============================================================================
# 磁碟資料庫測試
# ============================================================================

class TestDiskDatabase:
    """持久化磁碟資料庫測試"""

    @pytest.mark.skip(reason="磁碟資料庫可能尚未完全支援")
    def test_disk_database(self):
        """測試：持久化磁碟資料庫"""
        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
            db_path = f.name

        try:
            db = connect(db_path)
            db.execute("CREATE TABLE disk_test (id INTEGER, name TEXT)")
            db.execute("INSERT INTO disk_test VALUES (1, 'persisted')")
            db.close()

            db2 = connect(db_path)
            cursor = db2.execute("SELECT * FROM disk_test")
            row = cursor.fetchone()
            assert row == [1, "persisted"]
            db2.close()
        finally:
            if os.path.exists(db_path):
                os.unlink(db_path)

    @pytest.mark.skip(reason="磁碟資料庫可能尚未完全支援")
    def test_disk_database_multiple_connections(self):
        """測試：多個連線訪問同一磁碟資料庫"""
        with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
            db_path = f.name

        try:
            db1 = connect(db_path)
            db1.execute("CREATE TABLE multi_conn (id INTEGER, val TEXT)")
            db1.execute("INSERT INTO multi_conn VALUES (1, 'first')")
            db1.close()

            db2 = connect(db_path)
            db2.execute("INSERT INTO multi_conn VALUES (2, 'second')")
            db2.close()

            db3 = connect(db_path)
            cursor = db3.execute("SELECT COUNT(*) FROM multi_conn")
            count = cursor.fetchone()[0]
            assert count == 2
            db3.close()
        finally:
            if os.path.exists(db_path):
                os.unlink(db_path)


# ============================================================================
# 錯誤處理測試
# ============================================================================

class TestErrorHandling:
    """錯誤處理測試"""

    def test_nonexistent_table(self, db):
        """測試：查詢不存在的表格"""
        cursor = db.execute("SELECT * FROM nonexistent_table")
        rows = cursor.fetchall()
        assert rows == []

    def test_syntax_error(self, db):
        """測試：語法錯誤處理"""
        try:
            cursor = db.execute("SELECT * FORM invalid_syntax")
            rows = cursor.fetchall()
            assert isinstance(rows, list)
        except Exception:
            pass


# ============================================================================
# 多語句測試
# ============================================================================

class TestMultipleStatements:
    """多語句 SQL 測試"""

    def test_multiple_statements(self, db):
        """測試：在一個 execute 中執行多個語句"""
        db.execute("CREATE TABLE multi (id INTEGER)")
        db.execute("INSERT INTO multi VALUES (1)")
        db.execute("INSERT INTO multi VALUES (2)")
        cursor = db.execute("SELECT COUNT(*) FROM multi")
        count = cursor.fetchone()[0]
        assert count == 2


# ============================================================================
# 聚合函式測試
# ============================================================================

class TestAggregates:
    """聚合函式測試"""

    def test_count_aggregate(self, db):
        """測試：COUNT 聚合函式"""
        db.execute("CREATE TABLE agg_test (val INTEGER)")
        db.execute("INSERT INTO agg_test VALUES (10)")
        db.execute("INSERT INTO agg_test VALUES (20)")
        db.execute("INSERT INTO agg_test VALUES (30)")
        cursor = db.execute("SELECT COUNT(*) FROM agg_test")
        count = cursor.fetchone()[0]
        assert count == 3

    def test_sum_aggregate(self, db):
        """測試：SUM 聚合函式"""
        db.execute("CREATE TABLE sum_test (val INTEGER)")
        db.execute("INSERT INTO sum_test VALUES (10)")
        db.execute("INSERT INTO sum_test VALUES (20)")
        db.execute("INSERT INTO sum_test VALUES (30)")
        cursor = db.execute("SELECT SUM(val) FROM sum_test")
        total = cursor.fetchone()[0]
        assert total == 60

    def test_avg_aggregate(self, db):
        """測試：AVG 聚合函式"""
        db.execute("CREATE TABLE avg_test (val INTEGER)")
        db.execute("INSERT INTO avg_test VALUES (10)")
        db.execute("INSERT INTO avg_test VALUES (20)")
        db.execute("INSERT INTO avg_test VALUES (30)")
        cursor = db.execute("SELECT AVG(val) FROM avg_test")
        average = cursor.fetchone()[0]
        assert average == 20.0

    def test_min_max_aggregate(self, db):
        """測試：MIN/MAX 聚合函式"""
        db.execute("CREATE TABLE minmax_test (val INTEGER)")
        db.execute("INSERT INTO minmax_test VALUES (30)")
        db.execute("INSERT INTO minmax_test VALUES (10)")
        db.execute("INSERT INTO minmax_test VALUES (20)")
        cursor = db.execute("SELECT MIN(val), MAX(val) FROM minmax_test")
        row = cursor.fetchone()
        assert row == [10, 30]


# ============================================================================
# JOIN 測試
# ============================================================================

class TestJoins:
    """JOIN 連接測試"""

    @pytest.mark.skip(reason="JOIN 可能尚未完全支援")
    def test_inner_join(self, db):
        """測試：INNER JOIN 內連接"""
        db.execute("CREATE TABLE orders (id INTEGER, customer_id INTEGER)")
        db.execute("CREATE TABLE customers (id INTEGER, name TEXT)")
        db.execute("INSERT INTO customers VALUES (1, 'Alice')")
        db.execute("INSERT INTO customers VALUES (2, 'Bob')")
        db.execute("INSERT INTO orders VALUES (100, 1)")
        db.execute("INSERT INTO orders VALUES (101, 2)")
        cursor = db.execute("""
            SELECT o.id, c.name
            FROM orders o
            INNER JOIN customers c ON o.customer_id = c.id
        """)
        rows = cursor.fetchall()
        assert len(rows) == 2


# ============================================================================
# 子查詢測試
# ============================================================================

class TestSubqueries:
    """子查詢測試"""

    def test_subquery_in_where(self, db):
        """測試：WHERE 子句中的子查詢"""
        db.execute("CREATE TABLE outer_tbl (id INTEGER)")
        db.execute("CREATE TABLE inner_tbl (id INTEGER, val TEXT)")
        db.execute("INSERT INTO inner_tbl VALUES (1, 'found')")
        db.execute("INSERT INTO inner_tbl VALUES (2, 'not found')")
        cursor = db.execute("""
            SELECT * FROM inner_tbl
            WHERE id IN (SELECT id FROM inner_tbl WHERE id = 1)
        """)
        rows = cursor.fetchall()
        assert len(rows) == 1
        assert rows[0][1] == "found"


# ============================================================================
# 交易測試
# ============================================================================

class TestTransactions:
    """交易測試"""

    def test_explicit_transaction(self, db):
        """測試：明確的 BEGIN/COMMIT"""
        db.execute("CREATE TABLE trans_test (id INTEGER)")
        db.execute("BEGIN")
        db.execute("INSERT INTO trans_test VALUES (1)")
        db.execute("INSERT INTO trans_test VALUES (2)")
        db.execute("COMMIT")
        cursor = db.execute("SELECT COUNT(*) FROM trans_test")
        count = cursor.fetchone()[0]
        assert count == 2


if __name__ == "__main__":
    pytest.main([__file__, "-v"])