zemion
b73709c65a
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
392 lines
13 KiB
Python
392 lines
13 KiB
Python
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,
|
|
interpret_kosit_result,
|
|
verify_profile,
|
|
)
|
|
|
|
|
|
def _profile(tmp_path: Path, *, minimum_steps: int = 2) -> KoSITValidationProfile:
|
|
java = tmp_path / "java"
|
|
java.write_bytes(b"#!/bin/sh\n")
|
|
java.chmod(0o700)
|
|
jar = tmp_path / "validator.jar"
|
|
jar.write_bytes(b"pinned validator")
|
|
config = tmp_path / "config"
|
|
config.mkdir()
|
|
scenarios = config / "scenarios.xml"
|
|
scenarios.write_text("<scenarios/>", encoding="utf-8")
|
|
resources = config / "resources"
|
|
resources.mkdir()
|
|
(resources / "rules.xsl").write_text("<stylesheet/>", encoding="utf-8")
|
|
return KoSITValidationProfile(
|
|
profile_id="xrechnung-explicit-test",
|
|
xrechnung_version="explicit-test-only",
|
|
validator_version="validator-test",
|
|
configuration_version="configuration-test",
|
|
java_executable=java.resolve(),
|
|
validator_jar=jar.resolve(),
|
|
validator_jar_sha256=hashlib.sha256(jar.read_bytes()).hexdigest(),
|
|
configuration_root=config.resolve(),
|
|
configuration_tree_sha256=configuration_tree_sha256(config),
|
|
scenarios_file=scenarios.resolve(),
|
|
minimum_validation_steps=minimum_steps,
|
|
)
|
|
|
|
|
|
def _report(*, valid: bool, step_count: int = 2) -> bytes:
|
|
assessment = "accept" if valid else "reject"
|
|
flag = "true" if valid else "false"
|
|
steps = "".join(
|
|
f'<rep:validationStepResult id="step-{index}" valid="{flag}" />'
|
|
for index in range(step_count)
|
|
)
|
|
return (
|
|
f'<rep:report xmlns:rep="http://www.xoev.de/de/validator/varl/1" valid="{flag}" varlVersion="1.0.0">'
|
|
f"<rep:scenarioMatched>{steps}</rep:scenarioMatched>"
|
|
f"<rep:assessment><rep:{assessment}/></rep:assessment>"
|
|
"</rep:report>"
|
|
).encode()
|
|
|
|
|
|
def _invoice() -> InboundInvoice:
|
|
return InboundInvoice(
|
|
tenant_id="tenant-a",
|
|
source_reference="mail:message-1:attachment-1",
|
|
document=b"<Invoice/>",
|
|
received_at=datetime(2026, 8, 23, tzinfo=UTC),
|
|
)
|
|
|
|
|
|
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)
|
|
assert len(first) == 64
|
|
|
|
(profile.configuration_root / "resources" / "rules.xsl").write_text("changed", encoding="utf-8")
|
|
with pytest.raises(XRechnungValidationError, match="tree digest"):
|
|
verify_profile(profile)
|
|
|
|
|
|
def test_complete_valid_report_can_create_digest_bound_handoff(tmp_path: Path) -> None:
|
|
profile = _profile(tmp_path)
|
|
invoice = _invoice()
|
|
result = interpret_kosit_result(
|
|
profile=profile,
|
|
profile_sha256=verify_profile(profile),
|
|
document_sha256=invoice.document_sha256,
|
|
return_code=0,
|
|
runner_output=b"INFO validation completed",
|
|
report=_report(valid=True),
|
|
)
|
|
|
|
handoff = create_validated_handoff(invoice, result)
|
|
|
|
assert result.technical_outcome == "complete"
|
|
assert result.conformance == "valid"
|
|
assert result.assessment == "accept"
|
|
assert handoff.document_sha256 == invoice.document_sha256
|
|
assert len(handoff.handoff_sha256) == 64
|
|
|
|
|
|
def test_semantically_invalid_report_is_complete_but_cannot_handoff(tmp_path: Path) -> None:
|
|
profile = _profile(tmp_path)
|
|
invoice = _invoice()
|
|
result = interpret_kosit_result(
|
|
profile=profile,
|
|
profile_sha256=verify_profile(profile),
|
|
document_sha256=invoice.document_sha256,
|
|
return_code=0,
|
|
runner_output=b"INFO validation completed",
|
|
report=_report(valid=False),
|
|
)
|
|
|
|
assert result.technical_outcome == "complete"
|
|
assert result.conformance == "invalid"
|
|
assert result.assessment == "reject"
|
|
with pytest.raises(XRechnungValidationError, match="technically complete"):
|
|
create_validated_handoff(invoice, result)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("return_code", "output", "report", "expected"),
|
|
[
|
|
(1, b"", _report(valid=True), "failed"),
|
|
(0, b"ERROR Transformation failed", _report(valid=True), "incomplete"),
|
|
(0, b"", _report(valid=True, step_count=1), "incomplete"),
|
|
(0, b"", None, "incomplete"),
|
|
],
|
|
)
|
|
def test_technical_failures_never_trust_a_valid_looking_report(
|
|
tmp_path: Path,
|
|
return_code: int,
|
|
output: bytes,
|
|
report: bytes | None,
|
|
expected: str,
|
|
) -> None:
|
|
profile = _profile(tmp_path)
|
|
result = interpret_kosit_result(
|
|
profile=profile,
|
|
profile_sha256=verify_profile(profile),
|
|
document_sha256="a" * 64,
|
|
return_code=return_code,
|
|
runner_output=output,
|
|
report=report,
|
|
)
|
|
|
|
assert result.technical_outcome == expected
|
|
assert result.conformance == "unknown"
|
|
assert result.assessment == "unknown"
|
|
assert result.handoff_allowed is False
|
|
|
|
|
|
def test_manifest_does_not_select_an_active_standard_version() -> None:
|
|
manifest = get_manifest()
|
|
assert manifest.version == "0.1.21"
|
|
assert "none is activated by default" in manifest.architecture.known_limits[0].lower()
|