mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
c49e77d059
* fix(ingest): import sqlalchemy text where DBIngestor and DataExporter use it sqlalchemy.text was imported function-locally in DatabaseConnector.connect and test_connection, but called in DataExporter.export_table_data and DBIngestor.execute_query, which never imported it. Both raised NameError, re-wrapped by their except handlers into a ProcessingError reading 'Failed to execute query: name text is not defined' -- a message that looks like a database fault rather than a missing import. No test exercised either method, so this also repairs a pre-existing failure in tests/ingest/test_notebook_02.py::test_08_database_ingestion. Add SQLite-backed coverage for all three call sites, including the SELECT COUNT(*) branch that only runs when no limit is passed and would otherwise stay untested. Closes #1015 * test(ingest): register setUp cleanups with addCleanup TemporaryDirectory and the SQLAlchemy engine were released only in tearDown, which unittest skips when setUp raises partway through. Register each cleanup as soon as its resource exists so a failed setUp still disposes the engine and removes the temp directory. LIFO ordering keeps dispose before cleanup, as tearDown had it. --------- Co-authored-by: Pravit Ampapathini <pravitampapathini@Pravits-MacBook-Air-3.local> Co-authored-by: Pravit Ampapathini <pravit.amp@gmail.com>
113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
"""Query-execution coverage for DBIngestor and DataExporter.
|
|
|
|
Regression tests for #1015: ``sqlalchemy.text`` was imported function-locally inside
|
|
``DatabaseConnector.connect()`` and ``DatabaseConnector.test_connection()``, but called
|
|
in ``DataExporter.export_table_data()`` and ``DBIngestor.execute_query()``, which never
|
|
imported it. Both raised ``NameError``, re-wrapped by their ``except Exception`` handlers
|
|
into a ``ProcessingError`` reading "Failed to execute query: name 'text' is not defined"
|
|
-- a message that looks like a database problem rather than a missing import.
|
|
|
|
Nothing caught it because no test exercised either method; the only ``execute_query``
|
|
references under ``tests/`` are Mock stand-ins for the unrelated graph-store method of
|
|
the same name.
|
|
|
|
These tests run against a temporary SQLite database, so they need no external service.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
try:
|
|
from sqlalchemy import create_engine, text
|
|
|
|
SQLALCHEMY_AVAILABLE = True
|
|
except ImportError: # pragma: no cover - exercised only where sqlalchemy is absent
|
|
SQLALCHEMY_AVAILABLE = False
|
|
|
|
from semantica.ingest.db_ingestor import DataExporter, DBIngestor
|
|
|
|
|
|
@unittest.skipUnless(SQLALCHEMY_AVAILABLE, "sqlalchemy is required for these tests")
|
|
class TestDBIngestorQueryExecution(unittest.TestCase):
|
|
"""Both query paths must survive the call that needed sqlalchemy.text -- see #1015."""
|
|
|
|
def setUp(self):
|
|
# Register each cleanup as soon as the resource exists: tearDown is not
|
|
# called when setUp raises partway through, but addCleanup callbacks are.
|
|
self._tmpdir = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self._tmpdir.cleanup)
|
|
self.db_path = os.path.join(self._tmpdir.name, "test.db")
|
|
self.connection_string = f"sqlite:///{self.db_path}"
|
|
self.engine = create_engine(self.connection_string)
|
|
self.addCleanup(self.engine.dispose)
|
|
|
|
with self.engine.begin() as conn:
|
|
conn.execute(text("CREATE TABLE widgets (id INTEGER, name TEXT)"))
|
|
for row_id, name in [(1, "alpha"), (2, "beta"), (3, "gamma")]:
|
|
conn.execute(
|
|
text("INSERT INTO widgets VALUES (:id, :name)"),
|
|
{"id": row_id, "name": name},
|
|
)
|
|
|
|
def test_execute_query_returns_rows(self):
|
|
"""DBIngestor.execute_query -- the text() call that raised NameError."""
|
|
rows = DBIngestor().execute_query(
|
|
self.connection_string,
|
|
"SELECT id, name FROM widgets ORDER BY id",
|
|
)
|
|
self.assertEqual(
|
|
rows,
|
|
[
|
|
{"id": 1, "name": "alpha"},
|
|
{"id": 2, "name": "beta"},
|
|
{"id": 3, "name": "gamma"},
|
|
],
|
|
)
|
|
|
|
def test_execute_query_binds_parameters(self):
|
|
"""The params argument is passed alongside text(), so cover it explicitly."""
|
|
rows = DBIngestor().execute_query(
|
|
self.connection_string,
|
|
"SELECT name FROM widgets WHERE id = :wanted",
|
|
wanted=2,
|
|
)
|
|
self.assertEqual(rows, [{"name": "beta"}])
|
|
|
|
def test_export_table_data_with_limit(self):
|
|
"""Exercises the main text() call; the COUNT(*) branch is skipped when limit is set."""
|
|
result = DataExporter().export_table_data(self.engine, "widgets", limit=2)
|
|
|
|
self.assertEqual(result.table_name, "widgets")
|
|
self.assertEqual(len(result.rows), 2)
|
|
self.assertEqual(result.row_count, 2)
|
|
self.assertEqual([c["name"] for c in result.columns], ["id", "name"])
|
|
|
|
def test_export_table_data_without_limit_counts_rows(self):
|
|
"""Covers the second text() call.
|
|
|
|
``export_table_data`` only issues its ``SELECT COUNT(*)`` when no ``limit`` is
|
|
passed, so the test above never reaches that line. Without this case one of the
|
|
three call sites the bug touched would stay untested.
|
|
"""
|
|
result = DataExporter().export_table_data(self.engine, "widgets")
|
|
|
|
self.assertEqual(len(result.rows), 3)
|
|
self.assertEqual(result.row_count, 3)
|
|
|
|
def test_export_table_data_honors_where_and_order(self):
|
|
"""The WHERE/ORDER BY clauses are interpolated before text() wraps the query."""
|
|
result = DataExporter().export_table_data(
|
|
self.engine,
|
|
"widgets",
|
|
where="id >= 2",
|
|
order_by="id DESC",
|
|
)
|
|
|
|
self.assertEqual([r["name"] for r in result.rows], ["gamma", "beta"])
|
|
self.assertEqual(result.row_count, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|