Fix test isolation and add real Tika integration tests

- Clean queue table between tests (DROP + CREATE, not just IF NOT EXISTS)
- Use tuple_row cursor in enqueue tests (matches production cursor type)
- Add 2 integration tests using real Tika server (plain text + HTML)
- All 24 Tika tests pass with PGCATALOG_TIKA_URL set

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: jensens

tests/test_tika_enqueue.py

@@ -5,6 +5,7 @@
 from plone.pgcatalog.processor import TIKA_URL
 from plone.pgcatalog.schema import TEXT_EXTRACTION_QUEUE
 from psycopg.rows import dict_row
+from psycopg.rows import tuple_row
 from tests.conftest import DSN
 from tests.conftest import insert_object
 
@@ -20,10 +21,11 @@
 def pg_conn_with_queue(pg_conn_with_catalog):
     """Database with catalog schema + extraction queue table."""
     conn = pg_conn_with_catalog
-    # Install queue schema
+    # Drop and recreate to ensure clean state between tests
+    conn.execute("DROP TABLE IF EXISTS text_extraction_queue, blob_state CASCADE")
+    conn.commit()
     conn.execute(TEXT_EXTRACTION_QUEUE)
     conn.commit()
-    # Also create blob_state if it doesn't exist
     conn.execute(
         "CREATE TABLE IF NOT EXISTS blob_state ("
         "  zoid BIGINT NOT NULL,"
@@ -101,7 +103,8 @@ def test_enqueue_blob_with_tika_url(self, pg_conn_with_queue):
         # Simulate process() accumulating a candidate
         proc._tika_candidates = [{"zoid": zoid, "content_type": "application/pdf"}]
 
-        with conn.cursor() as cur:
+        # Use tuple_row to match production cursor (zodb-pgjsonb uses default)
+        with conn.cursor(row_factory=tuple_row) as cur:
             proc._enqueue_tika_jobs(cur)
         conn.commit()
 
@@ -124,7 +127,7 @@ def test_no_enqueue_without_blob(self, pg_conn_with_queue):
         proc = CatalogStateProcessor()
         proc._tika_candidates = [{"zoid": zoid, "content_type": "application/pdf"}]
 
-        with conn.cursor() as cur:
+        with conn.cursor(row_factory=tuple_row) as cur:
             proc._enqueue_tika_jobs(cur)
         conn.commit()
 
@@ -145,7 +148,7 @@ def test_idempotent_enqueue(self, pg_conn_with_queue):
         # Enqueue twice
         for _ in range(2):
             proc._tika_candidates = [{"zoid": zoid, "content_type": "application/pdf"}]
-            with conn.cursor() as cur:
+            with conn.cursor(row_factory=tuple_row) as cur:
                 proc._enqueue_tika_jobs(cur)
             conn.commit()
 

tests/test_tika_worker.py

@@ -13,6 +13,7 @@
 from unittest.mock import patch
 from zodb_pgjsonb.schema import HISTORY_FREE_SCHEMA
 
+import os
 import psycopg
 import pytest
 import threading
@@ -295,3 +296,88 @@ def test_shutdown_flag(self):
         assert not worker._shutdown.is_set()
         worker.shutdown()
         assert worker._shutdown.is_set()
+
+
+# ── Integration tests (require real Tika server) ─────────────────────
+
+TIKA_URL = os.environ.get("PGCATALOG_TIKA_URL", "").strip()
+
+
+class TestWorkerIntegration:
+    """End-to-end tests with real Tika server."""
+
+    pytestmark = pytest.mark.skipif(not TIKA_URL, reason="PGCATALOG_TIKA_URL not set")
+
+    def test_extract_plain_text(self, worker_db):
+        """Worker extracts text from a plain text blob via real Tika."""
+        conn = worker_db
+        zoid, tid = 50, 1
+        text_content = b"Important findings about quantum computing research"
+        _insert_object_with_blob(
+            conn,
+            zoid,
+            tid,
+            blob_data=text_content,
+            idx={"Language": "en", "Title": "Research"},
+        )
+        _enqueue_job(conn, zoid, tid, content_type="text/plain")
+
+        # Set initial searchable_text
+        conn.execute(
+            "UPDATE object_state SET searchable_text = "
+            "to_tsvector('english', 'Research') "
+            "WHERE zoid = %(zoid)s",
+            {"zoid": zoid},
+        )
+        conn.commit()
+
+        worker = TikaWorker(dsn=DSN, tika_url=TIKA_URL)
+        result = worker._process_one()
+        assert result is True
+
+        # Verify extraction completed
+        status = _get_queue_status(conn, zoid)
+        assert status["status"] == "done"
+        assert status["error"] is None
+
+        # Verify searchable_text was updated with extracted terms
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT searchable_text::text FROM object_state WHERE zoid = %(zoid)s",
+                {"zoid": zoid},
+            )
+            row = cur.fetchone()
+        tsv_text = row["searchable_text"]
+        assert "quantum" in tsv_text or "comput" in tsv_text
+
+    def test_extract_html_content(self, worker_db):
+        """Worker extracts text from HTML via real Tika."""
+        conn = worker_db
+        zoid, tid = 60, 1
+        html_blob = b"<html><body><h1>PostgreSQL Performance</h1><p>Indexes matter.</p></body></html>"
+        _insert_object_with_blob(conn, zoid, tid, blob_data=html_blob)
+        _enqueue_job(conn, zoid, tid, content_type="text/html")
+
+        conn.execute(
+            "UPDATE object_state SET searchable_text = ''::tsvector, "
+            'idx = \'{"Language": "en"}\'::jsonb '
+            "WHERE zoid = %(zoid)s",
+            {"zoid": zoid},
+        )
+        conn.commit()
+
+        worker = TikaWorker(dsn=DSN, tika_url=TIKA_URL)
+        worker._process_one()
+
+        status = _get_queue_status(conn, zoid)
+        assert status["status"] == "done"
+
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT searchable_text::text FROM object_state WHERE zoid = %(zoid)s",
+                {"zoid": zoid},
+            )
+            row = cur.fetchone()
+        tsv_text = row["searchable_text"]
+        # Tika should have extracted "PostgreSQL Performance" and "Indexes matter"
+        assert "postgresql" in tsv_text or "perform" in tsv_text or "index" in tsv_text