opsxsql/tests/test_sql_analyzer.py

"""Tests for SQL AST Analyzer (Phase 4.1)."""
import pytest
from app.services.sql_analyzer import SQLAnalyzer, SQLType, RiskLevel


analyzer = SQLAnalyzer()


class TestSQLTypeDetection:
    def test_select(self):
        result = analyzer.analyze("SELECT id, name FROM users WHERE id = 1;")
        assert result.sql_type == SQLType.SELECT

    def test_insert(self):
        result = analyzer.analyze("INSERT INTO users (name) VALUES ('test');")
        assert result.sql_type == SQLType.INSERT

    def test_update(self):
        result = analyzer.analyze("UPDATE users SET name = 'test' WHERE id = 1;")
        assert result.sql_type == SQLType.UPDATE

    def test_delete(self):
        result = analyzer.analyze("DELETE FROM users WHERE id = 1;")
        assert result.sql_type == SQLType.DELETE

    def test_drop_table(self):
        result = analyzer.analyze("DROP TABLE users;")
        assert result.sql_type == SQLType.DROP

    def test_truncate(self):
        result = analyzer.analyze("TRUNCATE TABLE logs;")
        assert result.sql_type == SQLType.TRUNCATE

    def test_alter_table(self):
        result = analyzer.analyze("ALTER TABLE users ADD COLUMN age INT;")
        assert result.sql_type == SQLType.ALTER

    def test_create_table(self):
        result = analyzer.analyze("CREATE TABLE test (id INT PRIMARY KEY);")
        assert result.sql_type == SQLType.CREATE

    def test_cte_select(self):
        result = analyzer.analyze("WITH cte AS (SELECT 1) SELECT * FROM cte;")
        assert result.sql_type == SQLType.SELECT

    def test_explain(self):
        result = analyzer.analyze("EXPLAIN SELECT * FROM users;")
        assert result.sql_type == SQLType.SELECT


class TestReadonlyValidation:
    def test_select_is_safe(self):
        safe, msg = analyzer.validate_readonly("SELECT id FROM users LIMIT 10;")
        assert safe is True

    def test_insert_not_safe(self):
        safe, msg = analyzer.validate_readonly("INSERT INTO users (name) VALUES ('test');")
        assert safe is False
        assert "INSERT" in msg

    def test_update_not_safe(self):
        safe, msg = analyzer.validate_readonly("UPDATE users SET name = 'test' WHERE id = 1;")
        assert safe is False

    def test_delete_not_safe(self):
        safe, msg = analyzer.validate_readonly("DELETE FROM users WHERE id = 1;")
        assert safe is False

    def test_drop_not_safe(self):
        safe, msg = analyzer.validate_readonly("DROP TABLE users;")
        assert safe is False

    def test_truncate_not_safe(self):
        safe, msg = analyzer.validate_readonly("TRUNCATE TABLE logs;")
        assert safe is False

    def test_alter_not_safe(self):
        safe, msg = analyzer.validate_readonly("ALTER TABLE users ADD COLUMN age INT;")
        assert safe is False

    def test_create_not_safe(self):
        safe, msg = analyzer.validate_readonly("CREATE TABLE test (id INT);")
        assert safe is False


class TestRiskLevels:
    def test_select_is_safe_risk(self):
        result = analyzer.analyze("SELECT * FROM users;")
        assert result.risk_level == RiskLevel.SAFE

    def test_dml_with_where_is_medium(self):
        result = analyzer.analyze("UPDATE users SET name = 'a' WHERE id = 1;")
        assert result.risk_level == RiskLevel.MEDIUM

    def test_dml_without_where_is_high(self):
        result = analyzer.analyze("UPDATE users SET name = 'a';")
        assert result.risk_level == RiskLevel.HIGH

    def test_ddl_is_dangerous(self):
        result = analyzer.analyze("DROP TABLE users;")
        assert result.risk_level == RiskLevel.DANGEROUS


class TestSecurityChecks:
    def test_multi_statement_rejected(self):
        safe, msg = analyzer.validate_readonly("SELECT 1; DROP TABLE users;")
        assert safe is False
        assert "多条" in msg or "分号" in msg

    def test_into_outfile_rejected(self):
        safe, msg = analyzer.validate_readonly("SELECT * INTO OUTFILE '/tmp/data.txt' FROM users;")
        assert safe is False
        assert "OUTFILE" in msg

    def test_subquery_dml_rejected(self):
        safe, msg = analyzer.validate_readonly("SELECT * FROM (DELETE FROM users) AS t;")
        assert safe is False

    def test_normal_subquery_ok(self):
        safe, msg = analyzer.validate_readonly("SELECT * FROM (SELECT id FROM users) AS t;")
        assert safe is True

    def test_semicolon_in_string_ok(self):
        safe, msg = analyzer.validate_readonly("SELECT 'hello;world' FROM users;")
        assert safe is True


class TestAnalysis:
    def test_has_where(self):
        result = analyzer.analyze("SELECT * FROM users WHERE id = 1;")
        assert result.has_where is True

    def test_no_where(self):
        result = analyzer.analyze("SELECT * FROM users;")
        assert result.has_where is False

    def test_to_dict(self):
        result = analyzer.analyze("SELECT id FROM users;")
        d = result.to_dict()
        assert "sql_type" in d
        assert "risk_level" in d
        assert "is_readonly" in d
        assert d["sql_type"] == "SELECT"
        assert d["is_readonly"] is True

    def test_complex_select(self):
        result = analyzer.analyze("""
            SELECT u.id, u.name, o.total
            FROM users u
            JOIN orders o ON u.id = o.user_id
            WHERE u.status = 'active'
            ORDER BY o.total DESC
            LIMIT 50;
        """)
        assert result.sql_type == SQLType.SELECT
        assert result.is_readonly is True
        assert result.has_where is True