172 lines
7.2 KiB
Python
172 lines
7.2 KiB
Python
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() -> Brief:
|
|
data = brief_dict()
|
|
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_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()
|