Spaces:
Runtime error
Runtime error
File size: 2,674 Bytes
0cac9cf | 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 | """Tests for the Chainlit result formatter.
Author: mohamedgamal04
"""
import unittest
from pathlib import Path
from queryquest.core.models import EngineResult, StatementResult, WritebackTarget
from queryquest.web.render import format_result_markdown
class FormatResultMarkdownTests(unittest.TestCase):
def test_engine_error(self) -> None:
result = EngineResult(prompt="p", error="Authentication failed")
self.assertIn("Authentication failed", format_result_markdown(result))
def test_select_renders_table(self) -> None:
statement = StatementResult(
sql="SELECT * FROM listings",
kind="select",
columns=["id", "price"],
rows=[{"id": 1, "price": 10}, {"id": 2, "price": 20}],
row_count=2,
)
out = format_result_markdown(EngineResult(prompt="p", statements=[statement]))
self.assertIn("| id | price |", out)
self.assertIn("| 1 | 10 |", out)
def test_empty_join_hint(self) -> None:
statement = StatementResult(sql="SELECT * FROM a JOIN b ON a.x = b.y", kind="select", row_count=0)
self.assertIn("join matched no rows", format_result_markdown(EngineResult(prompt="p", statements=[statement])))
def test_statement_error_shown(self) -> None:
statement = StatementResult(sql="SELECT * FROM read_csv('/etc/passwd')", kind="select", error="'READ_CSV' is not allowed")
out = format_result_markdown(EngineResult(prompt="p", statements=[statement]))
self.assertIn("Refused/failed", out)
self.assertIn("READ_CSV", out)
def test_writeback_saved_note(self) -> None:
statement = StatementResult(sql="UPDATE listings SET price = 9", kind="update", row_count=3)
target = WritebackTarget(
file_path=Path("/data/listings.xlsx"),
sheet_name="Sheet1",
table_name="listings__Sheet1",
affected_rows=3,
)
result = EngineResult(
prompt="p",
statements=[statement],
executed=True,
wrote_back=True,
writeback_targets=[target],
)
out = format_result_markdown(result)
self.assertIn("listings.xlsx", out)
self.assertIn("Saved", out)
def test_pipe_in_cell_is_escaped(self) -> None:
statement = StatementResult(
sql="SELECT * FROM t",
kind="select",
columns=["note"],
rows=[{"note": "a|b"}],
row_count=1,
)
self.assertIn("a\\|b", format_result_markdown(EngineResult(prompt="p", statements=[statement])))
if __name__ == "__main__":
unittest.main()
|