81 lines
3.8 KiB
Python
81 lines
3.8 KiB
Python
"""Remote-only regression checks for example output capture and failure reports."""
|
|
|
|
import contextlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import endor_examples as examples
|
|
|
|
|
|
def result(stdout="", code=0):
|
|
return {"command": ["mock"], "exit_code": code, "stdout": stdout, "stderr": "", "error": None}
|
|
|
|
|
|
def report():
|
|
return {"format_version": 1, "commit": "test", "generated_at": "test", "status": "NOT_RUN",
|
|
"preparation": [], "cases": [], "error": None}
|
|
|
|
|
|
class ExampleReportTests(unittest.TestCase):
|
|
def test_manifest_runs_every_source_and_captures_exact_output(self):
|
|
cases = examples.load_cases()
|
|
responses = [result(), result(json.dumps({"target_directory": "/tmp/ci-target"}))]
|
|
responses += [result(case["expected_stdout"]) for case in cases]
|
|
collected = report()
|
|
with patch.object(examples, "invoke", side_effect=responses) as invoke:
|
|
examples.collect(collected)
|
|
self.assertEqual(collected["status"], "PASS")
|
|
self.assertEqual(invoke.call_count, len(cases) + 2)
|
|
for actual, expected in zip(collected["cases"], cases):
|
|
self.assertEqual(actual["stdout"], expected["expected_stdout"])
|
|
self.assertEqual(actual["status"], "PASS")
|
|
self.assertEqual(actual["binary"], "/tmp/ci-target/debug/lyra")
|
|
|
|
def test_output_mismatch_fails_and_build_failure_never_claims_examples_ran(self):
|
|
cases = examples.load_cases()
|
|
responses = [result(), result(json.dumps({"target_directory": "/tmp/target"}))]
|
|
responses += [result("wrong\n")] + [result(case["expected_stdout"]) for case in cases[1:]]
|
|
collected = report()
|
|
with patch.object(examples, "invoke", side_effect=responses):
|
|
examples.collect(collected)
|
|
self.assertEqual(collected["status"], "FAIL")
|
|
self.assertEqual(collected["cases"][0]["stdout"], "wrong\n")
|
|
self.assertEqual(collected["cases"][0]["status"], "FAIL")
|
|
self.assertIn("wrong", examples.markdown(collected))
|
|
collected = report()
|
|
with patch.object(examples, "invoke", return_value=result(code=1)) as invoke:
|
|
examples.collect(collected)
|
|
self.assertEqual(invoke.call_count, 1)
|
|
self.assertEqual(collected["status"], "NOT_RUN")
|
|
self.assertTrue(all(case["status"] == "NOT_RUN" for case in collected["cases"]))
|
|
|
|
def test_timeout_keeps_partial_utf8_stdout_and_stderr(self):
|
|
failure = subprocess.TimeoutExpired(["cli"], 1, output="日".encode(), stderr=b"partial error")
|
|
with patch.object(examples.subprocess, "run", side_effect=failure):
|
|
captured = examples.invoke(["cli"], 1)
|
|
self.assertIsNone(captured["exit_code"])
|
|
self.assertEqual(captured["stdout"], "日")
|
|
self.assertEqual(captured["stderr"], "partial error")
|
|
self.assertIn("timed out", captured["error"])
|
|
|
|
def test_unexpected_failure_still_writes_both_artifacts(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
with patch("sys.argv", ["endor_examples.py", "--output-dir", directory]), \
|
|
patch.object(examples, "collect", side_effect=ValueError("broken manifest")), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(examples.main(), 1)
|
|
payload = json.loads(Path(directory, "endor-examples.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(payload["status"], "FAIL")
|
|
self.assertEqual(payload["error"], "broken manifest")
|
|
self.assertIn("broken manifest", Path(directory, "endor-examples.md").read_text(encoding="utf-8"))
|
|
self.assertTrue(examples.fence("```\n日").startswith("````text\n"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|