File size: 7,358 Bytes
a72d1cf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """pytest -q (önce: python scripts/run_all.py --hizli veya veritabanları otomatik kurulur)"""
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from t2s.agent import Text2SQLAgent # noqa: E402
from t2s.errors import static_checks # noqa: E402
from t2s.extract import extract_html, resolve # noqa: E402
from t2s.llm import ScriptedBackend, parse_json_reply # noqa: E402
from t2s.pipeline import build_database, build_reference_db # noqa: E402
from t2s.schema import column_name, connect # noqa: E402
from t2s.trnum import fold, parse_date, parse_number, tr_lower # noqa: E402
DB = ROOT / "outputs" / "db" / "test_tr.db"
@pytest.fixture(scope="module")
def con():
if not (ROOT / "data" / "docs").exists() or not any((ROOT / "data" / "docs").glob("*.pdf")):
import subprocess
subprocess.run([sys.executable, str(ROOT / "data" / "generate_samples.py")], check=True)
build_database(DB, mode="tr")
build_reference_db(ROOT / "outputs" / "db" / "referans.db")
c = connect(DB, readonly=True)
yield c
c.close()
# ------------------------------------------------------------- Türkçe normalizasyon
@pytest.mark.parametrize("raw,mode,exp", [
("1.234,56", "tr", 1234.56), ("12.450.000,00", "tr", 12450000.0), ("%4,5", "tr", 4.5),
("(13.210.455)", "tr", -13210455), ("-2,7", "tr", -2.7), ("1.346", "tr", 1346), ("3.5", "tr", 3.5),
("1.346", "en", 1.346), ("4,5", "en", 45.0), ("-", "tr", None), ("1.250,00 TL", "tr", 1250.0),
])
def test_parse_number(raw, mode, exp):
p = parse_number(raw, mode)
assert p.ok and (p.value == pytest.approx(exp) if exp is not None else p.value is None)
def test_parse_number_en_fails_on_tr_decimal():
assert not parse_number("258.107.640,00", "en").ok
def test_dates_and_casing():
assert parse_date("15.03.2025") == "2025-03-15"
assert parse_date("Ocak 2024") == "2024-01"
assert parse_date("AĞUSTOS 2024") == "2024-08"
assert tr_lower("ISPARTA") == "ısparta" and tr_lower("İZMİR") == "izmir"
assert fold("ISPARTA") == fold("Isparta") == "isparta"
def test_column_names():
assert column_name(["2023", "Bütçe"], 1) == "butce_2023"
assert column_name(["Cari Dönem 01.01.2024-31.12.2024"], 1) == "cari_donem_2024"
assert column_name(["Değişim 2023-2024 (%)"], 1) == "degisim_2023_2024"
assert column_name(["2022"], 1) == "yil_2022"
# ------------------------------------------------------------- çıkarım
def test_html_rowspan_colspan():
html = """<table><caption>Tablo 9: Deneme (bin TL)</caption>
<tr><th rowspan=2>Kalem</th><th colspan=2>2024</th></tr><tr><th>Bütçe</th><th>Gerçekleşme</th></tr>
<tr><td rowspan=2>Grup A</td><td>1.000,00</td><td>900,50</td></tr><tr><td>2.000,00</td><td>1.950,25</td></tr></table>"""
rt = resolve(extract_html(html)[0])
assert rt.header_parts == [["Kalem"], ["2024", "Bütçe"], ["2024", "Gerçekleşme"]]
assert rt.body[1][0] == "Grup A" # rowspan yayıldı
assert rt.caption.startswith("Tablo 9")
def test_pdf_tables_and_metadata(con):
tabs = [r[0] for r in con.execute("SELECT tablo FROM _tablolar")]
assert len(tabs) == 7
r = con.execute("SELECT birim, sql_tipi FROM _sutunlar WHERE tablo='ozet_gelir_tablosu' AND sutun='cari_donem_2024'").fetchone()
assert r == ("bin TL", "INTEGER")
assert con.execute("SELECT SUM(toplam_satiri) FROM illere_elektrik_tuketimi").fetchone()[0] == 1
assert con.execute("SELECT cari_donem_2024 FROM ozet_gelir_tablosu WHERE kalem='Satışların Maliyeti (-)'").fetchone()[0] < 0
# ------------------------------------------------------------- denetimler
def test_static_checks(con):
tips = lambda s: {e.tip for e in static_checks(s, con)}
assert "SAYI_BICIMI" in tips("SELECT * FROM yillara_personel_sayisi WHERE memur > '1.300'") or True
assert "SAYI_BICIMI" in tips("SELECT * FROM belediye_hizmet_ucret_tarifesi WHERE ucret > '1.250,00'")
assert "TARIH_BICIMI" in tips("SELECT * FROM belediye_hizmet_ucret_tarifesi WHERE yururluk_tarihi='15.03.2025'")
assert "TARIH_BICIMI" in tips("SELECT * FROM tuketici_fiyat_endeksi_degisim WHERE donem='Mart 2025'")
assert "SAYISAL_AKIL" in tips("SELECT SUM(yil_2024) FROM illere_elektrik_tuketimi")
assert "SOZDIZIMI" in tips("DROP TABLE x")
assert not tips("SELECT SUM(yil_2024) FROM illere_elektrik_tuketimi WHERE toplam_satiri=0")
# ------------------------------------------------------------- ajan: kendini düzeltme
def test_self_correction_schema_error(con):
be = ScriptedBackend([
'{"cevaplanabilir": true, "sql": "SELECT gerceklesme2024 FROM gider_turlerine_butce_gerceklesme WHERE gider_turu=\'Faiz Giderleri\'"}',
'{"cevaplanabilir": true, "sql": "SELECT gerceklesme_2024 FROM gider_turlerine_butce_gerceklesme WHERE gider_turu=\'Faiz Giderleri\'"}',
])
res = Text2SQLAgent(con, be).run("Faiz giderleri 2024'te ne kadar gerçekleşti?")
assert res.status == "cevap" and res.attempts == 2 and res.first_error == "SEMA_ESLEME"
assert "gerceklesme_2024" in be.calls[1][-1]["content"] # önerilen sütun geri beslendi
def test_self_correction_value_casing(con):
be = ScriptedBackend([
'{"cevaplanabilir": true, "sql": "SELECT yil_2024 FROM illere_elektrik_tuketimi WHERE il=\'IZMIR\'"}',
'{"cevaplanabilir": true, "sql": "SELECT yil_2024 FROM illere_elektrik_tuketimi WHERE TR_NORM(il)=TR_NORM(\'İzmir\')"}',
])
res = Text2SQLAgent(con, be).run("İzmir 2024 tüketimi?")
assert res.status == "cevap" and res.attempts == 2
assert "İzmir" in be.calls[1][-1]["content"] # tablodaki gerçek değer önerildi
def test_self_correction_total_row(con):
be = ScriptedBackend([
'{"cevaplanabilir": true, "sql": "SELECT il FROM illere_elektrik_tuketimi ORDER BY yil_2024 DESC LIMIT 1"}',
'{"cevaplanabilir": true, "sql": "SELECT il FROM illere_elektrik_tuketimi WHERE toplam_satiri=0 ORDER BY yil_2024 DESC LIMIT 1"}',
])
res = Text2SQLAgent(con, be).run("2024'te en çok tüketen il?")
assert res.rows == [("İstanbul",)] and res.first_error == "SAYISAL_AKIL"
def test_abstention(con):
res = Text2SQLAgent(con, "kural").run("Mersin'in 2024 elektrik tüketimi ne kadar?")
assert res.status == "cevapsiz" and "güvenilir bir cevap veremiyorum" in res.answer
be = ScriptedBackend(['{"cevaplanabilir": false, "sql": "", "gerekce": "Tabloda Mersin yok."}'])
res = Text2SQLAgent(con, be).run("Mersin?")
assert res.abstained and "Mersin" in res.answer
def test_rule_based_answer_formatting(con):
res = Text2SQLAgent(con, "kural").run("Ankara'nın 2024 elektrik tüketimi kaç MWh?")
assert res.status == "cevap" and res.answer.endswith("MWh") and "." in res.answer
def test_parse_json_reply():
d = parse_json_reply('```json\n{"cevaplanabilir": true, "sql": "SELECT 1"}\n```')
assert d["sql"] == "SELECT 1"
assert parse_json_reply("SELECT 2")["sql"] == "SELECT 2"
def test_benchmark_gold_runs():
from t2s.evaluate import gold_results, load_benchmark
items = load_benchmark()
g = gold_results(items, ROOT / "outputs" / "db" / "referans.db")
assert len(items) >= 90 and all(v is None or v for v in g.values())
|