From c49e77d059387e97319723c6d578561755044561 Mon Sep 17 00:00:00 2001 From: pravit-amp <43916793+pravit-amp@users.noreply.github.com> Date: Wed, 26 Aug 2026 10:10:40 -0700 Subject: [PATCH] fix(ingest): import sqlalchemy text where DBIngestor and DataExporter use it (#1017) * 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 Co-authored-by: Pravit Ampapathini --- semantica/ingest/db_ingestor.py | 2 + tests/ingest/test_db_ingestor_query.py | 112 +++++++++++++++++++++++++ 2 files changed, 114 insertions(+) create mode 100644 tests/ingest/test_db_ingestor_query.py diff --git a/semantica/ingest/db_ingestor.py b/semantica/ingest/db_ingestor.py index 351238d3..0d656dee 100644 --- a/semantica/ingest/db_ingestor.py +++ b/semantica/ingest/db_ingestor.py @@ -815,6 +815,8 @@ class DBIngestor: engine = connector.connect(connection_string) try: + from sqlalchemy import text + with engine.connect() as conn: # Execute query with parameters (parameterized queries for safety) result = conn.execute(text(query), params) diff --git a/tests/ingest/test_db_ingestor_query.py b/tests/ingest/test_db_ingestor_query.py new file mode 100644 index 00000000..4c2fcfa3 --- /dev/null +++ b/tests/ingest/test_db_ingestor_query.py @@ -0,0 +1,112 @@ +"""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()