From cf48cff380a412f7f99a9730344127bf4beeafa2 Mon Sep 17 00:00:00 2001 From: jscjscjscjscjsc <290224331+jscjscjscjscjsc@users.noreply.github.com> Date: Wed, 14 Oct 2026 13:52:05 +0800 Subject: [PATCH] octo: preserve UTF-8 text and LF bytes across Windows authoring --- tools/octo | 42 +++++++------ tools/test_octo.py | 3 +- tools/test_octo_encoding.py | 119 ++++++++++++++++++++++++++++++++++++ 3 files changed, 145 insertions(+), 19 deletions(-) create mode 100644 tools/test_octo_encoding.py diff --git a/tools/octo b/tools/octo index b4330484..11caeb87 100755 --- a/tools/octo +++ b/tools/octo @@ -59,6 +59,12 @@ PLATFORMS = ("android", "ios", "macos", "windows", "linux", "openharmony", "web" PNG_MAGIC = b"\x89PNG\r\n\x1a\n" +def write_utf8(path, content): + """Bundle digests require UTF-8 and LF on every authoring platform.""" + with path.open("w", encoding="utf-8", newline="\n") as stream: + stream.write(content) + + def sealed_release(manifest): integrity = manifest.get("integrity") or {} # Typed unsigned manifests may contain signature:null. Any non-null @@ -69,14 +75,14 @@ def sealed_release(manifest): def publisher_workflow(): """Use the reviewed immutable Hub toolchain, never a moving branch.""" - config = json.loads((REPO / "tools/publisher-toolchain.json").read_text()) + config = json.loads((REPO / "tools/publisher-toolchain.json").read_text(encoding="utf-8")) revision = config.get("revision") if (config.get("schema") != 1 or config.get("repository") != "OctoSense-org/OctoSense-App-Hub" or not isinstance(revision, str) or re.fullmatch(r"[0-9a-f]{40}", revision) is None): raise ValueError("GitHub publishing needs a reviewed immutable App Hub toolchain revision") - template = (REPO / "tools/publish-app.template.yml").read_text() + template = (REPO / "tools/publish-app.template.yml").read_text(encoding="utf-8") if template.count("__HUB_REVISION__") != 1: raise ValueError("GitHub publishing workflow template is invalid") return template.replace("__HUB_REVISION__", revision) @@ -90,7 +96,7 @@ def install_publisher_workflow(app, content, replace=False): if path.is_symlink(): raise ValueError("Refusing a symlink in the publishing workflow path") if target.exists(): - if target.is_file() and target.read_text() == content: + if target.is_file() and target.read_text(encoding="utf-8") == content: return False if not replace: raise ValueError(".github/workflows/publish-app.yml already exists; review it before using --replace") @@ -272,21 +278,21 @@ def cmd_new(args): shutil.copy2(src, dest / extra) manifest_path = dest / "bundle" / "manifest.json" - manifest = json.loads(manifest_path.read_text()) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["id"] = app_id manifest["name"] = name manifest["version"] = "0.1.0" manifest["integrity"] = {"bundle_blake3": ""} - manifest_path.write_text(json.dumps(manifest, indent=2) + "\n") + write_utf8(manifest_path, json.dumps(manifest, indent=2) + "\n") listing_path = dest / "bundle" / "listing.json" - listing = json.loads(listing_path.read_text()) + listing = json.loads(listing_path.read_text(encoding="utf-8")) listing["platforms"] = list(dict.fromkeys(args.platform)) - listing_path.write_text(json.dumps(listing, indent=2) + "\n") + write_utf8(listing_path, json.dumps(listing, indent=2) + "\n") main = dest / "bundle" / "main.splash" if args.name: - main.write_text(main.read_text().replace('text: "My Notes"', "text: " + json.dumps(name), 1)) + write_utf8(main, main.read_text(encoding="utf-8").replace('text: "My Notes"', "text: " + json.dumps(name), 1)) install_publisher_workflow(dest, workflow) hub, _, _ = find_binary("hub", "OCTO_HUB") @@ -382,7 +388,7 @@ def wait_for(log_path, proc, port, timeout): listening = admitted = error = None want = re.compile(r"\[makepad-remote\] listening on 127\.0\.0\.1:%d pid=%d\b" % (port, proc.pid)) while time.time() < deadline: - text = log_path.read_text(errors="replace") if log_path.exists() else "" + text = log_path.read_text(encoding="utf-8", errors="replace") if log_path.exists() else "" for line in text.splitlines(): if want.search(line) and not listening: listening = line.strip() @@ -416,7 +422,7 @@ def cmd_run(args): if not (bundle / "manifest.json").is_file(): die(f"{bundle} has no manifest.json; pass the bundle directory (…/bundle)") try: - manifest = json.loads((bundle / "manifest.json").read_text()) + manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8")) if sealed_release(manifest): die("card-host cannot run a sealed release. Use the editable source bundle or install the release in a compatible OctoSense host.") except (OSError, ValueError) as error: @@ -469,7 +475,7 @@ def cmd_run(args): proc.wait(timeout=5) except subprocess.TimeoutExpired: proc.kill() - print(log_path.read_text(errors="replace")[-4000:]) + print(log_path.read_text(encoding="utf-8", errors="replace")[-4000:]) die(msg + f"\n (stopped the card-host it started, pid {proc.pid}; log {log_path})") listening, admitted, error = wait_for(log_path, proc, args.port, args.timeout) @@ -543,7 +549,7 @@ def cmd_check(args, extra): if not (bundle / "manifest.json").is_file(): die(f"{bundle} has no manifest.json") hub = need("hub", "OCTO_HUB") - manifest = json.loads((bundle / "manifest.json").read_text()) + manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8")) signed = sealed_release(manifest) if signed: print("octo: the manifest carries signing metadata; checking the exact bytes without restamping.") @@ -575,7 +581,7 @@ def cmd_check(args, extra): # Things the gate does not look at, but a reviewer will. listing_path = bundle / "listing.json" if listing_path.is_file(): - text = listing_path.read_text() + text = listing_path.read_text(encoding="utf-8") hits = [p for p in PLACEHOLDERS if p in text] if hits: print(f"octo: note: listing.json still holds template placeholders ({', '.join(hits)}); " @@ -590,7 +596,7 @@ def cmd_publish_github(args): if not manifest_path.is_file(): die("Expected an app directory containing bundle/manifest.json") try: - manifest = json.loads(manifest_path.read_text()) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if sealed_release(manifest): die("Use the editable source repository, not a sealed release bundle") workflow = publisher_workflow() @@ -865,8 +871,8 @@ def cmd_wasm_new(args): die(f"bundle/fns/{name}.wasm already exists, and `wasm build` would replace it; choose another name") sdk, summary = sdk_dependency(crate, args.sdk) (crate / "src").mkdir(parents=True) - template = (COMPONENT_TEMPLATE / "Cargo.toml.template").read_text() - (crate / "Cargo.toml").write_text(template.replace("{{name}}", name).replace("{{sdk}}", sdk)) + template = (COMPONENT_TEMPLATE / "Cargo.toml.template").read_text(encoding="utf-8") + write_utf8(crate / "Cargo.toml", template.replace("{{name}}", name).replace("{{sdk}}", sdk)) shutil.copy2(COMPONENT_TEMPLATE / "src" / "lib.rs", crate / "src" / "lib.rs") shutil.copy2(COMPONENT_TEMPLATE / ".gitignore", crate / ".gitignore") print(f"created {crate}") @@ -1161,7 +1167,7 @@ def cmd_wasm_build(args): bundle = app / "bundle" manifest_path = bundle / "manifest.json" try: - manifest = json.loads(manifest_path.read_text()) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) except (OSError, ValueError) as error: die(f"{manifest_path}: {error}") if sealed_release(manifest): @@ -1198,7 +1204,7 @@ def cmd_wasm_build(args): print(f" {script_call(export)}") print(f" {crates_line(b['crates'])} (tools/octo wasm info lists them)") if notes: - manifest_path.write_text(json.dumps(manifest, indent=2) + "\n") + write_utf8(manifest_path, json.dumps(manifest, indent=2) + "\n") print("bundle/manifest.json:") for note in notes: print(f" {note}") diff --git a/tools/test_octo.py b/tools/test_octo.py index 4e3b6d58..34ee4ccf 100644 --- a/tools/test_octo.py +++ b/tools/test_octo.py @@ -431,7 +431,8 @@ def test_new_writes_the_template_with_the_sdk_by_path_or_commit(self): self.assertIn('name = "text-tools"', cargo) self.assertIn('crate-type = ["cdylib", "rlib"]', cargo) self.assertIn('because you chose --sdk path', cargo) - path = re.search(r'^octosense-component = \{ path = "(.+)" \}$', cargo, re.M).group(1) + encoded_path = re.search(r'^octosense-component = \{ path = (".+") \}$', cargo, re.M).group(1) + path = json.loads(encoded_path) self.assertEqual((crate / path).resolve(), (HERE.parent / 'sdk/rust/octosense-component').resolve()) self.assertEqual((crate / 'src/lib.rs').read_text(), (HERE.parent / 'templates/rust-component/src/lib.rs').read_text()) self.assertEqual((crate / '.gitignore').read_text(), '/target/\n') diff --git a/tools/test_octo_encoding.py b/tools/test_octo_encoding.py new file mode 100644 index 00000000..043d933b --- /dev/null +++ b/tools/test_octo_encoding.py @@ -0,0 +1,119 @@ +"""Bundle text keeps its UTF-8 bytes on non-UTF-8 Windows installations.""" +import argparse +import contextlib +import importlib.machinery +import importlib.util +import io +import json +from pathlib import Path +import shutil +import tempfile +import unittest +from unittest.mock import patch + +HERE = Path(__file__).resolve().parent +loader = importlib.machinery.SourceFileLoader('encoding_octo', str(HERE / 'octo')) +spec = importlib.util.spec_from_loader(loader.name, loader) +octo = importlib.util.module_from_spec(spec) +loader.exec_module(octo) + + +@contextlib.contextmanager +def legacy_windows_text(): + """Use real file I/O with CP936 and CRLF only when no format is specified.""" + original_open = io.open + + def open_file(file, mode='r', buffering=-1, encoding=None, errors=None, + newline=None, closefd=True, opener=None): + if 'b' not in mode: + encoding = encoding or 'cp936' + if newline is None and any(flag in mode for flag in ('w', 'a', '+')): + newline = '\r\n' + return original_open(file, mode, buffering, encoding, errors, newline, + closefd, opener) + + with patch.object(io, 'text_encoding', create=True, side_effect=lambda encoding, stacklevel=2: encoding or 'cp936'), \ + patch.object(io, 'open', side_effect=open_file): + yield + + +class BundleEncoding(unittest.TestCase): + def app(self, root): + app = Path(root) / 'app' + (app / 'bundle').mkdir(parents=True) + (app / 'bundle/manifest.json').write_bytes(json.dumps({ + 'id': 'dev.example.brief', 'name': '晨报📰…', 'version': '0.1.0', + 'integrity': {'signature': 'signed-fixture'}, + }, ensure_ascii=False).encode('utf-8')) + return app + + def test_new_preserves_unicode_source_and_lf_bytes(self): + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + template = root / 'template' + shutil.copytree(octo.TEMPLATE, template) + source = 'Label{text: "晨报📰…"}\n' + (template / 'bundle/main.splash').write_bytes(source.encode('utf-8')) + dest = root / 'brief' + args = argparse.Namespace(dir=str(dest), id='dev.example.brief', + name='Brief', platform=['windows'], system=False) + with patch.object(octo, 'TEMPLATE', template), \ + patch.object(octo, 'find_binary', return_value=(None, None, [])), \ + legacy_windows_text(), contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(octo.cmd_new(args), 0) + self.assertEqual((dest / 'bundle/main.splash').read_bytes(), source.encode('utf-8')) + for name in ('manifest.json', 'listing.json'): + raw = (dest / 'bundle' / name).read_bytes() + json.loads(raw.decode('utf-8')) + self.assertNotIn(b'\r\n', raw) + + def test_publish_reads_unicode_manifest_and_installs_only_once(self): + with tempfile.TemporaryDirectory() as temp: + app = self.app(temp) + manifest = app / 'bundle/manifest.json' + value = json.loads(manifest.read_bytes()) + value['integrity'] = {'bundle_blake3': 'development'} + manifest.write_bytes(json.dumps(value, ensure_ascii=False).encode('utf-8')) + original = manifest.read_bytes() + workflow_text = '# 晨报📰…\nname: fixture\n' + args = argparse.Namespace(dir=str(app), replace=False) + with patch.object(octo, 'publisher_workflow', return_value=workflow_text), \ + legacy_windows_text(), contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(octo.cmd_publish_github(args), 0) + workflow = app / '.github/workflows/publish-app.yml' + before = workflow.stat().st_mtime_ns + self.assertEqual(octo.cmd_publish_github(args), 0) + self.assertEqual(workflow.stat().st_mtime_ns, before) + self.assertEqual(workflow.read_bytes(), workflow_text.encode('utf-8')) + self.assertEqual(manifest.read_bytes(), original) + + def test_check_reads_unicode_listing_without_restamping_signed_bytes(self): + with tempfile.TemporaryDirectory() as temp: + app = self.app(temp) + bundle = app / 'bundle' + (bundle / 'listing.json').write_bytes('{"description":"晨报📰"}'.encode('utf-8')) + original = (bundle / 'manifest.json').read_bytes() + args = argparse.Namespace(bundle=str(bundle), catalog=None, publisher_key=[]) + with patch.object(octo, 'need', return_value=Path('/fixture/hub')), \ + patch.object(octo.subprocess, 'run') as run, \ + legacy_windows_text(), contextlib.redirect_stdout(io.StringIO()): + run.return_value.returncode = 0 + run.return_value.stdout = run.return_value.stderr = '' + self.assertEqual(octo.cmd_check(args, []), 0) + self.assertEqual(run.call_count, 1) + self.assertEqual((bundle / 'manifest.json').read_bytes(), original) + + def test_run_still_refuses_unicode_signed_manifest_before_startup(self): + with tempfile.TemporaryDirectory() as temp: + app = self.app(temp) + args = argparse.Namespace(bundle=str(app / 'bundle')) + error = io.StringIO() + with patch.object(octo, 'need') as binary, legacy_windows_text(), \ + contextlib.redirect_stderr(error), self.assertRaises(SystemExit): + octo.cmd_run(args) + self.assertIn('cannot run a sealed release', error.getvalue()) + binary.assert_not_called() + + +if __name__ == '__main__': + unittest.main()