fix: test_traverse_blobs_never_raises skips the meta-parse error path

Author: bradjin8

tests/test_blob_parsing_fuzz.py

@@ -24,9 +24,9 @@
 from models import Bubble, SchemaError
 from utils.cli_chat_reader import (
     classify_blob_data,
+    extract_blob_refs,
     messages_to_bubbles,
     traverse_blobs,
-    _extract_blob_refs,  # internal helper; covered directly alongside classify_blob_data
 )
 from utils.text_extract import extract_text_from_bubble
 
@@ -132,17 +132,39 @@ def _make_meta_value(meta: dict) -> str:
 
 
 def _build_store_db_raw(path: str, meta: dict, blobs: dict[str, bytes]) -> None:
-    """Minimal store.db with arbitrary blob payloads (for traverse_blobs fuzz)."""
+    """Minimal store.db with well-formed meta dict and arbitrary blob payloads."""
+    _build_store_db_meta_row(path, _make_meta_value(meta), blobs)
+
+
+def _build_store_db_meta_row(
+    path: str, meta_row: str | None, blobs: dict[str, bytes]
+) -> None:
+    """Minimal store.db; *meta_row* is the raw ``meta.value`` (hex JSON or adversarial)."""
     conn = sqlite3.connect(path)
     conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)")
     conn.execute("CREATE TABLE blobs (id TEXT PRIMARY KEY, data BLOB)")
-    conn.execute("INSERT INTO meta VALUES ('0', ?)", (_make_meta_value(meta),))
+    if meta_row is not None:
+        conn.execute("INSERT INTO meta VALUES ('0', ?)", (meta_row,))
     for blob_id, data in blobs.items():
         conn.execute("INSERT INTO blobs VALUES (?, ?)", (blob_id, data))
     conn.commit()
     conn.close()
 
 
+_FUZZ_META_ROW = st.one_of(
+    st.none(),
+    st.just(""),
+    st.text(min_size=0, max_size=200),
+    st.dictionaries(st.text(max_size=20), _JSON_VALUES, max_size=6).map(
+        lambda d: json.dumps(d).encode("utf-8").hex()
+    ),
+    st.builds(
+        lambda root: _make_meta_value({"latestRootBlobId": root, "createdAt": 1}),
+        _BLOB_ID_HEX,
+    ),
+)
+
+
 def _assemble_workspace_bubble(bubble_id: object, value: object) -> dict | None:
     """Mirror workspace_tabs KV bubble load (json.loads → Bubble.from_dict).
 
@@ -219,7 +241,7 @@ class TestBlobChainParsingFuzz(unittest.TestCase):
     @settings(max_examples=120, deadline=None)
     def test_extract_blob_refs_never_raises(self, data: bytes) -> None:
         try:
-            refs = _extract_blob_refs(data)
+            refs = extract_blob_refs(data)
         except Exception as exc:
             self.fail(f"unexpected {type(exc).__name__}: {exc}")
         self.assertIsInstance(refs, list)
@@ -230,7 +252,7 @@ def test_extract_blob_refs_never_raises(self, data: bytes) -> None:
     @given(data=st.binary(max_size=4096))
     @settings(max_examples=80, deadline=None)
     def test_extract_blob_refs_is_idempotent(self, data: bytes) -> None:
-        self.assertEqual(_extract_blob_refs(data), _extract_blob_refs(data))
+        self.assertEqual(extract_blob_refs(data), extract_blob_refs(data))
 
     @given(data=st.binary(max_size=4096))
     @settings(max_examples=80, deadline=None)
@@ -273,6 +295,23 @@ def test_traverse_blobs_never_raises(
                 self.fail(f"traverse_blobs raised {type(exc).__name__}: {exc}")
             self.assertIsInstance(messages, list)
 
+    @given(meta_row=_FUZZ_META_ROW)
+    @settings(
+        max_examples=30,
+        deadline=None,
+        suppress_health_check=[HealthCheck.too_slow],
+    )
+    def test_traverse_blobs_meta_parse_never_raises(self, meta_row: str | None) -> None:
+        """Covers meta decode / CliSessionMeta.from_dict failure → return [] (no crash)."""
+        with tempfile.TemporaryDirectory() as td:
+            db_path = os.path.join(td, "store.db")
+            _build_store_db_meta_row(db_path, meta_row, {})
+            try:
+                messages = traverse_blobs(db_path)
+            except Exception as exc:
+                self.fail(f"traverse_blobs raised {type(exc).__name__}: {exc}")
+            self.assertIsInstance(messages, list)
+
 
 class TestTextExtractionFuzz(unittest.TestCase):
     @given(bubble=_BUBBLE_LIKE)

tests/test_cli_chat_reader.py

@@ -20,7 +20,7 @@
 
 from utils.cli_chat_reader import (
     _content_to_text,
-    _extract_blob_refs,
+    extract_blob_refs,
     _extract_tool_calls,
     _strip_user_info,
     aggregate_session_stats,
@@ -75,34 +75,34 @@ def _build_store_db(path: str, meta: dict, json_blobs: dict[str, dict], chain: d
 
 
 # ---------------------------------------------------------------------------
-# _extract_blob_refs
+# extract_blob_refs
 # ---------------------------------------------------------------------------
 
 class TestExtractBlobRefs(unittest.TestCase):
     def test_empty_bytes_returns_empty(self):
-        self.assertEqual(_extract_blob_refs(b""), [])
+        self.assertEqual(extract_blob_refs(b""), [])
 
     def test_single_ref(self):
         ref = "a" * 64  # 32 bytes as hex
         raw = b"\x0a\x20" + bytes.fromhex(ref)
-        self.assertEqual(_extract_blob_refs(raw), [ref])
+        self.assertEqual(extract_blob_refs(raw), [ref])
 
     def test_two_refs(self):
         ref1 = "a" * 64
         ref2 = "b" * 64
         raw = b"\x0a\x20" + bytes.fromhex(ref1) + b"\x0a\x20" + bytes.fromhex(ref2)
-        self.assertEqual(_extract_blob_refs(raw), [ref1, ref2])
+        self.assertEqual(extract_blob_refs(raw), [ref1, ref2])
 
     def test_noise_bytes_ignored(self):
         ref = "c" * 64
         noise = b"\x00\xff\x01\x02\x03\x04"
         raw = noise + b"\x0a\x20" + bytes.fromhex(ref) + b"\xde\xad"
-        self.assertIn(ref, _extract_blob_refs(raw))
+        self.assertIn(ref, extract_blob_refs(raw))
 
     def test_partial_tag_at_end_ignored(self):
         # Only 0x0a without 0x20 immediately following should not produce a ref.
         raw = b"\x0a" + b"\x00" * 32
-        self.assertEqual(_extract_blob_refs(raw), [])
+        self.assertEqual(extract_blob_refs(raw), [])
 
 
 # ---------------------------------------------------------------------------

tests/test_models.py

@@ -17,7 +17,7 @@
     Workspace,
     WorkspaceLocalComposer,
 )
-from utils.cli_chat_reader import _extract_blob_refs
+from utils.cli_chat_reader import extract_blob_refs
 
 
 GOOD_COMPOSER_RAW: dict = {
@@ -252,7 +252,7 @@ def test_meta_parses_then_blob_chain_extracts_refs(self) -> None:
         self.assertEqual(meta.latest_root_blob_id, ref1)
 
         chain_blob = _make_blob_chain(ref1, ref2, ref3)
-        refs = _extract_blob_refs(chain_blob)
+        refs = extract_blob_refs(chain_blob)
         self.assertEqual(refs, [ref1, ref2, ref3])
 
     def test_blob_chain_skips_non_marker_bytes(self) -> None:
@@ -261,10 +261,10 @@ def test_blob_chain_skips_non_marker_bytes(self) -> None:
         garbage_after = b"\xff\xfe"
         raw = garbage_before + bytes([0x0A, 0x20]) + bytes.fromhex(ref) + garbage_after
 
-        self.assertEqual(_extract_blob_refs(raw), [ref])
+        self.assertEqual(extract_blob_refs(raw), [ref])
 
     def test_blob_chain_empty_returns_empty_list(self) -> None:
-        self.assertEqual(_extract_blob_refs(b""), [])
+        self.assertEqual(extract_blob_refs(b""), [])
 
 
 if __name__ == "__main__":

utils/cli_chat_reader.py

@@ -62,7 +62,7 @@ def _read_meta(db_path: str) -> dict:
     return {}
 
 
-def _extract_blob_refs(data: bytes) -> list[str]:
+def extract_blob_refs(data: bytes) -> list[str]:
     """Extract all 32-byte (SHA-256) blob references from a binary chain node.
 
     The encoding is: tag ``0x0a`` (field 1, length-delimited) followed by
@@ -84,7 +84,7 @@ def classify_blob_data(data: bytes) -> tuple[dict | None, list[str]]:
 
     Returns ``(message_dict, [])`` when *data* decodes to a dict with a
     ``role`` field; otherwise ``(None, refs)`` where *refs* are SHA-256 hex
-    ids from :func:`_extract_blob_refs`.  Used by :func:`traverse_blobs` and
+    ids from :func:`extract_blob_refs`.  Used by :func:`traverse_blobs` and
     property tests — keep in sync when the load loop changes.
     """
     try:
@@ -93,7 +93,7 @@ def classify_blob_data(data: bytes) -> tuple[dict | None, list[str]]:
             return msg, []
     except (UnicodeDecodeError, json.JSONDecodeError):
         pass
-    return None, _extract_blob_refs(data)
+    return None, extract_blob_refs(data)
 
 
 def traverse_blobs(db_path: str) -> list[dict]: