24 lines
1.0 KiB
Python
24 lines
1.0 KiB
Python
from services.integrations.ragflow_client import extract_table_name
|
|
from services.integrations.ragflow_sync import RagflowSync
|
|
|
|
|
|
def test_extract_table_name_normalizes_legacy_multiple_impact_alias():
|
|
assert extract_table_name({"metadata": {"table": "apbo_tp_multiple_impact"}}) == "apbo_eta_multiple_impact"
|
|
assert extract_table_name({"table_name": "apbo_tp_multiple_impact"}) == "apbo_eta_multiple_impact"
|
|
assert extract_table_name({"content": '{"table":"apbo_tp_multiple_impact"}'}) == "apbo_eta_multiple_impact"
|
|
|
|
|
|
def test_sync_table_retrieval_delegates_to_update(monkeypatch):
|
|
expected = {"ok": True, "source": "table"}
|
|
monkeypatch.setattr(RagflowSync, "update_table_retrieval_documents", lambda self: expected)
|
|
|
|
assert RagflowSync().sync_table_retrieval() == expected
|
|
|
|
|
|
def test_sync_sql_gen_prompts_delegates_to_update(monkeypatch):
|
|
expected = {"ok": True, "source": "sql_gen"}
|
|
monkeypatch.setattr(RagflowSync, "update_sql_gen_documents", lambda self: expected)
|
|
|
|
assert RagflowSync().sync_sql_gen_prompts() == expected
|
|
|