fix(xrechnung): bound validator output and report reads
Enforce capture limits while draining both subprocess pipes, kill and reap interrupted validators, and bound report allocation before parsing. Preserve result precedence and add synthetic-process regressions with EN/DE documentation. Refs #2
This commit is contained in:
@@ -1,15 +1,23 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterator
|
||||
from datetime import UTC, datetime
|
||||
import hashlib
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from govoplan_xrechnung.backend import validation
|
||||
from govoplan_xrechnung.backend.manifest import get_manifest
|
||||
from govoplan_xrechnung.backend.validation import (
|
||||
InboundInvoice,
|
||||
KoSITValidationProfile,
|
||||
KoSITValidator,
|
||||
MAX_RUNNER_OUTPUT_BYTES,
|
||||
XRechnungValidationError,
|
||||
configuration_tree_sha256,
|
||||
create_validated_handoff,
|
||||
@@ -70,6 +78,231 @@ def _invoice() -> InboundInvoice:
|
||||
)
|
||||
|
||||
|
||||
def _runner_profile(tmp_path: Path, script: str) -> KoSITValidationProfile:
|
||||
"""Install a synthetic executable; never invoke Java or a real invoice validator."""
|
||||
profile = _profile(tmp_path)
|
||||
profile.java_executable.write_text(
|
||||
f"#!{sys.executable}\n"
|
||||
"import os, sys, time\n"
|
||||
"from pathlib import Path\n"
|
||||
f"Path(sys.argv[-1]).with_name('invoice-report.xml').write_bytes({_report(valid=True)!r})\n"
|
||||
+ script,
|
||||
encoding="utf-8",
|
||||
)
|
||||
return profile
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def runner_processes(monkeypatch: pytest.MonkeyPatch) -> Iterator[list[subprocess.Popen[bytes]]]:
|
||||
processes: list[subprocess.Popen[bytes]] = []
|
||||
real_popen = subprocess.Popen
|
||||
|
||||
def start(*args, **kwargs):
|
||||
process = real_popen(*args, **kwargs)
|
||||
processes.append(process)
|
||||
return process
|
||||
|
||||
monkeypatch.setattr(validation.subprocess, "Popen", start)
|
||||
try:
|
||||
yield processes
|
||||
finally:
|
||||
for process in processes:
|
||||
if process.poll() is None:
|
||||
process.kill()
|
||||
process.wait()
|
||||
|
||||
|
||||
def _assert_runner_reaped(processes: list[subprocess.Popen[bytes]]) -> None:
|
||||
assert len(processes) == 1
|
||||
process = processes[0]
|
||||
assert process.returncode is not None
|
||||
assert process.stdout is not None and process.stdout.closed
|
||||
assert process.stderr is not None and process.stderr.closed
|
||||
with pytest.raises(ChildProcessError):
|
||||
os.waitpid(process.pid, os.WNOHANG)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def report_read_sizes(monkeypatch: pytest.MonkeyPatch) -> list[int]:
|
||||
sizes: list[int] = []
|
||||
real_open = Path.open
|
||||
|
||||
class ReportReader:
|
||||
def __init__(self, stream):
|
||||
self.stream = stream
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *_args):
|
||||
self.stream.close()
|
||||
|
||||
def read(self, size=-1):
|
||||
sizes.append(size)
|
||||
assert size == validation.MAX_REPORT_BYTES + 1
|
||||
return self.stream.read(size)
|
||||
|
||||
def open_file(path, mode="r", *args, **kwargs):
|
||||
stream = real_open(path, mode, *args, **kwargs)
|
||||
if path.name == "invoice-report.xml" and mode == "rb":
|
||||
return ReportReader(stream)
|
||||
return stream
|
||||
|
||||
monkeypatch.setattr(Path, "open", open_file)
|
||||
return sizes
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stream", ["stdout", "stderr", "both"])
|
||||
def test_noisy_runner_is_stopped_at_shared_output_limit(
|
||||
tmp_path: Path,
|
||||
runner_processes: list[subprocess.Popen[bytes]],
|
||||
report_read_sizes: list[int],
|
||||
stream: str,
|
||||
) -> None:
|
||||
count = MAX_RUNNER_OUTPUT_BYTES
|
||||
writes = {
|
||||
"stdout": f"sys.stdout.buffer.write(b'x' * {count})\nsys.stdout.flush()\n",
|
||||
"stderr": f"sys.stderr.buffer.write(b'x' * {count})\nsys.stderr.flush()\n",
|
||||
"both": (
|
||||
f"sys.stdout.buffer.write(b'x' * {count // 2})\nsys.stdout.flush()\n"
|
||||
f"sys.stderr.buffer.write(b'x' * {count // 2})\nsys.stderr.flush()\n"
|
||||
),
|
||||
}
|
||||
profile = _runner_profile(tmp_path, writes[stream] + "time.sleep(30)\n")
|
||||
|
||||
result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
|
||||
|
||||
assert result.technical_outcome == "failed"
|
||||
assert result.technical_reason == "KoSIT runner output exceeded the safety limit."
|
||||
assert result.conformance == result.assessment == "unknown"
|
||||
assert not result.handoff_allowed
|
||||
assert result.report_sha256 is None
|
||||
assert result.diagnostics == ()
|
||||
assert report_read_sizes == []
|
||||
_assert_runner_reaped(runner_processes)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("extra_bytes", "exit_code"), [(0, 0), (1, 0), (100_000, 0), (100_000, 7)])
|
||||
def test_report_reads_are_bounded_and_preserve_exact_limit(
|
||||
tmp_path: Path,
|
||||
runner_processes: list[subprocess.Popen[bytes]],
|
||||
report_read_sizes: list[int],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
extra_bytes: int,
|
||||
exit_code: int,
|
||||
) -> None:
|
||||
report_limit = len(_report(valid=True))
|
||||
monkeypatch.setattr(validation, "MAX_REPORT_BYTES", report_limit)
|
||||
profile = _runner_profile(
|
||||
tmp_path,
|
||||
"with Path(sys.argv[-1]).with_name('invoice-report.xml').open('ab') as report:\n"
|
||||
f" report.write(b' ' * {extra_bytes})\n"
|
||||
f"sys.exit({exit_code})\n",
|
||||
)
|
||||
|
||||
result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
|
||||
|
||||
assert report_read_sizes == [report_limit + 1]
|
||||
if extra_bytes:
|
||||
assert result.technical_outcome == ("failed" if exit_code else "incomplete")
|
||||
assert result.technical_reason == (
|
||||
f"KoSIT runner exited with status {exit_code}; report semantics are not trusted."
|
||||
if exit_code else "KoSIT did not produce a bounded XML report."
|
||||
)
|
||||
assert result.conformance == result.assessment == "unknown"
|
||||
assert result.report_sha256 is None
|
||||
assert not result.handoff_allowed
|
||||
else:
|
||||
assert result.technical_outcome == "complete"
|
||||
assert result.report_sha256 == hashlib.sha256(_report(valid=True)).hexdigest()
|
||||
assert result.handoff_allowed
|
||||
_assert_runner_reaped(runner_processes)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("excess_bytes", [0, 1])
|
||||
def test_completed_runner_output_preserves_exact_combined_limit(
|
||||
tmp_path: Path, runner_processes: list[subprocess.Popen[bytes]], excess_bytes: int
|
||||
) -> None:
|
||||
stdout_size = MAX_RUNNER_OUTPUT_BYTES // 2
|
||||
stderr_size = MAX_RUNNER_OUTPUT_BYTES - stdout_size - 1 + excess_bytes
|
||||
profile = _runner_profile(
|
||||
tmp_path,
|
||||
f"sys.stdout.buffer.write(b'x' * {stdout_size})\n"
|
||||
f"sys.stderr.buffer.write(b'x' * {stderr_size})\n",
|
||||
)
|
||||
|
||||
result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
|
||||
|
||||
assert result.technical_outcome == ("failed" if excess_bytes else "complete")
|
||||
assert result.handoff_allowed is (excess_bytes == 0)
|
||||
_assert_runner_reaped(runner_processes)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("close_pipes", [False, True])
|
||||
def test_hanging_runner_times_out_and_is_reaped(
|
||||
tmp_path: Path, runner_processes: list[subprocess.Popen[bytes]], close_pipes: bool
|
||||
) -> None:
|
||||
script = "os.close(1)\nos.close(2)\n" if close_pipes else ""
|
||||
profile = _runner_profile(tmp_path, script + "time.sleep(30)\n")
|
||||
started = time.monotonic()
|
||||
|
||||
result = KoSITValidator(profile, timeout_seconds=1).validate(_invoice())
|
||||
|
||||
assert time.monotonic() - started < 5
|
||||
assert result.technical_outcome == "failed"
|
||||
assert result.technical_reason == "KoSIT validation timed out; no handoff is allowed."
|
||||
assert result.conformance == result.assessment == "unknown"
|
||||
assert not result.handoff_allowed
|
||||
_assert_runner_reaped(runner_processes)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("script", "outcome", "reason"),
|
||||
[
|
||||
("sys.stderr.write('synthetic failure')\nsys.exit(7)\n", "failed", "status 7"),
|
||||
(
|
||||
"sys.stdout.buffer.write(b'x' * 65532 + b'\\nERROR synthetic failure')\n",
|
||||
"incomplete",
|
||||
"technical error",
|
||||
),
|
||||
("sys.stderr.write('ERROR synthetic failure')\n", "incomplete", "technical error"),
|
||||
],
|
||||
)
|
||||
def test_runner_failure_and_technical_output_never_trust_valid_report(
|
||||
tmp_path: Path,
|
||||
runner_processes: list[subprocess.Popen[bytes]],
|
||||
script: str,
|
||||
outcome: str,
|
||||
reason: str,
|
||||
) -> None:
|
||||
profile = _runner_profile(tmp_path, script)
|
||||
|
||||
result = KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
|
||||
|
||||
assert result.technical_outcome == outcome
|
||||
assert reason in result.technical_reason
|
||||
assert result.conformance == result.assessment == "unknown"
|
||||
assert not result.handoff_allowed
|
||||
_assert_runner_reaped(runner_processes)
|
||||
|
||||
|
||||
def test_runner_cancellation_kills_and_reaps_before_propagating(
|
||||
tmp_path: Path,
|
||||
runner_processes: list[subprocess.Popen[bytes]],
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
profile = _runner_profile(tmp_path, "time.sleep(30)\n")
|
||||
|
||||
def cancel(_selector, _timeout):
|
||||
raise KeyboardInterrupt
|
||||
|
||||
monkeypatch.setattr(validation.selectors.DefaultSelector, "select", cancel)
|
||||
with pytest.raises(KeyboardInterrupt):
|
||||
KoSITValidator(profile, timeout_seconds=2).validate(_invoice())
|
||||
|
||||
_assert_runner_reaped(runner_processes)
|
||||
|
||||
|
||||
def test_profile_verifies_exact_engine_and_complete_configuration_tree(tmp_path: Path) -> None:
|
||||
profile = _profile(tmp_path)
|
||||
first = verify_profile(profile)
|
||||
|
||||
Reference in New Issue
Block a user