Upload 466 files

a72d1cf verified 8 days ago

7.36 kB

"""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())

	"""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())