fix: test failure with bubble none and pytest missing

bradjin8 · bradjin8 · commit d23fe7188363 · 2026-05-25T11:26:32.000-04:00
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
@@ -116,7 +116,7 @@ jobs:
         # new endpoint file because `pytest tests/` would also re-collect the
         # 178 unittest.TestCase subclasses already run in the step above —
         # ~2× the CI minutes for zero extra signal.
-        run: python -m pytest tests/test_api_endpoints.py tests/test_parse_failure_logging.py -v --tb=short
+        run: python -m pytest tests/test_api_endpoints.py -v --tb=short
 
       # ── PyInstaller desktop build (Windows only, once per workflow) ────────
       # Closes #44. Builds the onedir bundle and smoke-tests --help so the
diff --git a/services/workspace_tabs.py b/services/workspace_tabs.py
@@ -114,6 +114,8 @@ def _safe_fetchall(query: str, params: tuple = ()) -> list:
             parts = row["key"].split(":")
             if len(parts) >= 3:
                 bid = parts[2]
+                if row["value"] is None:
+                    continue
                 try:
                     parsed = json.loads(row["value"])
                 except json.JSONDecodeError as e:
@@ -192,6 +194,8 @@ def _safe_fetchall(query: str, params: tuple = ()) -> list:
 
         for row in composer_rows:
             composer_id = row["key"].split(":")[1]
+            if row["value"] is None:
+                continue
             try:
                 parsed = json.loads(row["value"])
             except json.JSONDecodeError as e:
diff --git a/tests/test_parse_failure_logging.py b/tests/test_parse_failure_logging.py
@@ -1,17 +1,15 @@
-"""pytest caplog tests for structured logging at model parse sites (issue #66)."""
+"""Tests for structured logging at model parse sites (issue #66)."""
 
 from __future__ import annotations
 
 import json
-import logging
 import os
 import sqlite3
 import sys
 import tempfile
+import unittest
 from contextlib import closing
 
-import pytest
-
 REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 if REPO_ROOT not in sys.path:
     sys.path.insert(0, REPO_ROOT)
@@ -36,7 +34,6 @@ def _seed_listing_with_drifted_composer(parent: str) -> str:
 
     conn = sqlite3.connect(os.path.join(global_root, "state.vscdb"))
     conn.execute("CREATE TABLE cursorDiskKV ([key] TEXT PRIMARY KEY, value TEXT)")
-    # Missing createdAt — Composer.from_dict raises SchemaError.
     conn.execute(
         "INSERT INTO cursorDiskKV VALUES (?, ?)",
         (
@@ -85,7 +82,6 @@ def _seed_tabs_with_drifted_bubble(parent: str) -> str:
             }),
         ),
     )
-    # Non-dict bubble value trips Bubble.from_dict schema gate.
     conn.execute(
         "INSERT INTO cursorDiskKV VALUES (?, ?)",
         ("bubbleId:cmp-ok:b-bad", json.dumps("not-a-dict")),
@@ -99,102 +95,97 @@ def _seed_tabs_with_drifted_bubble(parent: str) -> str:
     return ws_root
 
 
-@pytest.fixture
-def caplog_at_warning(caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
-    caplog.set_level(logging.WARNING)
-    return caplog
-
-
-def test_listing_logs_composer_schema_drift(caplog_at_warning: pytest.LogCaptureFixture) -> None:
-    with tempfile.TemporaryDirectory() as tmp:
-        ws_root = _seed_listing_with_drifted_composer(tmp)
-        with caplog_at_warning.at_level(logging.WARNING, logger="services.workspace_listing"):
-            list_workspace_projects(ws_root, rules=[])
+class TestParseFailureLogging(unittest.TestCase):
+    def test_listing_logs_composer_schema_drift(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            ws_root = _seed_listing_with_drifted_composer(tmp)
+            with self.assertLogs("services.workspace_listing", level="WARNING") as cm:
+                list_workspace_projects(ws_root, rules=[])
 
-    messages = [r.getMessage() for r in caplog_at_warning.records]
-    assert any("Composer" in m and "cmp-drift" in m for m in messages), (
-        f"expected Composer parse warning for cmp-drift, got: {messages}"
-    )
+        messages = [r.getMessage() for r in cm.records]
+        self.assertTrue(
+            any("Composer" in m and "cmp-drift" in m for m in messages),
+            f"expected Composer parse warning for cmp-drift, got: {messages}",
+        )
 
+    def test_workspace_tabs_logs_bubble_json_decode_failure(self) -> None:
+        from flask import Flask
+
+        app = Flask(__name__)
+        app.config["TESTING"] = True
+        app.config["EXCLUSION_RULES"] = []
+
+        with tempfile.TemporaryDirectory() as tmp:
+            ws_root = _seed_tabs_with_drifted_bubble(tmp)
+            global_db = os.path.join(tmp, "globalStorage", "state.vscdb")
+            with closing(sqlite3.connect(global_db)) as conn:
+                conn.execute(
+                    "INSERT INTO cursorDiskKV ([key], value) VALUES (?, ?)",
+                    ("bubbleId:cmp-ok:b-json", "{not valid json"),
+                )
+                conn.commit()
+            with self.assertLogs("services.workspace_tabs", level="WARNING") as cm:
+                with app.test_request_context("/api/workspaces/global/tabs"):
+                    payload, status = assemble_workspace_tabs("global", ws_root, rules=[])
+
+        self.assertEqual(status, 200)
+        messages = [r.getMessage() for r in cm.records]
+        self.assertTrue(
+            any("decode Bubble" in m and "b-json" in m for m in messages),
+            f"expected JSON decode warning for b-json, got: {messages}",
+        )
 
-def test_workspace_tabs_logs_bubble_json_decode_failure(
-    caplog_at_warning: pytest.LogCaptureFixture,
-) -> None:
-    from flask import Flask
+    def test_workspace_tabs_logs_composer_json_decode_failure(self) -> None:
+        from flask import Flask
 
-    app = Flask(__name__)
-    app.config["TESTING"] = True
-    app.config["EXCLUSION_RULES"] = []
+        app = Flask(__name__)
+        app.config["TESTING"] = True
+        app.config["EXCLUSION_RULES"] = []
 
-    with tempfile.TemporaryDirectory() as tmp:
-        ws_root = _seed_tabs_with_drifted_bubble(tmp)
-        global_db = os.path.join(tmp, "globalStorage", "state.vscdb")
-        with closing(sqlite3.connect(global_db)) as conn:
-            conn.execute(
-                "INSERT INTO cursorDiskKV ([key], value) VALUES (?, ?)",
-                ("bubbleId:cmp-ok:b-json", "{not valid json"),
+        with tempfile.TemporaryDirectory() as tmp:
+            ws_root = _seed_tabs_with_drifted_bubble(tmp)
+            global_db = os.path.join(tmp, "globalStorage", "state.vscdb")
+            bad_composer_value = (
+                '{"fullConversationHeadersOnly": [{"bubbleId": "b1"}], "createdAt":'
             )
-            conn.commit()
-        with caplog_at_warning.at_level(logging.WARNING, logger="services.workspace_tabs"):
-            with app.test_request_context("/api/workspaces/global/tabs"):
-                payload, status = assemble_workspace_tabs("global", ws_root, rules=[])
-
-    assert status == 200
-    messages = [r.getMessage() for r in caplog_at_warning.records]
-    assert any("decode Bubble" in m and "b-json" in m for m in messages), (
-        f"expected JSON decode warning for b-json, got: {messages}"
-    )
-
-
-def test_workspace_tabs_logs_composer_json_decode_failure(
-    caplog_at_warning: pytest.LogCaptureFixture,
-) -> None:
-    from flask import Flask
-
-    app = Flask(__name__)
-    app.config["TESTING"] = True
-    app.config["EXCLUSION_RULES"] = []
-
-    with tempfile.TemporaryDirectory() as tmp:
-        ws_root = _seed_tabs_with_drifted_bubble(tmp)
-        global_db = os.path.join(tmp, "globalStorage", "state.vscdb")
-        # Value must match composer_rows LIKE '%fullConversationHeadersOnly%' to reach parse.
-        bad_composer_value = (
-            '{"fullConversationHeadersOnly": [{"bubbleId": "b1"}], "createdAt":'
+            with closing(sqlite3.connect(global_db)) as conn:
+                conn.execute(
+                    "INSERT INTO cursorDiskKV ([key], value) VALUES (?, ?)",
+                    ("composerData:cmp-json", bad_composer_value),
+                )
+                conn.commit()
+            with self.assertLogs("services.workspace_tabs", level="WARNING") as cm:
+                with app.test_request_context("/api/workspaces/global/tabs"):
+                    payload, status = assemble_workspace_tabs("global", ws_root, rules=[])
+
+        self.assertEqual(status, 200)
+        messages = [r.getMessage() for r in cm.records]
+        self.assertTrue(
+            any("decode Composer" in m and "cmp-json" in m for m in messages),
+            f"expected JSON decode warning for cmp-json, got: {messages}",
         )
-        with closing(sqlite3.connect(global_db)) as conn:
-            conn.execute(
-                "INSERT INTO cursorDiskKV ([key], value) VALUES (?, ?)",
-                ("composerData:cmp-json", bad_composer_value),
-            )
-            conn.commit()
-        with caplog_at_warning.at_level(logging.WARNING, logger="services.workspace_tabs"):
-            with app.test_request_context("/api/workspaces/global/tabs"):
-                payload, status = assemble_workspace_tabs("global", ws_root, rules=[])
-
-    assert status == 200
-    messages = [r.getMessage() for r in caplog_at_warning.records]
-    assert any("decode Composer" in m and "cmp-json" in m for m in messages), (
-        f"expected JSON decode warning for cmp-json, got: {messages}"
-    )
 
+    def test_workspace_tabs_logs_bubble_schema_drift(self) -> None:
+        from flask import Flask
+
+        app = Flask(__name__)
+        app.config["TESTING"] = True
+        app.config["EXCLUSION_RULES"] = []
+
+        with tempfile.TemporaryDirectory() as tmp:
+            ws_root = _seed_tabs_with_drifted_bubble(tmp)
+            with self.assertLogs("services.workspace_tabs", level="WARNING") as cm:
+                with app.test_request_context("/api/workspaces/global/tabs"):
+                    payload, status = assemble_workspace_tabs("global", ws_root, rules=[])
+
+        self.assertEqual(status, 200)
+        self.assertIn("cmp-ok", [t["id"] for t in payload.get("tabs", [])])
+        messages = [r.getMessage() for r in cm.records]
+        self.assertTrue(
+            any("Bubble" in m and "b-bad" in m for m in messages),
+            f"expected Bubble parse warning for b-bad, got: {messages}",
+        )
 
-def test_workspace_tabs_logs_bubble_schema_drift(caplog_at_warning: pytest.LogCaptureFixture) -> None:
-    from flask import Flask
-
-    app = Flask(__name__)
-    app.config["TESTING"] = True
-    app.config["EXCLUSION_RULES"] = []
-
-    with tempfile.TemporaryDirectory() as tmp:
-        ws_root = _seed_tabs_with_drifted_bubble(tmp)
-        with caplog_at_warning.at_level(logging.WARNING, logger="services.workspace_tabs"):
-            with app.test_request_context("/api/workspaces/global/tabs"):
-                payload, status = assemble_workspace_tabs("global", ws_root, rules=[])
 
-    assert status == 200
-    assert "cmp-ok" in [t["id"] for t in payload.get("tabs", [])]
-    messages = [r.getMessage() for r in caplog_at_warning.records]
-    assert any("Bubble" in m and "b-bad" in m for m in messages), (
-        f"expected Bubble parse warning for b-bad, got: {messages}"
-    )
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_workspace_tabs_null_bubble.py b/tests/test_workspace_tabs_null_bubble.py
@@ -2,8 +2,8 @@
 
 A cursorDiskKV row with a NULL value column previously caused
 json.loads(None) -> TypeError, which propagated as a 500 response.
-The fix uses ``_try_loads_kv_value`` in ``services/workspace_tabs.py`` so
-NULL / unparseable cursorDiskKV values are skipped without raising.
+Bubble rows with NULL or invalid JSON values are skipped in
+``services/workspace_tabs.py`` without raising.
 """
 
 import json
