"""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) == []