from __future__ import annotations import hashlib import json import tempfile import unittest from pathlib import Path from unittest.mock import patch from claridoc.models import Brief, PipelineConfig from claridoc.pipeline import _mock_provider_warning, run_pipeline from claridoc.providers.base import ProviderRequest, ProviderResponse from claridoc.providers.mock import MockProvider from claridoc.templates import mock_pipeline_config from tests.helpers import brief_dict, make_brief, make_sources def _korean_experience_brief(document_type: str = "technical_blog") -> Brief: data = brief_dict(document_type) data.update( { "title": "기술적 선택을 경험과 근거로 설명하기", "language": "ko-KR", "reader_goal": "기술적 선택의 이유와 검증 방법을 이해한다", "core_message": "선택의 배경과 대안, 비용, 검증을 경험의 흐름으로 연결해야 합니다.", "scope": ["하나의 기술적 선택"], "non_scope": ["근거가 없는 일반화"], "required_topics": ["문제", "대안", "선택 이유", "검증"], } ) data["constraints"]["style_profile"] = "auto" return Brief.from_dict(data) class _PlainEndingWriter(MockProvider): def generate(self, request: ProviderRequest) -> ProviderResponse: response = super().generate(request) if request.stage == "draft": response.text = response.text.replace( "처음에는 작은 구현 선택 하나만 고치면 된다고 생각했습니다.", "처음에는 작은 구현 선택 하나만 고치면 된다고 생각했다.", 1, ) return response class PipelineTests(unittest.TestCase): def test_mixed_mock_pipeline_is_explicitly_flagged(self) -> None: config_data = mock_pipeline_config() config_data["writer"] = {"provider": "claude"} warning = _mock_provider_warning(PipelineConfig.from_dict(config_data)) self.assertIn("mixes external providers", warning) self.assertIn("synthetic", warning) def test_end_to_end_mock_run_creates_auditable_artifacts(self) -> None: with tempfile.TemporaryDirectory() as temp: output = Path(temp) / "run" result = run_pipeline( make_brief(), make_sources(), PipelineConfig.from_dict(mock_pipeline_config()), output, ) self.assertTrue(result.passed) self.assertTrue(result.final_path.is_file()) self.assertTrue(result.report_path.is_file()) self.assertTrue(result.manifest_path.is_file()) self.assertTrue((output / "final" / "provenance.md").is_file()) self.assertTrue((output / "final" / "evidence-map.json").is_file()) run_data = json.loads((output / "run.json").read_text(encoding="utf-8")) self.assertTrue(run_data["passed"]) self.assertTrue(any("deterministic mocks" in warning for warning in result.warnings)) report_text = result.report_path.read_text(encoding="utf-8") self.assertIn("Provider topology", report_text) self.assertIn("synthetic", report_text) manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) paths = {item["path"] for item in manifest["files"]} self.assertIn("final/document.md", paths) self.assertIn("provider-events.jsonl", paths) self.assertIn("final/provenance.md", paths) self.assertIn("final/evidence-map.json", paths) self.assertNotIn("manifest.json", paths) for item in manifest["files"]: artifact = output / item["path"] self.assertEqual(artifact.stat().st_size, item["bytes"]) self.assertEqual(hashlib.sha256(artifact.read_bytes()).hexdigest(), item["sha256"]) def test_revision_limit_is_enforced_when_gate_cannot_pass(self) -> None: with tempfile.TemporaryDirectory() as temp: output = Path(temp) / "run" config_data = mock_pipeline_config() config_data["quality_gate"]["minimum_score"] = 99 config_data["quality_gate"]["max_revisions"] = 1 result = run_pipeline( make_brief(), make_sources(), PipelineConfig.from_dict(config_data), output, ) self.assertFalse(result.passed) self.assertEqual(len(result.rounds), 2) self.assertTrue((output / "rounds" / "round-01" / "revision.raw.txt").is_file()) def test_korean_mock_run_records_reader_prose_contract(self) -> None: with tempfile.TemporaryDirectory() as temp: output = Path(temp) / "run" result = run_pipeline( _korean_experience_brief(), make_sources(), PipelineConfig.from_dict(mock_pipeline_config()), output, ) self.assertTrue(result.passed) self.assertEqual( result.rounds[-1].lint_report.metrics["style_contract"], "korean_first_person_experience_v1", ) report_text = result.report_path.read_text(encoding="utf-8") self.assertIn("Reader-prose contract", report_text) self.assertIn("korean_first_person_experience_v1", report_text) def test_korean_readme_mock_run_satisfies_reader_prose_contract(self) -> None: with tempfile.TemporaryDirectory() as temp: result = run_pipeline( _korean_experience_brief("readme"), make_sources(), PipelineConfig.from_dict(mock_pipeline_config()), Path(temp) / "run", ) self.assertTrue(result.passed) metrics = result.rounds[-1].lint_report.metrics self.assertEqual(metrics["plain_form_ending_count"], 0) self.assertTrue(metrics["opening_has_first_person"]) self.assertGreaterEqual(metrics["experience_section_coverage"], 0.5) def test_style_blocker_cannot_be_hidden_by_permissive_error_limit(self) -> None: with tempfile.TemporaryDirectory() as temp: output = Path(temp) / "run" config_data = mock_pipeline_config() config_data["quality_gate"].update( { "minimum_score": 0, "max_errors": 99, "max_revisions": 0, } ) with patch( "claridoc.pipeline.create_provider", side_effect=lambda spec: _PlainEndingWriter(spec), ): result = run_pipeline( _korean_experience_brief(), make_sources(), PipelineConfig.from_dict(config_data), output, ) self.assertFalse(result.passed) self.assertGreater(result.rounds[-1].blocker_count, 0) self.assertIn( "STYLE002", {issue.code for issue in result.rounds[-1].lint_report.issues}, ) def test_reviewer_role_cannot_escape_artifact_directory(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) output = root / "run" config_data = mock_pipeline_config() config_data["reviewers"] = [{"role": "../../logic reviewer", "provider": "mock"}] run_pipeline( make_brief(), make_sources(), PipelineConfig.from_dict(config_data), output, ) review_files = list((output / "rounds" / "round-01").glob("review-*")) self.assertTrue(any(path.name.endswith("logic-reviewer.json") for path in review_files)) self.assertEqual(list(root.glob("logic*")), []) if __name__ == "__main__": unittest.main()