| """Otomasyon modülü: kademe/rastgele seçim, dış test ekleme, uçtan uca kural tabanlı çalıştırma ve raporlar.""" |
| import io |
| import json |
| import zipfile |
| from collections import Counter |
|
|
| import pytest |
|
|
| from otomasyon import calistir, isler, katalog, modeller |
| from otomasyon.metin import M |
|
|
|
|
| def test_tiers(): |
| assert modeller.tier("claude|claude-opus-5") == 1 |
| assert modeller.tier("claude|claude-fable-5-1") == 1 |
| assert modeller.tier("claude|claude-sonnet-5-5") == 2 |
| assert modeller.tier("claude|claude-haiku-4-5-20251001") == 3 |
| assert modeller.tier("claude|claude-3-opus-20240229") == 3 |
| assert modeller.tier("openai|gpt-6-astra") == 1 |
| assert modeller.tier("openai|gpt-5-mini") == 2 |
| assert modeller.tier("openai|gpt-4.1-nano") == 3 |
| assert modeller.tier("ollama|qwen2.5-coder:32b") == 1 |
| assert modeller.tier("ollama|qwen2.5-coder:14b") == 2 |
| assert modeller.tier("ollama|qwen2.5-coder:7b") == 3 |
| assert modeller.tier("ollama|x:latest", {"note": "32.8B · Q4_K_M"}) == 1 |
| assert modeller.tier("kural") == 3 |
|
|
|
|
| def test_picker_prefers_top_tier(): |
| pool = [{"secim": "a", "kademe": 1}, {"secim": "b", "kademe": 2}, {"secim": "c", "kademe": 3}] |
| pk = modeller.Picker(pool, seed=1, bias=6) |
| c = Counter(pk.pick()["secim"] for _ in range(3000)) |
| assert c["a"] > c["b"] > c["c"] |
| assert c["a"] / 3000 > 0.5 |
| pk.drop("a") |
| assert all(pk.pick()["secim"] != "a" for _ in range(50)) |
| assert pk.pick(exclude={"b", "c"}) is None |
|
|
|
|
| def test_metin_has_both_languages(): |
| for k, v in M.items(): |
| assert v.get("tr") and v.get("en"), k |
|
|
|
|
| def test_validate_questions(): |
| ok = [{"soru": "x?", "tip": "sayi", "cevap": 5, "belge": "a.pdf"}, {"soru": "y?", "tip": "cevapsiz"}] |
| assert katalog.validate_questions(ok, {"a.pdf"}, False) == [] |
| bad = [{"soru": "x?", "tip": "sayi", "cevap": "beş"}, {"soru": "", "tip": "metin"}, |
| {"soru": "z?", "tip": "liste", "cevap": "a"}, {"soru": "w?", "tip": "metin", "cevap": "k", "belge": "yok.pdf"}] |
| errs = katalog.validate_questions(bad, {"a.pdf"}, False) |
| assert len(errs) == 4 |
|
|
|
|
| @pytest.fixture() |
| def tmp_dirs(tmp_path, monkeypatch): |
| monkeypatch.setattr(katalog, "EXT_DIR", tmp_path / "testler") |
| monkeypatch.setattr(isler, "JOBS_DIR", tmp_path / "isler") |
| return tmp_path |
|
|
|
|
| def _sample_zip() -> bytes: |
| csv = "İl;2023;2024\nAnkara;1.234,5;1.300,0\nİzmir;980,0;1.010,5\nToplam;2.214,5;2.310,5\n" |
| qs = {"sorular": [ |
| {"soru": "Ankara'nın 2024 değeri nedir?", "tip": "sayi", "cevap": 1300.0, "belge": "iller.csv"}, |
| {"soru": "2024'te en yüksek değere sahip il hangisidir?", "tip": "metin", "cevap": "Ankara", "belge": "iller.csv"}, |
| {"soru": "Bursa'nın 2024 değeri nedir?", "tip": "cevapsiz", "belge": "iller.csv"}]} |
| buf = io.BytesIO() |
| with zipfile.ZipFile(buf, "w") as z: |
| z.writestr("paket/iller.csv", csv.encode("utf-8")) |
| z.writestr("paket/sorular.json", json.dumps(qs, ensure_ascii=False)) |
| z.writestr("paket/test.json", json.dumps({"ad": "İl değerleri", "aciklama": "küçük CSV testi"})) |
| return buf.getvalue() |
|
|
|
|
| def test_add_external_and_run_end_to_end(tmp_dirs): |
| test, errs = katalog.add_external_test("", "", False, zip_bytes=_sample_zip()) |
| assert errs == [] and test is not None |
| assert test.name() == "İl değerleri" and test.n_docs() == 1 and len(test.questions()) == 3 |
| assert any(x.key == test.key for x in katalog.all_tests()) |
| spec = {"ad": "birim testi", "testler": [test.key, "pilot"], "kip": "single", "model": "kural", "modeller": [], |
| "havuz": {}, "ust_agirlik": 6, "sinir": 4, "tohum": 1, "kendini_duzeltme": True, "max_deneme": 3, |
| "yedek": True, "diller": ["tr", "en"], "bicimler": ["html", "pdf"]} |
| d = isler.create_job(spec) |
| calistir.main(d) |
| st = isler.read_status(d) |
| assert st["faz"] == "bitti", st.get("hata") |
| assert [u["durum"] for u in st["birimler"]] == ["bitti", "bitti"] |
| u = st["birimler"][0] |
| assert u["yapilan"] == 3 |
| for k in ("tr_html", "tr_pdf", "en_html", "en_pdf"): |
| p = d / u["klasor"] / u["raporlar"][k] |
| assert p.exists() and p.stat().st_size > 1000 |
| assert (d / "genel_rapor_tr.pdf").read_bytes()[:4] == b"%PDF" |
| assert "İl değerleri" in (d / "genel_rapor_tr.html").read_text(encoding="utf-8") |
| z = zipfile.ZipFile(d / "tum_raporlar.zip") |
| assert sum(n.endswith(".pdf") for n in z.namelist()) == 6 |
| |
| alone = (d / "genel_rapor_tr.html").read_text(encoding="utf-8") |
| assert "<a href" not in alone and "t2sTema" in alone |
| packed = z.read(f"{d.name}/genel_rapor_tr.html").decode("utf-8") |
| hrefs = [h for h in packed.split("href='")[1:]] |
| assert hrefs and all(f"{d.name}/" + h.split("'")[0] in z.namelist() for h in hrefs) |
| assert "t2sTema" in z.read(f"{d.name}/{u['klasor']}/rapor_en.html").decode("utf-8") |
| assert isler.unacked_finished() == d |
| isler.ack(d) |
| assert isler.unacked_finished() is None |
| assert katalog.delete_external_test(test.key) |
|
|
|
|
| def test_sampling_is_reproducible(): |
| qs = [{"i": i} for i in range(100)] |
| a = calistir.sample_questions(qs, 10, 5, "x") |
| assert a == calistir.sample_questions(qs, 10, 5, "x") |
| assert a != calistir.sample_questions(qs, 10, 6, "x") |
| assert calistir.sample_questions(qs, 0, 5, "x") == qs |
|
|
|
|
| def test_model_broken(): |
| recs = [{"durum": "cevap", "aciklama": ""}] * 5 |
| assert not calistir.model_broken({"kayitlar": recs}, 10, 2) |
| assert calistir.model_broken({"kayitlar": recs}, 10, 6) |
| assert calistir.model_broken(None, 0, 0) |
| assert calistir.model_broken({"kayitlar": [{"durum": "istisna", "aciklama": "istisna: x"}] * 5}, 0, 0) |
|
|
|
|
| def test_health_check_filters_broken_models(tmp_path): |
| import os |
| import socket |
| import subprocess |
| import sys |
| import time |
| from pathlib import Path |
|
|
| from t2s.llm import list_models |
| from t2s.settings import Settings |
| root = Path(__file__).resolve().parent.parent |
| s = socket.socket() |
| s.bind(("127.0.0.1", 0)) |
| port = s.getsockname()[1] |
| s.close() |
| p = subprocess.Popen([sys.executable, str(root / "tests" / "mock_llm_server.py"), str(port)], |
| env={**os.environ, "MOCK_EXTRA": "1"}, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) |
| try: |
| url = f"http://127.0.0.1:{port}" |
| for _ in range(50): |
| try: |
| list_models("ollama", None, url) |
| break |
| except Exception: |
| time.sleep(0.1) |
| cfg = Settings(tmp_path / "ayar.json") |
| for prov, base in (("claude", url), ("openai", url + "/v1"), ("ollama", url)): |
| cfg.conn(prov)["url"] = base |
| if prov != "ollama": |
| cfg.set_key(prov, "test") |
| cfg.set_models(prov, [{"id": m.id, "label": m.label, "note": m.note} |
| for m in list_models(prov, "test" if prov != "ollama" else None, base)]) |
| pool = [c["secim"] for c in modeller.candidates(cfg)] |
| res = {r["secim"]: r for r in modeller.health_check(cfg, pool)} |
| assert not res["openai|gpt-mock-bozuk"]["ok"] |
| assert res["claude|claude-mock-large"]["ok"] and res["ollama|qwen2.5-coder:7b"]["ok"] |
| assert res["claude|claude-mock-dengesiz"]["ok"] |
| finally: |
| p.terminate() |
|
|
|
|
| def test_costly_models_excluded_by_default(): |
| ch = ["openai|gpt-5", "openai|o3-pro", "openai|gpt-5-pro-2025", "openai|gpt-4o-search-preview", |
| "claude|claude-opus-5", "ollama|qwen2.5-coder:7b"] |
| assert modeller.default_excluded(ch) == ["openai|o3-pro", "openai|gpt-5-pro-2025", "openai|gpt-4o-search-preview"] |
|
|
