test_engine.py

"""
Tests for the Intelligent Query & Insight Engine.
Run with: pytest tests/ -v
"""

import pytest
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app.database import init_db, seed_sample_data, execute_safe_query, get_analytics
from app.cache import InMemoryCache
from app.query_engine import _validate_sql, _inject_user_id
import time


# ── Database tests ────────────────────────────────────────────────────────────

class TestDatabase:
    def setup_method(self):
        os.environ["DB_PATH"] = ":memory:"
        # Need a fresh connection each time
        import app.database as db
        db.DB_PATH = ":memory:"

    def test_init_and_seed(self):
        init_db()
        seed_sample_data()
        rows = execute_safe_query("SELECT COUNT(*) as c FROM transactions", ())
        assert rows[0]["c"] == 600  # 3 users × 200

    def test_parameterized_query_filters_by_user(self):
        init_db()
        seed_sample_data()
        rows = execute_safe_query(
            "SELECT COUNT(*) as c FROM transactions WHERE user_id = ?", (1,)
        )
        assert rows[0]["c"] == 200

    def test_analytics_structure(self):
        init_db()
        seed_sample_data()
        data = get_analytics(1)
        assert "total_spent" in data
        assert "by_category" in data
        assert "by_month" in data
        assert data["total_transactions"] == 200


# ── Cache tests ───────────────────────────────────────────────────────────────

class TestInMemoryCache:
    def test_set_get(self):
        c = InMemoryCache()
        c.set("k", {"foo": "bar"}, ttl=60)
        assert c.get("k") == {"foo": "bar"}

    def test_miss_returns_none(self):
        c = InMemoryCache()
        assert c.get("nonexistent") is None

    def test_ttl_expiry(self):
        c = InMemoryCache()
        c.set("expire_me", "value", ttl=1)
        time.sleep(1.1)
        assert c.get("expire_me") is None

    def test_size(self):
        c = InMemoryCache()
        c.set("a", 1, ttl=60)
        c.set("b", 2, ttl=60)
        assert c.size() == 2

    def test_delete(self):
        c = InMemoryCache()
        c.set("del_me", "v", ttl=60)
        c.delete("del_me")
        assert c.get("del_me") is None


# ── SQL validation tests ──────────────────────────────────────────────────────

class TestSQLValidation:
    def test_valid_select(self):
        assert _validate_sql("SELECT SUM(amount) FROM transactions WHERE user_id = ?") is True

    def test_blocks_insert(self):
        assert _validate_sql("INSERT INTO transactions VALUES (1,2,'Food','X','2024-01-01')") is False

    def test_blocks_drop(self):
        assert _validate_sql("SELECT 1; DROP TABLE transactions") is False

    def test_blocks_update(self):
        assert _validate_sql("UPDATE transactions SET amount = 0") is False

    def test_blocks_delete(self):
        assert _validate_sql("DELETE FROM transactions") is False

    def test_blocks_pragma(self):
        assert _validate_sql("PRAGMA user_version") is False


# ── Param injection tests ─────────────────────────────────────────────────────

class TestParamInjection:
    def test_injects_at_none_position(self):
        params = [None, "Food"]
        result = _inject_user_id(params, 42)
        assert result == [42, "Food"]

    def test_prepends_when_no_none(self):
        params = ["Food"]
        result = _inject_user_id(params, 7)
        assert result == [7, "Food"]

    def test_empty_params(self):
        result = _inject_user_id([], 5)
        assert result == [5]