add: json receipt (closes #2)

examples/single-file/file-no-main.py

@@ -1,3 +1,4 @@
+#!/usr/bin/env python
 """Example: file without `__main__`"""
 
 # if __name__ == "__main__":

pyproject.toml

@@ -134,7 +134,7 @@ dev-all.shell = """
 """
 
 # Build
-cosmo = "cosmofy src/cosmofy --args '-m cosmofy --cosmo' --output dist/cosmofy"
+cosmo = "cosmofy src/cosmofy --args '-m cosmofy --cosmo' --output dist/cosmofy --receipt"
 build = ["pip install -e .", "python -m build", "cosmo"]
 clean = """
   rm -rf .mypy_cache;

src/cosmofy/__init__.py

@@ -61,8 +61,6 @@ def main(argv: Optional[List[str]] = None) -> int:
         )
         return 1
 
-    # output = Bundler(args).run()
-    # TODO: generate json receipt if requested
     Bundler(args).run()
     return 0
 

src/cosmofy/args.py

@@ -65,6 +65,9 @@
   -o PATH, --output PATH
     Path to output file (see below for default value).
 
+  --receipt
+    Whether to create a JSON file with the output date, version, and hash.
+
   --args STRING
     Cosmopolitan Python arguments (see below for default value).
 
@@ -118,8 +121,15 @@ class Args:
     dry_run: bool = False
     """Whether we should suppress any file-system operations."""
 
-    for_real: bool = True
-    """Internal value for the opposite of `dry_run`."""
+    @property
+    def for_real(self) -> bool:
+        """Internal value for the opposite of `dry_run`."""
+        return not self.dry_run
+
+    @for_real.setter
+    def for_real(self, value: bool) -> None:
+        """Set dry_run."""
+        self.dry_run = not value
 
     # cache
 
@@ -140,6 +150,9 @@ class Args:
     output: Optional[Path] = None
     """Path to the output file."""
 
+    receipt: bool = False
+    """Whether to create a JSON file with the output date, version, and hash."""
+
     paths: List[Path] = dataclasses.field(default_factory=list)
     """Paths to add."""
 
@@ -177,6 +190,7 @@ def parse(argv: List[str]) -> Args:
                 "--debug",
                 "--dry-run",
                 "--help",
+                "--receipt",
                 "--version",
             ]:
                 setattr(args, prop, True)

src/cosmofy/bundler.py

@@ -11,6 +11,7 @@
 from typing import Tuple
 from typing import Union
 import io
+import json
 import logging
 import marshal
 import os
@@ -24,6 +25,7 @@
 # pkg
 from .args import Args
 from .args import COSMOFY_PYTHON_URL
+from .updater import create_receipt
 from .updater import download
 from .updater import download_if_newer
 from .zipfile2 import ZipFile2
@@ -147,7 +149,7 @@ def fs_move_executable(self, src: Path, dest: Path) -> Path:
             dest.parent.mkdir(parents=True, exist_ok=True)
             # TODO 2024-10-31 @ py3.8 EOL: use `Path` instead of `str`
             shutil.move(str(src), str(dest))
-            mode = dest.stat().st_mode | stat.S_IEXEC
+            mode = dest.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
             dest.chmod(mode)
         return dest
 
@@ -273,6 +275,19 @@ def write_output(self, archive: ZipFile2, main: Pkg) -> Path:
             self.fs_move_executable(Path(archive.filename), output)
         return output
 
+    def write_receipt(self, path: Path) -> Bundler:
+        """Write a JSON receipt for the bundle."""
+        if not self.args.receipt:
+            return self
+
+        output = path.with_suffix(f"{path.suffix}.json")
+        receipt = json.dumps(create_receipt(path))
+        log.debug(f"{self.banner} receipt: {receipt}")
+        if self.args.for_real:
+            output.write_text(receipt)
+        log.debug(f"{self.banner}wrote JSON receipt {output}")
+        return self
+
     def run(self) -> Path:
         """Run the bundler."""
         archive = self.setup_archive()
@@ -281,7 +296,9 @@ def run(self) -> Path:
         main = self.zip_add(archive, include, exclude)
         self.zip_remove(archive, *self.args.remove)
         self.write_args(archive, main)
+        archive.close()
 
         output = self.write_output(archive, main)
         log.info(f"{self.banner}bundled: {output}")
+        self.write_receipt(output)
         return output

src/cosmofy/updater.py

@@ -5,15 +5,26 @@
 from datetime import timezone
 from email.utils import parsedate_to_datetime
 from pathlib import Path
+from typing import Dict
+from typing import Optional
 from urllib.request import Request
 from urllib.request import urlopen
+import hashlib
 import logging
+import re
+import subprocess
 
 log = logging.getLogger(__name__)
 
+DEFAULT_HASH = "sha256"
+"""Default hashing algorithm."""
+
 CHUNK_SIZE = 65536
 """Default chunk size."""
 
+RE_VERSION = re.compile(rb"\d+\.\d+\.\d+(-[\da-zA-Z-.]+)?(\+[\da-zA-Z-.]+)?")
+"""Regex for a semver version string."""
+
 
 def download(url: str, path: Path, chunk_size: int = CHUNK_SIZE) -> Path:
     """Download `url` to path."""
@@ -35,3 +46,31 @@ def download_if_newer(url: str, path: Path, chunk_size: int = CHUNK_SIZE) -> Pat
         remote = parsedate_to_datetime(response.headers.get("Last-Modified"))
         need_download = remote > local  # only download if newer
     return download(url, path, chunk_size) if need_download else path
+
+
+def create_receipt(
+    path: Path,
+    algo: str = DEFAULT_HASH,
+    date: Optional[datetime] = None,
+    version: str = "",
+) -> Dict[str, str]:
+    """Create JSON receipt data."""
+    digest = getattr(hashlib, algo)()
+    with path.open("rb") as f:
+        while chunk := f.read(CHUNK_SIZE):
+            digest.update(chunk)
+
+    date = date or datetime.now()
+    if not version:
+        out = subprocess.run(
+            f"{path} --version", shell=True, capture_output=True, check=True
+        ).stdout
+        if match := RE_VERSION.search(out):
+            version = match.group().decode("utf-8")
+
+    return {
+        "algo": algo,
+        "hash": digest.hexdigest(),
+        "date": date.isoformat()[:19] + "Z",
+        "version": version,
+    }

test/test_bundler.py

@@ -6,6 +6,7 @@
 from typing import Set
 from typing import Tuple
 from unittest.mock import patch
+from unittest.mock import MagicMock
 import io
 import os
 import tempfile
@@ -83,7 +84,7 @@ def test_copy() -> None:
         f.flush()
 
         src, dest = Path(f.name), Path(g.name)
-        Bundler(Args(dry_run=True, for_real=False)).fs_copy(src, dest)
+        Bundler(Args(dry_run=True)).fs_copy(src, dest)
         Bundler(Args()).fs_copy(src, dest)
         assert dest.read_bytes() == content
 
@@ -98,15 +99,15 @@ def test_move() -> None:
         f.flush()
 
         src, dest = Path(f.name), Path(g.name)
-        Bundler(Args(dry_run=True, for_real=False)).fs_move_executable(src, dest)
+        Bundler(Args(dry_run=True)).fs_move_executable(src, dest)
         Bundler(Args()).fs_move_executable(src, dest)
         assert os.access(dest, os.X_OK)
         assert dest.read_bytes() == content
 
 
 def test_from_download() -> None:
     """Archive from cache/download."""
-    test = Bundler(Args(dry_run=True, for_real=False))
+    test = Bundler(Args(dry_run=True))
     real = Bundler(Args())
     with tempfile.NamedTemporaryFile() as f, tempfile.NamedTemporaryFile() as g:
         src, dest = Path(f.name), Path(g.name)
@@ -125,7 +126,7 @@ def test_from_download() -> None:
 
 def test_setup_temp() -> None:
     """Setup tempfile."""
-    test = Bundler(Args(dry_run=True, for_real=False)).setup_temp()
+    test = Bundler(Args(dry_run=True)).setup_temp()
     assert isinstance(test[0], Path)
     assert isinstance(test[1], ZipFile2)
 
@@ -138,20 +139,18 @@ def test_setup_temp() -> None:
 def test_setup_archive() -> None:
     """Setup archive."""
     # clone (dry run)
-    assert Bundler(
-        Args(dry_run=True, for_real=False, cosmo=True, clone=True)
-    ).setup_archive()
+    assert Bundler(Args(dry_run=True, cosmo=True, clone=True)).setup_archive()
 
     # cache (dry run)
-    assert Bundler(Args(dry_run=True, for_real=False)).setup_archive()
+    assert Bundler(Args(dry_run=True)).setup_archive()
 
     # fresh (dry run)
-    assert Bundler(Args(dry_run=True, for_real=False, cache=None)).setup_archive()
+    assert Bundler(Args(dry_run=True, cache=None)).setup_archive()
 
 
 def test_process() -> None:
     """Process a file."""
-    test = Bundler(Args(dry_run=True, for_real=False))
+    test = Bundler(Args(dry_run=True))
 
     # find __main__ in file
     path = EXAMPLES / "pkg-with-init" / "__init__.py"
@@ -177,7 +176,7 @@ def test_process() -> None:
 
 def test_add() -> None:
     """Add files."""
-    test = Bundler(Args(dry_run=True, for_real=False))
+    test = Bundler(Args(dry_run=True))
     real = Bundler(Args())
     archive = _archive(io.BytesIO())
 
@@ -208,7 +207,7 @@ def test_add() -> None:
 
 def test_remove() -> None:
     """Remove files."""
-    test = Bundler(Args(dry_run=True, for_real=False))
+    test = Bundler(Args(dry_run=True))
     real = Bundler(Args())
     with tempfile.NamedTemporaryFile() as f:
         archive = _archive(f.name)
@@ -219,7 +218,7 @@ def test_remove() -> None:
 
 def test_write_args() -> None:
     """Write .args."""
-    test = Bundler(Args(dry_run=True, for_real=False))
+    test = Bundler(Args(dry_run=True))
     archive = _archive(io.BytesIO())
     assert test.write_args(archive, tuple())  # skip writing
     assert test.write_args(archive, ("foo", "__init__"))
@@ -230,7 +229,7 @@ def test_write_args() -> None:
 
 def test_write_output() -> None:
     """Write output zip."""
-    test = Bundler(Args(dry_run=True, for_real=False))
+    test = Bundler(Args(dry_run=True))
     with tempfile.NamedTemporaryFile() as f:
         assert test.write_output(_archive(f.name), tuple()) == Path("out.com")
 
@@ -239,7 +238,23 @@ def test_write_output() -> None:
     assert test.write_output(archive, ("foo", "bar")) == Path("bar.com")
 
 
+@patch("cosmofy.bundler.create_receipt")
+def test_write_receipt(_create_receipt: MagicMock) -> None:
+    """Write receipt."""
+    _create_receipt.return_value = {}
+
+    test = Bundler(Args(receipt=True, dry_run=True))
+    real = Bundler(Args(receipt=True))
+    with tempfile.NamedTemporaryFile() as f:
+        path = Path(f.name)
+        test.write_receipt(path)
+        _create_receipt.assert_called()
+
+        real.write_receipt(path)
+        _create_receipt.assert_called()
+
+
 def test_run() -> None:
     """Run bundler."""
-    test = Bundler(Args(dry_run=True, for_real=False))
+    test = Bundler(Args(dry_run=True))
     assert test.run() == Path("out.com")

test/test_updater.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 from unittest.mock import MagicMock
 from unittest.mock import patch
+import sys
 
 # pkg
 from cosmofy import updater
@@ -116,3 +117,17 @@ def test_download_if_not_newer(
     _urlopen.assert_called_once()
     _download.assert_not_called()
     assert result == path
+
+
+def test_receipt() -> None:
+    """Generate a receipt."""
+    path = Path(sys.executable)
+    data = updater.create_receipt(path, version="1.0.0")
+    assert isinstance(data, dict)
+
+    data = updater.create_receipt(path)
+    assert isinstance(data, dict)
+
+    path = Path(__file__).parent.parent / "examples" / "single-file" / "file-no-main.py"
+    data = updater.create_receipt(path)
+    assert isinstance(data, dict)