from __future__ import annotations import csv from pathlib import Path import pytest from scripts import run_small_classification from gcmd_classifier.articles import validate_article_records from gcmd_classifier.config import ModelSettings from gcmd_classifier.llm import FakeModelClient from gcmd_classifier.models import ( ArticleClassificationOutcome, ArticleProcessingStatus, ArticleRecord, ArticleResult, ReviewStatus, ) from gcmd_classifier.vocabulary import load_vocabulary FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json") def test_small_run_script_imports_without_live_api_calls() -> None: assert callable(run_small_classification.main) assert callable(run_small_classification.fake_scripted_responses) def test_limit_article_load_uses_first_valid_articles() -> None: load_result = validate_article_records( [ {"DOI": "10.example/1", "Title": "One", "Year": 2025, "Abstract": ""}, {"DOI": "10.example/2", "Title": "Two", "Year": 2025, "Abstract": ""}, {"DOI": "10.example/3", "Title": "Three", "Year": 2025, "Abstract": ""}, ] ) limited = run_small_classification.limit_article_load(load_result, 2) assert [article.DOI for article in limited.articles] == ["10.example/1", "10.example/2"] assert limited.source_count == 3 def test_fake_scripted_responses_support_requested_article_count() -> None: vocabulary = load_vocabulary(FIXTURE_PATH) actions = run_small_classification.fake_scripted_responses(vocabulary, article_count=3) assert actions[0]["selected"][0]["candidate_id"].startswith("topic_") assert actions[1]["selected"][0]["candidate_id"].startswith("term_") no_topic_count = sum(1 for action in actions if action.get("no_selection_reason")) assert no_topic_count == 2 def test_build_model_client_fake_returns_fake_model() -> None: client = run_small_classification.build_model_client( settings=ModelSettings(provider="fake"), fake=True, vocabulary=load_vocabulary(FIXTURE_PATH), article_count=1, ) assert isinstance(client, FakeModelClient) def test_live_mode_requires_openai_provider() -> None: with pytest.raises(SystemExit, match="requires --provider openai"): run_small_classification.build_model_client( settings=ModelSettings(provider="fake"), fake=False, vocabulary=load_vocabulary(FIXTURE_PATH), article_count=1, ) def test_live_mode_requires_credentials(monkeypatch) -> None: monkeypatch.delenv("OPENAI_API_KEY", raising=False) with pytest.raises(SystemExit, match="requires OPENAI_API_KEY"): run_small_classification.build_model_client( settings=ModelSettings(provider="openai", api_key_env_var="OPENAI_API_KEY"), fake=False, vocabulary=load_vocabulary(FIXTURE_PATH), article_count=1, ) def test_console_summary_includes_classification_and_no_classification(capsys) -> None: classified = ArticleResult( DOI="10.example/classified", Title="Classified title", Year=2025, Abstract="Text.", processing_status=ArticleProcessingStatus.COMPLETED, classification_outcome=ArticleClassificationOutcome.CLASSIFIED, classifications=( { "UUID": "topic-atmosphere", "name": "ATMOSPHERE", "level": "Topic", "canonical_path": "ATMOSPHERE", "path_components": ("ATMOSPHERE",), "topic": "ATMOSPHERE", "deterministic_validation": {"valid": True}, "final_status": "accepted", "review_required": True, "warnings": ( { "code": "REVIEW_RECOMMENDED_WEAK_SUPPORT", "message": "Manual scientific review is recommended.", }, ), }, ), review_status=ReviewStatus.NOT_REQUIRED, ) not_classified = ArticleResult( DOI="10.example/no", Title="No title", Year=2025, Abstract="", processing_status=ArticleProcessingStatus.COMPLETED, classification_outcome=ArticleClassificationOutcome.NOT_CLASSIFIED, classifications=(), no_classification_reason="No Topic selected.", review_status=ReviewStatus.NOT_REQUIRED, ) run_small_classification.print_console_summary((classified, not_classified)) output = capsys.readouterr().out assert "topic-atmosphere" in output assert "ATMOSPHERE" in output assert "No Topic selected." in output def test_review_csv_path_follows_output_dir(tmp_path: Path) -> None: output_dir = tmp_path / "small_run_gpt56" result = ArticleResult( DOI="10.example/no", Title="No title", Year=2025, Abstract="", processing_status=ArticleProcessingStatus.COMPLETED, classification_outcome=ArticleClassificationOutcome.NOT_CLASSIFIED, classifications=(), no_classification_reason="No Topic selected.", review_status=ReviewStatus.NOT_REQUIRED, ) csv_path = run_small_classification.write_review_csv((result,), output_dir) assert csv_path == output_dir / "review_table.csv" assert csv_path.exists() def test_review_csv_contains_classification_and_article_level_rows(tmp_path: Path) -> None: classified = ArticleResult( DOI="10.example/classified", Title="Classified title", Year=2025, Abstract="Text.", processing_status=ArticleProcessingStatus.COMPLETED, classification_outcome=ArticleClassificationOutcome.CLASSIFIED, classifications=( { "UUID": "topic-atmosphere", "name": "ATMOSPHERE", "level": "Topic", "canonical_path": "ATMOSPHERE", "path_components": ("ATMOSPHERE",), "topic": "ATMOSPHERE", "classifier_evidence": "Evidence text.", "support_type": "explicit", "reason_for_stopping": "Stopped here.", "confidence": {"final": 0.7}, "deterministic_validation": {"valid": True}, "final_status": "accepted", "review_required": True, "warnings": ( { "code": "REVIEW_RECOMMENDED_WEAK_SUPPORT", "message": "Manual scientific review is recommended.", }, ), }, ), review_status=ReviewStatus.NOT_REQUIRED, ) failed = ArticleResult( DOI="10.example/failed", Title="Failed title", Year=2025, Abstract="Text.", processing_status=ArticleProcessingStatus.FAILED, classification_outcome=None, classifications=(), errors=({"code": "MODEL_ERROR", "message": "Model failed."},), review_status=ReviewStatus.NOT_REQUIRED, ) csv_path = run_small_classification.write_review_csv((classified, failed), tmp_path) rows = list(csv.DictReader(csv_path.open())) assert len(rows) == 2 assert rows[0]["DOI"] == "10.example/classified" assert rows[0]["level"] == "Topic" assert rows[0]["UUID"] == "topic-atmosphere" assert rows[0]["canonical_path"] == "ATMOSPHERE" assert rows[0]["support_type"] == "explicit" assert rows[0]["confidence_final"] == "0.7" assert rows[0]["classifier_evidence"] == "Evidence text." assert rows[0]["reason_for_stopping"] == "Stopped here." assert rows[0]["deterministic_valid"] == "True" assert rows[0]["review_required"] == "True" assert "REVIEW_RECOMMENDED_WEAK_SUPPORT" in rows[0]["warnings"] assert rows[1]["DOI"] == "10.example/failed" assert rows[1]["UUID"] == "" assert rows[1]["errors"] == "MODEL_ERROR: Model failed." def test_progress_helpers_print_safe_truncated_output(capsys) -> None: article = ArticleRecord( DOI="10.example/progress", Title="A very long title " * 12, Year=2025, Abstract="Do not print this abstract.", ) result = 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, ) run_small_classification.print_article_start(0, 3, article) run_small_classification.print_article_finish(0, 3, result, 1.234) output = capsys.readouterr().out assert "Starting article 1/3" in output assert "DOI=10.example/progress" in output assert "..." in output assert "Finished article 1/3" in output assert "processing_status=completed" in output assert "classification_outcome=not_classified" in output assert "accepted_classifications=0" in output assert "elapsed_seconds=1.23" in output assert "Do not print this abstract" not in output def test_truncate_title_uses_one_line_and_limit() -> None: title = "Title with\nnewlines and " + "x" * 100 truncated = run_small_classification.truncate_title(title, max_length=30) assert "\n" not in truncated assert len(truncated) == 30 assert truncated.endswith("...") def test_one_line_compacts_tabs_and_newlines() -> None: assert run_small_classification.one_line("A\tlong\n title") == "A long title"