File size: 3,937 Bytes
e726170 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | from __future__ import annotations
import json
from pathlib import Path
import pytest
from gcmd_classifier.models import (
ArticleClassificationOutcome,
ArticleProcessingStatus,
ArticleResult,
OutputError,
ProcessingMetadata,
ReviewStatus,
RunSummary,
)
from gcmd_classifier.persistence import JsonResultStore
def _result(doi: str, *, failed: bool = False) -> ArticleResult:
if failed:
return ArticleResult(
DOI=doi,
Title="Failed article",
Year=2025,
Abstract="Text.",
processing_status=ArticleProcessingStatus.FAILED,
classification_outcome=None,
classifications=(),
review_status=ReviewStatus.NOT_REQUIRED,
errors=(OutputError(code="MODEL_FAILED", message="Model failed."),),
processing_metadata=ProcessingMetadata(cache_used=False),
)
return ArticleResult(
DOI=doi,
Title="Completed article",
Year=2025,
Abstract="Text.",
processing_status=ArticleProcessingStatus.COMPLETED,
classification_outcome=ArticleClassificationOutcome.NOT_CLASSIFIED,
classifications=(),
no_classification_reason="No Topic selected.",
review_status=ReviewStatus.NOT_REQUIRED,
processing_metadata=ProcessingMetadata(cache_used=False),
)
def test_article_result_is_saved_after_completion(tmp_path: Path) -> None:
store = JsonResultStore(tmp_path)
result = _result("10.example/done")
store.save_article_result(result)
loaded = store.load_article_result("10.example/done")
assert loaded == result
def test_failed_article_result_is_saved(tmp_path: Path) -> None:
store = JsonResultStore(tmp_path)
result = _result("10.example/failed", failed=True)
store.save_article_result(result)
assert store.load_article_result("10.example/failed") == result
def test_previous_results_survive_simulated_later_failure(tmp_path: Path) -> None:
store = JsonResultStore(tmp_path)
first = _result("10.example/first")
store.save_article_result(first)
with pytest.raises(RuntimeError):
raise RuntimeError("later article failed before save")
assert store.load_article_result("10.example/first") == first
def test_loading_existing_results_works(tmp_path: Path) -> None:
store = JsonResultStore(tmp_path)
first = _result("10.example/first")
second = _result("10.example/second", failed=True)
store.save_article_result(first)
store.save_article_result(second)
loaded = store.load_all_results()
assert {result.DOI for result in loaded} == {"10.example/first", "10.example/second"}
def test_duplicate_output_records_for_same_doi_are_avoided(tmp_path: Path) -> None:
store = JsonResultStore(tmp_path)
first = _result("10.example/same")
second = _result("10.example/same", failed=True)
store.save_article_result(first)
store.save_article_result(second)
loaded = store.load_all_results()
assert len(loaded) == 1
assert loaded[0].processing_status is ArticleProcessingStatus.FAILED
def test_consolidated_output_is_valid_json(tmp_path: Path) -> None:
store = JsonResultStore(tmp_path)
result = _result("10.example/done")
summary = RunSummary(run_id="run-1", articles_received=1, articles_completed=1)
output_path = store.write_consolidated(results=[result], summary=summary)
payload = json.loads(output_path.read_text())
assert payload["summary"]["run_id"] == "run-1"
assert payload["articles"][0]["DOI"] == "10.example/done"
def test_atomic_checkpoint_temp_file_is_removed_after_success(tmp_path: Path) -> None:
store = JsonResultStore(tmp_path)
result = _result("10.example/done")
store.save_article_result(result)
assert store.article_path(result.DOI).exists()
assert not list(store.article_path(result.DOI).parent.glob("*.tmp"))
|