diff --git a/ingestify/__init__.py b/ingestify/__init__.py index 5c59e9e..8798985 100644 --- a/ingestify/__init__.py +++ b/ingestify/__init__.py @@ -11,4 +11,4 @@ from .exceptions import StopProcessing from .main import debug_source -__version__ = "0.17.0" +__version__ = "0.17.1" diff --git a/ingestify/domain/models/dataset/dataset.py b/ingestify/domain/models/dataset/dataset.py index bf30304..e832fe9 100644 --- a/ingestify/domain/models/dataset/dataset.py +++ b/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=""), ) diff --git a/ingestify/tests/test_refetch_validation_failed.py b/ingestify/tests/test_refetch_validation_failed.py index a962d01..16c940f 100644 --- a/ingestify/tests/test_refetch_validation_failed.py +++ b/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