S19: consume versioned engineering resources and move landscape documentation

scripts/engineering.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""Materialize a pinned engineering release without relying on generated files."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import tempfile
+
+
+def git(directory, *arguments):
+    result = subprocess.run(['git', '-C', str(directory), *arguments],
+                            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+                            env={**os.environ, 'GIT_TERMINAL_PROMPT': '0'}, timeout=180)
+    if result.returncode:
+        raise ValueError('Git operation failed; check repository access and credentials')
+    return result.stdout.strip()
+
+
+def files(directory):
+    result = {}
+    if not directory.exists():
+        return result
+    for path in directory.rglob('*'):
+        relative = path.relative_to(directory)
+        if any(part in {'target', '__pycache__'} for part in relative.parts):
+            continue
+        if path.is_symlink():
+            raise ValueError('Symlinks are not allowed in generated resources: ' + str(relative))
+        if path.is_file():
+            result[str(relative)] = (hashlib.sha256(path.read_bytes()).hexdigest(),
+                                    bool(path.stat().st_mode & 0o111))
+    return result
+
+
+def materialize(root, check=False):
+    root = Path(root).resolve()
+    lock = json.loads((root / 'engineering.lock.json').read_text())
+    if not re.fullmatch(r'v\d+\.\d+\.\d+', lock['tag']):
+        raise ValueError('Engineering tag must be vMAJOR.MINOR.PATCH')
+    if not re.fullmatch(r'[0-9a-f]{40}', lock['commit']):
+        raise ValueError('Engineering commit must be a full SHA')
+    with tempfile.TemporaryDirectory(prefix='engineering-fetch-') as temporary:
+        checkout = Path(temporary)
+        git(checkout, 'init', '-q')
+        git(checkout, 'fetch', '--quiet', '--depth=1', '--no-tags', '--',
+            lock['repository'], 'refs/tags/' + lock['tag'])
+        actual = git(checkout, 'rev-parse', 'FETCH_HEAD^{commit}')
+        if actual != lock['commit']:
+            raise ValueError('Engineering tag moved: resolved commit differs from lock')
+        git(checkout, 'checkout', '--quiet', '--detach', actual)
+        source = checkout / 'resources'
+        if not source.is_dir():
+            raise ValueError('Engineering release has no resources directory')
+        files(source)  # Validate before copying, including symlinks.
+        (source / 'SOURCE.json').write_text(json.dumps(lock, indent=2, sort_keys=True) + '\n')
+        destination = root / 'generated'
+        if destination.is_symlink():
+            raise ValueError('generated/ must not be a symlink')
+        expected, existing = files(source), files(destination)
+        if check:
+            changed = sorted(k for k in expected.keys() | existing.keys()
+                             if expected.get(k) != existing.get(k))
+            if changed:
+                raise ValueError('Generated resources differ; regenerate and commit:\n' + '\n'.join(changed))
+            print('Engineering resources match ' + lock['tag'] + ' at ' + actual)
+            return
+        # Stage on the same filesystem. Preserve the previous tree until replacement succeeds.
+        with tempfile.TemporaryDirectory(prefix='.engineering-', dir=root) as staging:
+            staged = Path(staging) / 'next'
+            previous = Path(staging) / 'previous'
+            shutil.copytree(source, staged)
+            if destination.exists():
+                destination.rename(previous)
+            try:
+                staged.rename(destination)
+            except BaseException:
+                if previous.exists():
+                    previous.rename(destination)
+                raise
+        print('Generated engineering resources from ' + lock['tag'] + ' at ' + actual)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--check', action='store_true')
+    parser.add_argument('--root', type=Path, default=Path(__file__).resolve().parents[1])
+    args = parser.parse_args()
+    try:
+        materialize(args.root, args.check)
+    except (ValueError, KeyError, OSError, subprocess.TimeoutExpired) as error:
+        parser.exit(1, str(error) + '\n')
+
+
+if __name__ == '__main__':
+    main()