Spaces:
Running on Zero
Running on Zero
File size: 4,225 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 | from __future__ import annotations
from pathlib import Path
from gcmd_classifier.config import ModelSettings
from gcmd_classifier.models import (
ArticleClassificationOutcome,
ArticleProcessingStatus,
ArticleRecord,
ArticleResult,
ProcessingMetadata,
ReviewStatus,
)
from gcmd_classifier.persistence import ArticleResultCache, build_cache_identity
from gcmd_classifier.vocabulary import load_vocabulary
FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json")
def _article(**overrides: object) -> ArticleRecord:
values = {
"DOI": "10.example/cache",
"Title": "Carbon dioxide observations",
"Year": 2025,
"Abstract": "Atmospheric carbon dioxide profiles are discussed.",
}
values.update(overrides)
return ArticleRecord.model_validate(values)
def _result(article: ArticleRecord) -> ArticleResult:
return ArticleResult(
DOI=article.DOI,
Title=article.Title,
Year=article.Year,
Abstract=article.Abstract,
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 _identity(article: ArticleRecord, settings: ModelSettings | None = None, **kwargs):
return build_cache_identity(
article=article,
vocabulary=load_vocabulary(FIXTURE_PATH),
settings=settings or ModelSettings(),
application_version=kwargs.pop("application_version", "0.1.0"),
relevant_config=kwargs.pop("relevant_config", None),
)
def test_cache_key_includes_article_fingerprint() -> None:
original = _identity(_article())
changed = _identity(_article(Title="Changed title"))
assert original.article_fingerprint != changed.article_fingerprint
assert original.cache_key != changed.cache_key
def test_cache_key_changes_when_article_content_changes() -> None:
assert (
_identity(_article(Abstract="A")).cache_key != _identity(_article(Abstract="B")).cache_key
)
def test_cache_key_changes_when_vocabulary_hash_changes() -> None:
identity = _identity(_article())
changed = identity.model_copy(update={"vocabulary_version": "different-vocab"})
assert identity.cache_key != changed.cache_key
def test_cache_key_changes_when_model_provider_name_or_parameters_change() -> None:
default = _identity(_article())
provider = _identity(_article(), ModelSettings(provider="other"))
name = _identity(_article(), ModelSettings(model_name="other-model"))
temperature = _identity(_article(), ModelSettings(temperature=0.2))
assert len({default.cache_key, provider.cache_key, name.cache_key, temperature.cache_key}) == 4
def test_cache_key_changes_when_prompt_versions_change() -> None:
default = _identity(_article())
changed = _identity(_article(), ModelSettings(prompt_version_topic="topic-v2"))
assert default.cache_key != changed.cache_key
def test_cache_key_changes_when_relevant_config_changes() -> None:
default = _identity(_article(), relevant_config={"candidate_limit": 100})
changed = _identity(_article(), relevant_config={"candidate_limit": 200})
assert default.cache_key != changed.cache_key
def test_cache_hit_returns_cache_used_true_and_completed_status(tmp_path: Path) -> None:
article = _article()
identity = _identity(article)
cache = ArticleResultCache(tmp_path)
cache.put(identity, _result(article))
cached = cache.get(identity)
assert cached is not None
assert cached.processing_metadata.cache_used is True
assert cached.processing_status is ArticleProcessingStatus.COMPLETED
assert cached.processing_status is not ArticleProcessingStatus.SKIPPED
def test_stale_cache_is_not_reused(tmp_path: Path) -> None:
article = _article()
identity = _identity(article)
stale_identity = _identity(_article(Title="Changed title"))
cache = ArticleResultCache(tmp_path)
cache.put(identity, _result(article))
assert cache.get(stale_identity) is None
|