Spaces:
Sleeping
Sleeping
| from app.schemas.commit_adaptation import CommitAdaptationReview | |
| from app.services import pending_review_store | |
| def _review() -> CommitAdaptationReview: | |
| return CommitAdaptationReview(memory_proposals=[], inferred_persona_text="Be concise.") | |
| def test_load_pending_review_returns_none_when_absent(tmp_path, monkeypatch): | |
| monkeypatch.setattr(pending_review_store, "_ROOT", tmp_path) | |
| assert pending_review_store.load_pending_review("project-a") is None | |
| def test_save_and_load_round_trips(tmp_path, monkeypatch): | |
| monkeypatch.setattr(pending_review_store, "_ROOT", tmp_path) | |
| pending_review_store.save_pending_review( | |
| "project-a", status="ready_to_flush", candidate_ids=["c1"], | |
| interaction_from_offset=0, interaction_through_offset=100, | |
| profile_staging_from_offset=0, profile_staging_through_offset=50, | |
| review=_review(), | |
| ) | |
| pending = pending_review_store.load_pending_review("project-a") | |
| assert pending.status == "ready_to_flush" | |
| assert pending.interaction_through_offset == 100 | |
| assert pending.review.inferred_persona_text == "Be concise." | |
| def test_update_pending_status_transitions_and_never_resets_to_ready(tmp_path, monkeypatch): | |
| monkeypatch.setattr(pending_review_store, "_ROOT", tmp_path) | |
| pending_review_store.save_pending_review( | |
| "project-a", status="ready_to_flush", candidate_ids=[], | |
| interaction_from_offset=0, interaction_through_offset=100, | |
| profile_staging_from_offset=0, profile_staging_through_offset=0, | |
| review=_review(), | |
| ) | |
| pending_review_store.update_pending_status("project-a", status="cognee_flush_started", flush_attempt_id="run-1") | |
| pending = pending_review_store.load_pending_review("project-a") | |
| assert pending.status == "cognee_flush_started" | |
| assert pending.flush_attempt_id == "run-1" | |
| pending_review_store.update_pending_status("project-a", status="cognee_flush_uncertain") | |
| assert pending_review_store.load_pending_review("project-a").status == "cognee_flush_uncertain" | |
| def test_delete_pending_review_removes_file(tmp_path, monkeypatch): | |
| monkeypatch.setattr(pending_review_store, "_ROOT", tmp_path) | |
| pending_review_store.save_pending_review( | |
| "project-a", status="ready_to_flush", candidate_ids=[], | |
| interaction_from_offset=0, interaction_through_offset=0, | |
| profile_staging_from_offset=0, profile_staging_through_offset=0, | |
| review=_review(), | |
| ) | |
| pending_review_store.delete_pending_review("project-a") | |
| assert pending_review_store.load_pending_review("project-a") is None | |
| def test_archive_as_uncertain_writes_audit_record(tmp_path, monkeypatch): | |
| monkeypatch.setattr(pending_review_store, "_ROOT", tmp_path) | |
| monkeypatch.setattr(pending_review_store, "_UNCERTAIN_ROOT", tmp_path / "uncertain_flushes") | |
| pending_review_store.save_pending_review( | |
| "project-a", status="cognee_flush_started", candidate_ids=["c1"], | |
| interaction_from_offset=0, interaction_through_offset=100, | |
| profile_staging_from_offset=0, profile_staging_through_offset=0, | |
| review=_review(), flush_attempt_id="run-1", | |
| ) | |
| pending = pending_review_store.load_pending_review("project-a") | |
| pending_review_store.archive_as_uncertain("project-a", pending) | |
| archived_files = list((tmp_path / "uncertain_flushes").glob("*.json")) | |
| assert len(archived_files) == 1 | |