Add failing tests for squashed current_revision losing VALIDATION_FAILED state

ingestify/__init__.py

@@ -11,4 +11,4 @@
     from .exceptions import StopProcessing
     from .main import debug_source
 
-__version__ = "0.17.0"
+__version__ = "0.17.1"

ingestify/domain/models/dataset/dataset.py

@@ -7,7 +7,7 @@
 from .dataset_state import DatasetState
 from .file import DraftFile
 from .identifier import Identifier
-from .revision import Revision, RevisionSource, SourceType
+from .revision import Revision, RevisionSource, RevisionState, SourceType
 from ..base import BaseModel
 
 
@@ -112,12 +112,28 @@ def current_revision(self) -> Optional[Revision]:
                     files[file_id] = file
                     files[file_id].revision_id = revision.revision_id
 
+            # Derive the squashed state from the revisions that actually
+            # contribute the current files: if any current file still comes from
+            # a VALIDATION_FAILED revision, the squashed dataset holds invalid
+            # data and must report VALIDATION_FAILED (so it gets refetched).
+            revision_state = {r.revision_id: r.state for r in self.revisions}
+            contributing_revision_ids = {file.revision_id for file in files.values()}
+            squashed_state = (
+                RevisionState.VALIDATION_FAILED
+                if any(
+                    revision_state[revision_id] == RevisionState.VALIDATION_FAILED
+                    for revision_id in contributing_revision_ids
+                )
+                else self.revisions[-1].state
+            )
+
             return Revision(
                 revision_id=self.revisions[-1].revision_id,
                 created_at=self.revisions[-1].created_at,
                 # created_at=max([file.modified_at for file in files.values()]),
                 description="Squashed revision",
                 is_squashed=True,
+                state=squashed_state,
                 modified_files=list(files.values()),
                 source=RevisionSource(source_type=SourceType.SQUASHED, source_id=""),
             )

ingestify/tests/test_refetch_validation_failed.py

@@ -1,12 +1,17 @@
 """Tests for invalidating revisions and refetching."""
 from datetime import datetime, timezone
+from pathlib import Path
 
 from ingestify import Source, DatasetResource
 from ingestify.domain import DataSpecVersionCollection, DraftFile, Selector
 from ingestify.domain.models.dataset.collection_metadata import (
     DatasetCollectionMetadata,
 )
-from ingestify.domain.models.dataset.revision import RevisionState
+from ingestify.domain.models.dataset.dataset import Dataset
+from ingestify.domain.models.dataset.dataset_state import DatasetState
+from ingestify.domain.models.dataset.file import File
+from ingestify.domain.models.dataset.identifier import Identifier
+from ingestify.domain.models.dataset.revision import Revision, RevisionState
 from ingestify.domain.models.fetch_policy import FetchPolicy
 from ingestify.domain.models.ingestion.ingestion_plan import IngestionPlan
 
@@ -106,6 +111,53 @@ def test_invalidate_revision_triggers_refetch(engine):
     assert call_count == 2, "Dataset with invalidated revision should be refetched"
 
 
+def test_invalidate_multi_revision_dataset_refetches(engine):
+    """A dataset with 2+ revisions must still refetch after invalidation.
+
+    Regression: Dataset.current_revision squashes multiple revisions into a new
+    Revision without a state, defaulting to PENDING_VALIDATION. This drops the
+    VALIDATION_FAILED state set by invalidate_revisions, so FetchPolicy.should_refetch
+    returns False and the dataset is silently skipped instead of refetched.
+    """
+    global call_count
+    call_count = 0
+    _setup(engine)
+
+    # Build a dataset with TWO revisions: fetch -> invalidate -> refetch.
+    engine.run()  # revision 0
+    assert call_count == 1
+    ds = list(
+        engine.store.get_dataset_collection(
+            provider="test_provider", dataset_type="test"
+        )
+    )[0]
+    engine.store.invalidate_revisions([ds], reason="first")
+    engine.run()  # revision 1 (refetch)
+    assert call_count == 2
+
+    ds = list(
+        engine.store.get_dataset_collection(
+            provider="test_provider", dataset_type="test"
+        )
+    )[0]
+    assert len(ds.revisions) == 2, "precondition: dataset should have two revisions"
+
+    # Invalidate again — every revision is now VALIDATION_FAILED.
+    engine.store.invalidate_revisions([ds], reason="second")
+    ds = list(
+        engine.store.get_dataset_collection(
+            provider="test_provider", dataset_type="test"
+        )
+    )[0]
+    assert (
+        ds.current_revision.state == RevisionState.VALIDATION_FAILED
+    ), "squashed current_revision must preserve the VALIDATION_FAILED state"
+
+    # And the next run must refetch it.
+    engine.run()
+    assert call_count == 3, "invalidated multi-revision dataset should be refetched"
+
+
 def test_invalidate_revisions_batch(engine):
     """invalidate_revisions works on multiple datasets at once."""
     global call_count
@@ -148,3 +200,88 @@ def test_invalidate_revisions_batch(engine):
     # Second run: should refetch all 5
     engine.run()
     assert call_count == 10, "All 5 invalidated datasets should be refetched"
+
+
+# --- Unit tests for the squashed current_revision state -------------------
+
+
+def _file(file_id: str) -> File:
+    return File(
+        file_id=file_id,
+        created_at=FIXED_TIME,
+        modified_at=FIXED_TIME,
+        tag=file_id,
+        size=10,
+        content_type="application/json",
+        data_feed_key=file_id,
+        data_spec_version="v1",
+        data_serialization_format="json",
+        storage_size=10,
+        storage_compression_method=None,
+        storage_path=Path(f"/{file_id}"),
+    )
+
+
+def _revision(revision_id: int, files: list[File], state: RevisionState) -> Revision:
+    return Revision(
+        revision_id=revision_id,
+        created_at=FIXED_TIME,
+        description="",
+        modified_files=files,
+        source=None,
+        state=state,
+    )
+
+
+def _dataset(revisions: list[Revision]) -> Dataset:
+    return Dataset(
+        bucket="b",
+        dataset_id="d1",
+        name="d",
+        state=DatasetState.COMPLETE,
+        dataset_type="test",
+        provider="p",
+        identifier=Identifier(item_id=1),
+        metadata={},
+        created_at=FIXED_TIME,
+        updated_at=FIXED_TIME,
+        revisions=revisions,
+        last_modified_at=None,
+    )
+
+
+def test_squashed_state_failed_when_a_current_file_comes_from_a_failed_revision():
+    """Two files, both fail in rev0; rev1 fixes only f1 (f2 not re-fetched).
+
+    The squashed current view has f1 from rev1 (good) but f2 still from rev0
+    (VALIDATION_FAILED). The dataset still contains invalid data, so the
+    squashed state must be VALIDATION_FAILED — not the latest revision's state.
+    """
+    rev0 = _revision(0, [_file("f1"), _file("f2")], RevisionState.VALIDATION_FAILED)
+    rev1 = _revision(1, [_file("f1")], RevisionState.PENDING_VALIDATION)
+    dataset = _dataset([rev0, rev1])
+
+    current = dataset.current_revision
+
+    # precondition: f1 is now from rev1, f2 still from rev0
+    assert current.modified_files_map["f1"].revision_id == 1
+    assert current.modified_files_map["f2"].revision_id == 0
+
+    assert current.state == RevisionState.VALIDATION_FAILED, (
+        "f2 still comes from a failed revision, so the squashed state must be "
+        "VALIDATION_FAILED"
+    )
+
+
+def test_squashed_state_ok_when_all_current_files_come_from_good_revisions():
+    """Counter-case: if the failed files were all superseded by good revisions,
+    the squash must NOT be VALIDATION_FAILED (otherwise it would refetch forever)."""
+    rev0 = _revision(0, [_file("f1"), _file("f2")], RevisionState.VALIDATION_FAILED)
+    rev1 = _revision(1, [_file("f1"), _file("f2")], RevisionState.PENDING_VALIDATION)
+    dataset = _dataset([rev0, rev1])
+
+    current = dataset.current_revision
+
+    assert current.modified_files_map["f1"].revision_id == 1
+    assert current.modified_files_map["f2"].revision_id == 1
+    assert current.state != RevisionState.VALIDATION_FAILED