diff --git a/src/deltatrack/diff_bill.py b/src/deltatrack/diff_bill.py index dac6b1a2..037a4887 100644 --- a/src/deltatrack/diff_bill.py +++ b/src/deltatrack/diff_bill.py @@ -2096,7 +2096,7 @@ def cmd_compare(args: argparse.Namespace) -> None: output = json.dumps(diff_dict, indent=2) if args.output: - with open(args.output, "w") as f: + with open(args.output, "w", encoding="utf-8") as f: f.write(output) else: print(output) diff --git a/src/deltatrack/diff_pdf.py b/src/deltatrack/diff_pdf.py index 3284f14b..2878afd9 100644 --- a/src/deltatrack/diff_pdf.py +++ b/src/deltatrack/diff_pdf.py @@ -1378,7 +1378,7 @@ def main(argv: list[str] | None = None) -> None: v2_label=args.v2_label, ) if args.output: - args.output.write_text(html) + args.output.write_text(html, encoding="utf-8") print(f"Wrote {args.output}", file=sys.stderr) else: print(html) diff --git a/tests/test_diff_bill.py b/tests/test_diff_bill.py index c517dd43..1120b08a 100644 --- a/tests/test_diff_bill.py +++ b/tests/test_diff_bill.py @@ -1,5 +1,6 @@ import argparse import json +import os import subprocess import sys from pathlib import Path @@ -641,6 +642,155 @@ def _run_compare(monkeypatch, *argv: str) -> None: main() +REPORT = "

old → new

⚠ unanchored

" + +_UTF8_ROUTE_CHILD = r""" +import locale +import os +import sys +from pathlib import Path + +REPORT = "

old \u2192 new

\u26a0 unanchored

" + + +def _ascii(value): + return str(value).encode("ascii", "backslashreplace").decode("ascii") + + +def _status(name): + print("DT627_STATUS=" + name) + return 0 + + +def _utf8_name(value): + return str(value).lower().replace("-", "").replace("_", "") in {"utf8", "utf"} + + +def main(): + route, output, old_input, new_input = sys.argv[1:] + try: + locale_encoding = locale.getencoding() + with open(os.devnull, "w") as probe: + file_encoding = probe.encoding + except Exception as exc: + print("DT627_STATUS=locale-unavailable") + print("DT627_DETAIL=" + type(exc).__name__) + return 11 + + print("DT627_DEFAULT_ENCODING=" + _ascii(locale_encoding)) + print("DT627_FILE_ENCODING=" + _ascii(file_encoding)) + if _utf8_name(locale_encoding) or _utf8_name(file_encoding): + return _status("locale-unavailable") + + try: + if route == "xml": + import deltatrack.diff_bill as diff_bill + from deltatrack.compare import xml as compare_xml + + diff_bill.normalize_bill = lambda _path: object() + compare_xml.compare_xml_trees_html = lambda *_args, **_kwargs: REPORT + sys.argv = [ + "diff_bill.py", + "compare", + old_input, + new_input, + "--format", + "html", + "-o", + output, + ] + route_main = diff_bill.main + elif route == "pdf": + import deltatrack.diff_pdf as diff_pdf + + diff_pdf.render_pdf_diff_html = lambda *_args, **_kwargs: REPORT + route_main = lambda: diff_pdf.main( + [old_input, new_input, "--output", output] + ) + else: + print("DT627_STATUS=setup-error") + print("DT627_DETAIL=unknown-route") + return 12 + except Exception as exc: + print("DT627_STATUS=setup-error") + print("DT627_DETAIL=" + type(exc).__name__) + return 12 + + try: + route_main() + except UnicodeEncodeError: + return _status("unicode-error") + except SystemExit as exc: + print("DT627_STATUS=route-error") + print("DT627_DETAIL=SystemExit:" + _ascii(exc.code)) + return 13 + except Exception as exc: + print("DT627_STATUS=route-error") + print("DT627_DETAIL=" + type(exc).__name__) + return 13 + + try: + report = Path(output).read_bytes() + except Exception as exc: + print("DT627_STATUS=byte-mismatch") + print("DT627_DETAIL=" + type(exc).__name__) + return 0 + if report != REPORT.encode("utf-8"): + return _status("byte-mismatch") + try: + decoded = report.decode("utf-8") + except UnicodeDecodeError: + return _status("decode-error") + if "\u2192" not in decoded or "\u26a0" not in decoded: + return _status("marker-mismatch") + return _status("ok") + + +raise SystemExit(main()) +""" + + +def _run_non_utf8_report_child(route: str, output: Path, old_input: Path, new_input: Path) -> None: + repo_root = Path(__file__).resolve().parents[1] + environment = os.environ.copy() + environment.update({"PYTHONUTF8": "0", "PYTHONCOERCECLOCALE": "0", "LC_ALL": "C", "LANG": "C"}) + pythonpath = [str(repo_root / "src"), str(repo_root)] + if environment.get("PYTHONPATH"): + pythonpath.append(environment["PYTHONPATH"]) + environment["PYTHONPATH"] = os.pathsep.join(pythonpath) + result = subprocess.run( + [ + sys.executable, + "-c", + _UTF8_ROUTE_CHILD, + route, + str(output), + str(old_input), + str(new_input), + ], + cwd=repo_root, + env=environment, + capture_output=True, + text=True, + ) + status = next( + (line for line in result.stdout.splitlines() if line.startswith("DT627_STATUS=")), + "", + ) + assert result.returncode == 0, ( + f"child setup/route failure ({status}); stdout={result.stdout!r}; stderr={result.stderr!r}" + ) + assert status == "DT627_STATUS=ok", ( + f"report route did not preserve UTF-8 ({status}); stdout={result.stdout!r}; stderr={result.stderr!r}" + ) + + report = output.read_bytes() + assert report == REPORT.encode("utf-8") + decoded = report.decode("utf-8") + assert "→" in decoded + assert "⚠" in decoded + + class TestIntermixedSubParserGuard: """The re-entrancy guard in _IntermixedSubParser, pinned on ANY interpreter (#426). @@ -870,6 +1020,15 @@ def test_output_flag_still_writes_the_file_and_nothing_to_stdout( assert data["old_version"] == "reported-in-house" assert data["new_version"] == "enrolled-bill" + def test_html_output_uses_utf8_when_host_default_is_cp1252(self, tmp_path): + """The real CLI writer is tested under a verified non-UTF-8 child locale.""" + old_xml = tmp_path / "old.xml" + new_xml = tmp_path / "new.xml" + output = tmp_path / "diff.html" + old_xml.write_bytes(b"") + new_xml.write_bytes(b"") + _run_non_utf8_report_child("xml", output, old_xml, new_xml) + class TestCompareVersionAddressableForm: """`compare ` resolves under --bills-dir and diffs (#152).""" diff --git a/tests/test_diff_pdf_cli.py b/tests/test_diff_pdf_cli.py index 49bb9804..2923b82e 100644 --- a/tests/test_diff_pdf_cli.py +++ b/tests/test_diff_pdf_cli.py @@ -2,6 +2,9 @@ from __future__ import annotations +import os +import subprocess +import sys from pathlib import Path from deltatrack.diff_pdf import build_parser, main @@ -9,6 +12,135 @@ V1 = fixture_path("118-hr-8752", "1_reported-in-house.pdf") V2 = fixture_path("118-hr-8752", "2_engrossed-in-house.pdf") +REPORT = "

old → new

⚠ unanchored

" + +_UTF8_ROUTE_CHILD = r""" +import locale +import os +import sys +from pathlib import Path + +REPORT = "

old \u2192 new

\u26a0 unanchored

" + + +def _ascii(value): + return str(value).encode("ascii", "backslashreplace").decode("ascii") + + +def _status(name): + print("DT627_STATUS=" + name) + return 0 + + +def _utf8_name(value): + return str(value).lower().replace("-", "").replace("_", "") in {"utf8", "utf"} + + +def main(): + route, output, old_input, new_input = sys.argv[1:] + try: + locale_encoding = locale.getencoding() + with open(os.devnull, "w") as probe: + file_encoding = probe.encoding + except Exception as exc: + print("DT627_STATUS=locale-unavailable") + print("DT627_DETAIL=" + type(exc).__name__) + return 11 + + print("DT627_DEFAULT_ENCODING=" + _ascii(locale_encoding)) + print("DT627_FILE_ENCODING=" + _ascii(file_encoding)) + if _utf8_name(locale_encoding) or _utf8_name(file_encoding): + return _status("locale-unavailable") + + try: + if route != "pdf": + print("DT627_STATUS=setup-error") + print("DT627_DETAIL=unknown-route") + return 12 + import deltatrack.diff_pdf as diff_pdf + + diff_pdf.render_pdf_diff_html = lambda *_args, **_kwargs: REPORT + route_main = lambda: diff_pdf.main( + [old_input, new_input, "--output", output] + ) + except Exception as exc: + print("DT627_STATUS=setup-error") + print("DT627_DETAIL=" + type(exc).__name__) + return 12 + + try: + route_main() + except UnicodeEncodeError: + return _status("unicode-error") + except SystemExit as exc: + print("DT627_STATUS=route-error") + print("DT627_DETAIL=SystemExit:" + _ascii(exc.code)) + return 13 + except Exception as exc: + print("DT627_STATUS=route-error") + print("DT627_DETAIL=" + type(exc).__name__) + return 13 + + try: + report = Path(output).read_bytes() + except Exception as exc: + print("DT627_STATUS=byte-mismatch") + print("DT627_DETAIL=" + type(exc).__name__) + return 0 + if report != REPORT.encode("utf-8"): + return _status("byte-mismatch") + try: + decoded = report.decode("utf-8") + except UnicodeDecodeError: + return _status("decode-error") + if "\u2192" not in decoded or "\u26a0" not in decoded: + return _status("marker-mismatch") + return _status("ok") + + +raise SystemExit(main()) +""" + + +def _run_non_utf8_report_child(route: str, output: Path, old_input: Path, new_input: Path) -> None: + repo_root = Path(__file__).resolve().parents[1] + environment = os.environ.copy() + environment.update({"PYTHONUTF8": "0", "PYTHONCOERCECLOCALE": "0", "LC_ALL": "C", "LANG": "C"}) + pythonpath = [str(repo_root / "src"), str(repo_root)] + if environment.get("PYTHONPATH"): + pythonpath.append(environment["PYTHONPATH"]) + environment["PYTHONPATH"] = os.pathsep.join(pythonpath) + result = subprocess.run( + [ + sys.executable, + "-c", + _UTF8_ROUTE_CHILD, + route, + str(output), + str(old_input), + str(new_input), + ], + cwd=repo_root, + env=environment, + capture_output=True, + text=True, + ) + status = next( + (line for line in result.stdout.splitlines() if line.startswith("DT627_STATUS=")), + "", + ) + assert result.returncode == 0, ( + f"child setup/route failure ({status}); stdout={result.stdout!r}; stderr={result.stderr!r}" + ) + assert status == "DT627_STATUS=ok", ( + f"report route did not preserve UTF-8 ({status}); stdout={result.stdout!r}; stderr={result.stderr!r}" + ) + + report = output.read_bytes() + assert report == REPORT.encode("utf-8") + decoded = report.decode("utf-8") + assert "→" in decoded + assert "⚠" in decoded def test_fixtures_committed(): @@ -47,6 +179,15 @@ def test_writes_html_file(self, tmp_path): assert "full-bill" in html assert "diff.json" in html + def test_html_output_uses_utf8_when_host_default_is_cp1252(self, tmp_path): + """The real CLI writer is tested under a verified non-UTF-8 child locale.""" + output = tmp_path / "diff.html" + v1_pdf = tmp_path / "old.pdf" + v2_pdf = tmp_path / "new.pdf" + v1_pdf.write_bytes(b"old") + v2_pdf.write_bytes(b"new") + _run_non_utf8_report_child("pdf", output, v1_pdf, v2_pdf) + def test_stdout_when_no_output(self, capsys): main([str(V1), str(V2)]) captured = capsys.readouterr()