"""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 = """
Tablo 9: Deneme (bin TL)
Kalem2024
BütçeGerçekleşme
Grup A1.000,00900,50
2.000,001.950,25
""" 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())