tests/unit/test_sqlite_connector.py

"""
tests/unit/test_sqlite_connector.py
Tests for SqliteConnector using a real in-memory SQLite file written to /tmp.
No network calls.
"""

import os
import sqlite3
import tempfile

import pandas as pd
import pytest


@pytest.fixture
def sqlite_file_bytes():
    """Create a real SQLite file in /tmp and return its bytes."""
    with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as f:
        path = f.name

    conn = sqlite3.connect(path)
    conn.execute("CREATE TABLE sales (id INTEGER PRIMARY KEY, region TEXT, amount REAL)")
    conn.execute("INSERT INTO sales VALUES (1, 'North', 100.0)")
    conn.execute("INSERT INTO sales VALUES (2, 'South', 200.0)")
    conn.execute("INSERT INTO sales VALUES (3, 'North', 150.0)")
    conn.commit()
    conn.close()

    with open(path, "rb") as f:
        data = f.read()

    os.unlink(path)
    return data


@pytest.fixture
def sqlite_connector(sqlite_file_bytes, mocker):
    """SqliteConnector with mocked HTTP download."""
    mock_resp = mocker.MagicMock()
    mock_resp.content = sqlite_file_bytes
    mock_resp.raise_for_status.return_value = None
    mocker.patch("requests.get", return_value=mock_resp)

    from connectors.sqlite_connector import SqliteConnector
    return SqliteConnector(supabase_url="http://fake/file.sqlite")


@pytest.mark.unit
class TestSqliteConnector:
    def test_get_schema_lists_tables(self, sqlite_connector):
        schema = sqlite_connector.get_schema()
        table_names = [t["table"] for t in schema]
        assert "sales" in table_names

    def test_get_schema_has_columns(self, sqlite_connector):
        schema = sqlite_connector.get_schema()
        sales = next(t for t in schema if t["table"] == "sales")
        col_names = {c["name"] for c in sales["columns"]}
        assert "region" in col_names
        assert "amount" in col_names

    def test_get_schema_row_count(self, sqlite_connector):
        schema = sqlite_connector.get_schema()
        sales = next(t for t in schema if t["table"] == "sales")
        assert sales["row_count"] == 3

    def test_execute_sql_select_all(self, sqlite_connector):
        rows = sqlite_connector.execute_sql("SELECT * FROM sales")
        assert len(rows) == 3

    def test_execute_sql_where_clause(self, sqlite_connector):
        rows = sqlite_connector.execute_sql("SELECT * FROM sales WHERE region = 'North'")
        assert len(rows) == 2

    def test_execute_sql_aggregation(self, sqlite_connector):
        rows = sqlite_connector.execute_sql("SELECT SUM(amount) as total FROM sales")
        assert rows[0]["total"] == pytest.approx(450.0)

    def test_load_dataframe_returns_dataframe(self, sqlite_connector):
        df = sqlite_connector.load_dataframe(table="sales")
        assert isinstance(df, pd.DataFrame)
        assert len(df) == 3

    def test_load_dataframe_without_table_uses_first(self, sqlite_connector):
        df = sqlite_connector.load_dataframe()
        assert isinstance(df, pd.DataFrame)
        assert len(df) == 3