DataOps-platform/tests/test_ddl_parser_view.py

"""Tests for DDL parser with CREATE VIEW SQL files."""

from pathlib import Path

from app.core.llm.ddl_parser import DDLParser


SQL_FILE = Path(__file__).with_name("TB_JC_SFXM表ddl.sql")


def test_parse_create_view_fallback():
    sql_content = SQL_FILE.read_text(encoding="utf-8")
    parser = DDLParser(api_key="test-key")

    result = parser._parse_sql_ddl_fallback(sql_content)

    assert len(result) == 1
    table_info = result[0]["table_info"]
    assert table_info["name_en"] == "TB_JC_SFXM"
    assert len(result[0]["columns"]) == 23
    assert result[0]["columns"][0]["name_en"] == "YLJGDM"
    assert result[0]["columns"][1]["name_en"] == "XMDM"


def test_parse_ddl_prefers_local_sql_before_llm(monkeypatch):
    sql_content = SQL_FILE.read_text(encoding="utf-8")
    parser = DDLParser(api_key="test-key")

    def fail_if_called(_sql):
        raise AssertionError("LLM should not be called when local SQL parse succeeds")

    monkeypatch.setattr(parser, "_parse_ddl_with_llm", fail_if_called)
    result = parser.parse_ddl(sql_content)

    assert isinstance(result, list)
    assert result[0]["table_info"]["name_en"] == "TB_JC_SFXM"


def test_normalize_empty_llm_list_uses_fallback(monkeypatch):
    sql_content = "SELECT 1"
    parser = DDLParser(api_key="test-key")
    monkeypatch.setattr(parser, "_parse_ddl_with_llm", lambda _sql: [])

    result = parser.parse_ddl(sql_content)

    assert result == []


def test_normalize_error_dict_without_table_info():
    raw = {"code": 500, "message": "API请求失败"}
    assert DDLParser.normalize_ddl_parse_result(raw) == []