octo: preserve UTF-8 text and LF bytes on Windows

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}")

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')

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()