fix(040): update all tests for Parquet-only pipeline

- test_staging.py: uses write_staged_batch + iter_staged (no write_staged_entity)
- test_embeddings.py: relaxed assertions for comprehensive embedding text
- test_pipeline_e2e.py: counts from ParquetStore (not YAML glob)

T007, T009, T014, T019 marked done. 26/33 tasks complete.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: satra

library/tests/test_embeddings.py

@@ -23,7 +23,8 @@ def test_build_element_text_full():
             {"class": "Subject", "name": "age", "description": "Age of the subject in years"}
         ]
     }
-    assert _build_element_text(data) == "Subject age: Age of the subject in years"
+    text = _build_element_text(data)
+    assert "Subject age: Age of the subject in years" in text
 
 
 def test_build_element_text_no_description():
@@ -177,4 +178,5 @@ def test_build_element_embeddings_from_files(tmp_path):
 
     # We can test the text construction without needing the model
     data = yaml.safe_load((elements_dir / "age_abc123.yaml").read_text())
-    assert _build_element_text(data) == "Subject age: Age"
+    text = _build_element_text(data)
+    assert "Subject age: Age" in text

library/tests/test_pipeline_e2e.py

@@ -24,16 +24,15 @@ def _run_pipeline(source: str, output_dir: Path, skip_enrich: bool = True) -> di
     if not skip_enrich:
         enrich_elements(staging_dir=staging)
     commit_stats = commit_staged(staging, output_dir)
+    from undata_library.storage.parquet_store import ParquetStore
+
+    store = ParquetStore(output_dir)
     return {
         "ingest": stats,
         "commit": commit_stats,
-        "elements": len(list((output_dir / "elements").glob("*.yaml"))),
-        "schemas": len(list((output_dir / "schemas").glob("*.yaml")))
-        if (output_dir / "schemas").exists()
-        else 0,
-        "values": len(list((output_dir / "values").glob("*.yaml")))
-        if (output_dir / "values").exists()
-        else 0,
+        "elements": store.count("elements"),
+        "schemas": store.count("schemas"),
+        "values": store.count("values"),
     }
 
 
@@ -67,7 +66,7 @@ class TestNewEntityFlow:
     def test_synthetic_element_committed(self, tmp_path):
         # First run: normal BIDS extraction
         _run_pipeline("bids", tmp_path)
-        initial_count = len(list((tmp_path / "elements").glob("*.yaml")))
+        initial_count = ParquetStore(tmp_path).count("elements")
 
         # Add a synthetic element to a new staging run
         run_id = generate_run_id()
@@ -81,7 +80,7 @@ def test_synthetic_element_committed(self, tmp_path):
         )
         commit_staged(staging, tmp_path)
 
-        final_count = len(list((tmp_path / "elements").glob("*.yaml")))
+        final_count = ParquetStore(tmp_path).count("elements")
         assert final_count == initial_count + 1
 
 
@@ -100,7 +99,7 @@ def test_double_ingest_merges_not_duplicates(self, tmp_path):
     def test_entity_level_dedup(self, tmp_path):
         """T038e: Ingesting a duplicate entity merges provenance, not duplicates."""
         _run_pipeline("bids", tmp_path)
-        initial = len(list((tmp_path / "elements").glob("*.yaml")))
+        initial = ParquetStore(tmp_path).count("elements")
 
         # Create a second staging with same content + different provenance
         from undata_library.staging import create_staging_dir, generate_run_id
@@ -115,7 +114,7 @@ def test_entity_level_dedup(self, tmp_path):
         )
         commit_staged(staging, tmp_path)
 
-        final = len(list((tmp_path / "elements").glob("*.yaml")))
+        final = ParquetStore(tmp_path).count("elements")
         # Should have merged into existing, not created a new one
         # (or created exactly 1 if TaskName didn't exist before)
         assert final <= initial + 1

library/tests/test_staging.py

@@ -1,12 +1,14 @@
-"""Tests for staging directory management."""
+"""Tests for staging directory management (Parquet-only)."""
 
 import time
 
 from undata_library.staging import (
     cleanup_stale_staging,
+    count_staged,
     create_staging_dir,
     generate_run_id,
-    write_staged_entity,
+    iter_staged,
+    write_staged_batch,
 )
 
 
@@ -21,35 +23,53 @@ def test_create_staging_dir(tmp_path):
     staging = create_staging_dir(tmp_path, rid)
     assert staging.exists()
     assert (staging / "elements").exists()
-    assert (staging / "schemas").exists()
-    assert (staging / "values").exists()
-    assert (staging / "valuesets").exists()
 
 
-def test_write_staged_entity(tmp_path):
+def test_write_staged_batch(tmp_path):
     staging = create_staging_dir(tmp_path, "test-run")
-    data = {
-        "semantic": {"data_type": "string"},
-        "provenance": [{"source": "test", "class": "X", "name": "field"}],
-    }
-    path = write_staged_entity(staging, "elements", data)
-    assert path.exists()
-    assert path.parent.name == "elements"
-    assert path.suffix == ".yaml"
-    # UUID filename (not content-addressed)
-    assert len(path.stem) == 36  # UUID length
+    entities = [
+        {
+            "semantic": {"data_type": "string"},
+            "provenance": [{"source": "test", "class": "X", "name": "field1"}],
+        },
+        {
+            "semantic": {"data_type": "integer"},
+            "provenance": [{"source": "test", "class": "X", "name": "field2"}],
+        },
+    ]
+    written = write_staged_batch(staging, "elements", entities, source="test")
+    assert written == 2
+
+
+def test_count_staged(tmp_path):
+    staging = create_staging_dir(tmp_path, "test-run")
+    write_staged_batch(
+        staging,
+        "elements",
+        [{"semantic": {"data_type": "string"}, "provenance": [{"source": "t", "class": "X", "name": "a"}]}],
+        source="test",
+    )
+    assert count_staged(staging, "elements") == 1
+
+
+def test_iter_staged(tmp_path):
+    staging = create_staging_dir(tmp_path, "test-run")
+    write_staged_batch(
+        staging,
+        "elements",
+        [{"semantic": {"data_type": "string"}, "provenance": [{"source": "t", "class": "X", "name": "a"}]}],
+        source="test",
+    )
+    entities = list(iter_staged(staging, "elements"))
+    assert len(entities) == 1
 
 
 def test_cleanup_stale_staging(tmp_path):
-    # Create an "old" staging dir
-    old = create_staging_dir(tmp_path, "old-run")
-    # Backdate its mtime
     import os
 
-    old_time = time.time() - (25 * 3600)  # 25 hours ago
+    old = create_staging_dir(tmp_path, "old-run")
+    old_time = time.time() - (25 * 3600)
     os.utime(old, (old_time, old_time))
-
-    # Create a "new" staging dir
     create_staging_dir(tmp_path, "new-run")
 
     removed = cleanup_stale_staging(tmp_path, max_age_hours=24)

specs/040-unified-embedding-storage/tasks.md

@@ -26,9 +26,9 @@
 - [X] T004 [US5] Rewrite FileBackend to delegate all entity operations to ParquetStore (remove YAML read/write) in library/src/undata_library/storage/file_backend.py
 - [X] T005 [US5] Remove FileEntityStore class (YAML-backed entity store) from library/src/undata_library/storage/file_backend.py
 - [X] T006 [US5] Update StorageBackend protocol — remove write_batch/read_batch (ParquetStore handles natively) in library/src/undata_library/storage/protocol.py
-- [ ] T007 [US5] Update all pipeline callers to use ParquetStore directly: enrich.py, align.py, alias_detection.py, transform.py in library/src/undata_library/
+- [X] T007 [US5] Update all pipeline callers to use ParquetStore directly: enrich.py, align.py, alias_detection.py, transform.py in library/src/undata_library/
 - [X] T008 [US5] Remove iter_staged YAML path and write_staged_entity from library/src/undata_library/staging.py
-- [ ] T009 [US5] Run full test suite — verify no tests depend on YAML entity files in library/tests/
+- [X] T009 [US5] Run full test suite — verify no tests depend on YAML entity files in library/tests/
 
 **Checkpoint**: All entity access goes through ParquetStore; no YAML entity I/O remains
 
@@ -44,7 +44,7 @@
 - [X] T011 [US1] Rewrite commit_staged to read from ParquetStore, compute sha256, write committed entities to ParquetStore (no YAML output) in library/src/undata_library/commit.py
 - [X] T012 [US1] Rewrite cross-reference resolution (_resolve_cross_references) to operate on DataFrames from ParquetStore in library/src/undata_library/commit.py
 - [X] T013 [US1] Remove yaml.dump calls from commit path and yaml_to_parquet conversion step in library/src/undata_library/commit.py and library/src/undata_library/staging.py
-- [ ] T014 [US1] Update batch_ingest to use ParquetStore throughout (no YAML intermediaries) in library/src/undata_library/ingest.py
+- [X] T014 [US1] Update batch_ingest to use ParquetStore throughout (no YAML intermediaries) in library/src/undata_library/ingest.py
 - [ ] T015 [US1] Add test: pipeline run produces zero YAML entity files in library/tests/test_parquet_pipeline.py
 
 **Checkpoint**: `find output -name "*.yaml" -not -path "*/runs/*"` returns 0
@@ -60,7 +60,7 @@
 - [X] T016 [US2] Create build_comprehensive_embedding_text() that uses name + description + type + unit + annotations + provenance in library/src/undata_library/embeddings.py
 - [X] T017 [US2] Add embedding computation step to commit_staged — after sha256 computation, compute embeddings for all entities in batch in library/src/undata_library/commit.py
 - [X] T018 [US2] Ensure all entity types (elements, schemas, values, valuesets) get embeddings at commit in library/src/undata_library/commit.py
-- [ ] T019 [US2] Update backend import to skip embedding model loading when all entities have pre-computed embeddings in backend/src/storage/database_backend.py
+- [X] T019 [US2] Update backend import to skip embedding model loading when all entities have pre-computed embeddings in backend/src/storage/database_backend.py
 - [ ] T020 [US2] Update backend import_service to read Parquet-only (remove YAML import path) in backend/src/services/import_service.py
 - [ ] T021 [US2] Add test: committed entity has embedding field with 384 floats in library/tests/test_parquet_pipeline.py
 - [ ] T022 [US2] Add test: backend import of 100 entities with embeddings completes without loading sentence-transformers model in backend/tests/test_import.py