tests(pipeline,verifier): add unit and integration tests to increase coverage and validate end-to-end flow

tests/test_pipeline_extra.py

@@ -0,0 +1,52 @@
+from nl2sql.pipeline import Pipeline
+from nl2sql.types import StageResult
+
+class DetectorOK:
+    def detect(self, *a, **k):
+        return []
+
+class PlannerOK:
+    def run(self, *a, **k):
+        return StageResult(ok=True, data={"plan": "p"})
+
+class GeneratorOK:
+    def run(self, *a, **k):
+        return StageResult(ok=True, data={"sql": "SELECT * FROM t", "rationale": "ok"})
+
+class SafetyOK:
+    def run(self, *a, **k):
+        sql = k.get("sql", "SELECT * FROM t")
+        return StageResult(ok=True, data={"sql": sql})
+
+class ExecOK:
+    def run(self, *a, **k):
+        return StageResult(ok=True, data={"rows": [{"x": 1}]})
+
+class VerifierThenOK:
+    """اولین بار fail، بعد از repair pass می‌کند."""
+    def __init__(self):
+        self.calls = 0
+    def run(self, *, sql, exec_result):
+        self.calls += 1
+        if self.calls == 1:
+            return StageResult(ok=False, error=["first verify fail"])
+        return StageResult(ok=True, data={"verified": True})
+
+class RepairOK:
+    def run(self, *, sql, error_msg, schema_preview):
+        return StageResult(ok=True, data={"sql": "SELECT * FROM t LIMIT 1"})
+
+def test_pipeline_repair_success_path():
+    p = Pipeline(
+        detector=DetectorOK(),
+        planner=PlannerOK(),
+        generator=GeneratorOK(),
+        safety=SafetyOK(),
+        executor=ExecOK(),
+        verifier=VerifierThenOK(),
+        repair=RepairOK(),
+    )
+    out = p.run(user_query="?", schema_preview="")
+    assert out.ok
+    assert out.verified
+    assert not out.error

tests/test_pipeline_integration_real.py

@@ -0,0 +1,98 @@
+import sqlite3
+from nl2sql.pipeline import Pipeline
+from nl2sql.types import StageResult, StageTrace
+
+# --- Realistic dummy stages ----------------------------------
+class DetectorOK:
+    """Always returns no ambiguities."""
+    def detect(self, *a, **k):
+        return []
+
+class PlannerLLM:
+    def run(self, *, user_query, schema_preview):
+        plan = f"Understand user query '{user_query}' and map to table."
+        return StageResult(ok=True, data={"plan": plan},
+                           trace=StageTrace(stage="planner", duration_ms=0))
+
+class GeneratorSimple:
+    def run(self, *, user_query, schema_preview, plan_text, clarify_answers):
+        sql = "SELECT city, COUNT(*) AS cnt FROM users GROUP BY city"
+        return StageResult(ok=True, data={"sql": sql, "rationale": plan_text},
+                           trace=StageTrace(stage="generator", duration_ms=0))
+
+class SafetyReadOnly:
+    def run(self, *, sql):
+        if sql.strip().lower().startswith("select"):
+            return StageResult(ok=True, data={"sql": sql},
+                               trace=StageTrace(stage="safety", duration_ms=0))
+        return StageResult(ok=False, error=["Unsafe query"],
+                           trace=StageTrace(stage="safety", duration_ms=0, notes={"reason": "unsafe"}))
+
+class ExecutorSQLite:
+    """Executes the SQL query on a temporary in-memory SQLite database."""
+    def __init__(self):
+        # create in-memory DB and seed some rows
+        self.conn = sqlite3.connect(":memory:")
+        self._seed()
+
+    def _seed(self):
+        cur = self.conn.cursor()
+        cur.execute("CREATE TABLE users (id INTEGER, city TEXT)")
+        cur.executemany("INSERT INTO users VALUES (?, ?)", [
+            (1, "Berlin"), (2, "Berlin"), (3, "Munich"),
+        ])
+        self.conn.commit()
+
+    def run(self, *, sql):
+        cur = self.conn.cursor()
+        cur.execute(sql)
+        rows = [dict(zip([d[0] for d in cur.description], r)) for r in cur.fetchall()]
+        return StageResult(
+            ok=True,
+            data={"rows": rows},
+            trace=StageTrace(stage="executor", duration_ms=0)
+        )
+
+
+class VerifierCheckCount:
+    def run(self, *, sql, exec_result):
+        rows = exec_result.get("rows", [])
+        ok = bool(rows and "city" in rows[0] and "cnt" in rows[0])
+        return StageResult(ok=ok, data={"verified": ok},
+                           trace=StageTrace(stage="verifier", duration_ms=0,
+                                            notes={"rows_len": len(rows)}))
+
+
+class RepairNoOp:
+    """Dummy repair stage (not triggered in this scenario)."""
+    def run(self, *a, **k):
+        return StageResult(ok=False, error=["no repair needed"])
+
+
+# --- Integration test ----------------------------------------
+def test_pipeline_end_to_end_real_sqlite():
+    """Full NL2SQL pipeline test on real SQLite DB with no mocks."""
+    pipe = Pipeline(
+        detector=DetectorOK(),
+        planner=PlannerLLM(),
+        generator=GeneratorSimple(),
+        safety=SafetyReadOnly(),
+        executor=ExecutorSQLite(),
+        verifier=VerifierCheckCount(),
+        repair=RepairNoOp(),
+    )
+
+    result = pipe.run(user_query="count users per city", schema_preview="users(id, city)")
+
+    # --- Assertions ---
+    assert result.ok
+    assert result.verified
+    assert not result.error
+    assert "SELECT" in result.sql
+
+    # Ensure pipeline produced valid SQL and traces
+    assert isinstance(result.traces, list)
+    assert result.traces  # not empty
+
+    # Logical validation
+    assert "city" in result.sql.lower()

tests/test_verifier.py

@@ -0,0 +1,27 @@
+from nl2sql.verifier import Verifier
+from nl2sql.types import StageResult, StageTrace
+
+def make_exec_result(ok=True, error=None):
+    return StageResult(ok=ok, data={"dummy": True} if ok else None, trace=None, error=error)
+
+def test_verifier_handles_execution_error():
+    v = Verifier()
+    r = v.run(sql="SELECT 1", exec_result=make_exec_result(ok=False, error=["db error"]))
+    assert not r.ok
+    assert "execution_error" in r.trace.notes["reason"]
+    assert r.error == ["db error"]
+
+def test_verifier_detects_agg_without_group():
+    v = Verifier()
+    sql = "SELECT COUNT(*) FROM users"
+    r = v.run(sql=sql, exec_result=make_exec_result(ok=True))
+    assert not r.ok
+    assert any("Aggregation without GROUP BY" in e for e in r.error)
+
+def test_verifier_parses_valid_sql_ok():
+    v = Verifier()
+    sql = "SELECT COUNT(*), city FROM users GROUP BY city"
+    r = v.run(sql=sql, exec_result=make_exec_result(ok=True))
+    assert r.ok
+    assert r.data == {"verified": True}
+    assert isinstance(r.trace, StageTrace)