From b0505369914fef630fb96cf9db929beb7ff2ecb5 Mon Sep 17 00:00:00 2001 From: Simon Sawert Date: Fri, 23 Oct 2026 21:54:46 +0200 Subject: [PATCH] feat: serve the Apps page's GitHub requests from local checkouts Patch `adafruit_requests.Session.request` so raw file and tree requests for the main repo are answered from the checkout being run, including uncommitted changes. `--repo` picks another checkout or `github` for real GitHub, and `--repo-map` maps other repos such as third-party apps. Co-Authored-By: Claude Opus 5.5 --- README.md | 27 +++ matrixbox_simulator/device/local_repo.py | 280 +++++++++++++++++++++++ matrixbox_simulator/device/run_app.py | 90 +++++++- tests/test_local_repo.py | 43 ++++ 4 files changed, 439 insertions(+), 1 deletion(-) create mode 100644 matrixbox_simulator/device/local_repo.py create mode 100644 tests/test_local_repo.py diff --git a/README.md b/README.md index 55b0e6d..a37046d 100644 --- a/README.md +++ b/README.md @@ -196,6 +196,33 @@ page's own Connect button fails outright rather than faking success, same as there being no real network for it to join), or toggle it live with `n`. +## Testing the Apps page without pushing + +The Apps page (store, System update, third-party apps) reads its file +list and files from GitHub. By default `matrixbox app` answers those +requests from the checkout being run instead, so the store shows that +checkout as if it were pushed: uncommitted edits and new untracked files +are served, `.gitignore`d and deleted files are not. Which repo counts +as "the checkout" follows `repository_url` in the device's settings. +Every other request still goes to the real network, and all of them fail +while wifi is off. + +```sh +matrixbox app /path/to/matrixbox # store served from this checkout +matrixbox app /path/to/matrixbox --repo github # real GitHub +matrixbox app /path/to/matrixbox --repo /path/to/matrixbox-feature +``` + +`--repo-map OWNER/REPO=PATH` does the same for any other GitHub repo, such +as a third-party app you're working on. It can be given more than once: + +```sh +matrixbox app /path/to/matrixbox --repo-map user/app=/path/to/app +``` + +The ref in a request is ignored: whatever branch the device asks for, it +gets the working tree. + ## Screenshots For CI, PR previews, or anywhere else a terminal isn't available: diff --git a/matrixbox_simulator/device/local_repo.py b/matrixbox_simulator/device/local_repo.py new file mode 100644 index 0000000..7e6fd42 --- /dev/null +++ b/matrixbox_simulator/device/local_repo.py @@ -0,0 +1,280 @@ +"""Answers the device's GitHub requests (raw files and tree listings) from +local checkouts instead, so the Apps page shows unpushed changes exactly as +it would after a push. Everything else goes out to the real network. +""" + +import errno +import io +import json +import os +import stat +import subprocess +from collections.abc import Callable +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit + +RAW_HOST = "raw.githubusercontent.com" +API_HOST = "api.github.com" + +# Matches the device's own default when settings.txt has no repository_url. +DEFAULT_REPOSITORY_URL = ( + "https://raw.githubusercontent.com/MatrixBOX-dev/matrixbox/refs/heads/main/" +) + +# Captured before the path sandbox replaces them, which would otherwise +# redirect reads of the real checkout into the staged copy. +_REAL_OPEN = io.open +_REAL_STAT = os.stat + +_SKIPPED_DIRECTORY_NAMES = {"__pycache__"} + + +def repository_identity(repository_url: str) -> tuple[str, str] | None: + """The (owner, repo) in a raw.githubusercontent.com URL, taken the same + way the device does: its 2nd and 3rd path segments.""" + parts = repository_url.replace("https://", "").split("/") + if len(parts) < 3 or not parts[1] or not parts[2]: + return None + + return parts[1], parts[2] + + +def repository_key(owner: str, repo: str) -> str: + # GitHub is case-insensitive here, and the device rewrites the owner's + # case on load, so all comparisons go through this. + return f"{owner}/{repo}".lower() + + +class LocalResponse: + """The subset of an `adafruit_requests.Response` the device uses.""" + + def __init__(self, status_code: int, content: bytes) -> None: + self.status_code = status_code + self.content = content + + @property + def text(self) -> str: + return self.content.decode("utf-8", errors="replace") + + def json(self) -> Any: + return json.loads(self.content) + + def close(self) -> None: + pass + + +class LocalRepository: + """One local directory, served the way GitHub would serve it if the + working tree were committed and pushed.""" + + def __init__(self, root: Path) -> None: + self.root = root.resolve() + + def file_paths(self) -> list[str]: + paths = self._git_file_paths() + if paths is None: + paths = self._walked_file_paths() + + # Deleted but not yet committed files still show up in git's index. + return sorted(path for path in set(paths) if self._is_file(path)) + + def read(self, path: str) -> bytes | None: + # Only what a push would publish: no ignored or stray files. + if path not in self.file_paths(): + return None + + with _REAL_OPEN(self.root / path, "rb") as file: + return file.read() + + def tree(self) -> dict[str, Any]: + blobs = self.file_paths() + directories = { + "/".join(path.split("/")[:depth]) + for path in blobs + for depth in range(1, path.count("/") + 1) + } + entries = [ + { + "path": path, + "type": "blob", + "mode": "100644", + } + for path in blobs + ] + entries += [ + { + "path": path, + "type": "tree", + "mode": "040000", + } + for path in directories + ] + entries.sort(key=lambda entry: entry["path"]) + + return { + "sha": "local", + "truncated": False, + "tree": entries, + } + + def _git_file_paths(self) -> list[str] | None: + try: + result = subprocess.run( + ["git", "ls-files", "-z", "--cached", "--others", "--exclude-standard"], + cwd=self.root, + capture_output=True, + check=True, + ) + except (OSError, subprocess.CalledProcessError): + return None + + return [path for path in result.stdout.decode("utf-8").split("\0") if path] + + def _walked_file_paths(self) -> list[str]: + paths = [] + for directory, directory_names, file_names in os.walk(self.root): + directory_names[:] = [ + name + for name in directory_names + if not name.startswith(".") and name not in _SKIPPED_DIRECTORY_NAMES + ] + relative_directory = Path(directory).relative_to(self.root) + for name in file_names: + if not name.startswith("."): + paths.append((relative_directory / name).as_posix()) + + return paths + + def _is_file(self, path: str) -> bool: + try: + return stat.S_ISREG(_REAL_STAT(str(self.root / path)).st_mode) + except OSError: + return False + + +class GitHubInterceptor: + """Routes GitHub raw and tree requests to local repositories. + + The main repo is whatever the staged settings.txt's `repository_url` + names, re-read on every request since the web UI can change it.""" + + def __init__( + self, + settings_path: Path, + main_repository: LocalRepository | None, + mapped_repositories: dict[str, LocalRepository], + is_online: Callable[[], bool], + ) -> None: + self.settings_path = settings_path + self.main_repository = main_repository + self.mapped_repositories = mapped_repositories + self.is_online = is_online + + def repository_url(self) -> str: + try: + with _REAL_OPEN(self.settings_path) as file: + url = json.loads(file.read()).get("repository_url") + except (OSError, ValueError, AttributeError): + url = None + + return url if isinstance(url, str) and url else DEFAULT_REPOSITORY_URL + + def respond(self, method: str, url: str) -> LocalResponse | None: + """A local answer for this request, or None to let it go out to the + real network.""" + if method.upper() != "GET": + return None + + parts = urlsplit(url) + segments = parts.path.lstrip("/").split("/") + if parts.hostname == RAW_HOST and len(segments) >= 3: + return self._respond_raw(url, segments) + + if ( + parts.hostname == API_HOST + and len(segments) >= 6 + and segments[0] == "repos" + and segments[3:5] == ["git", "trees"] + ): + return self._respond_tree(segments[1], segments[2]) + + return None + + def _repository_for(self, owner: str, repo: str) -> LocalRepository | None: + key = repository_key(owner, repo) + if key in self.mapped_repositories: + return self.mapped_repositories[key] + + main_identity = repository_identity(self.repository_url()) + if main_identity is not None and key == repository_key(*main_identity): + return self.main_repository + + return None + + def _respond_raw(self, url: str, segments: list[str]) -> LocalResponse | None: + repository = self._repository_for(segments[0], segments[1]) + if repository is None: + return None + + self._require_online() + path = self._strip_main_prefix(url) or _strip_ref(segments[2:]) + content = repository.read(path) if path else None + if content is None: + # GitHub's raw host answers a missing file with exactly this body. + return LocalResponse(404, b"404: Not Found") + + return LocalResponse(200, content) + + def _respond_tree(self, owner: str, repo: str) -> LocalResponse | None: + repository = self._repository_for(owner, repo) + if repository is None: + return None + + self._require_online() + + return LocalResponse(200, json.dumps(repository.tree()).encode("utf-8")) + + def _strip_main_prefix(self, url: str) -> str | None: + # Matching repository_url literally is the only way to know where a + # branch name containing "/" ends. + prefix = self.repository_url().rstrip("/") + "/" + + request_path = url.split("?")[0] + if request_path.lower().startswith(prefix.lower()): + return request_path[len(prefix) :] + + return None + + def _require_online(self) -> None: + # Same error the socket stub raises, so offline handling is identical. + if not self.is_online(): + raise OSError(errno.ENETUNREACH, "Network is unreachable") + + def install(self, session_class: type) -> None: + """Patches `request`, which every get/post/... goes through.""" + real_request = getattr(session_class, "request") + + def request( + session: object, + method: str, + url: str, + *args: Any, + **kwargs: Any, + ) -> object: + response = self.respond(method, url) + if response is None: + return real_request(session, method, url, *args, **kwargs) + + return response + + setattr(session_class, "request", request) + + +def _strip_ref(segments: list[str]) -> str: + # refs/heads//..., /... or HEAD/... ; a branch with "/" + # in it can't be told apart from a directory here. + if segments[:2] == ["refs", "heads"]: + segments = segments[2:] + + return "/".join(segments[1:]) diff --git a/matrixbox_simulator/device/run_app.py b/matrixbox_simulator/device/run_app.py index ffe008b..942cb84 100644 --- a/matrixbox_simulator/device/run_app.py +++ b/matrixbox_simulator/device/run_app.py @@ -33,7 +33,7 @@ from types import FrameType from typing import Any, NoReturn, TextIO -from matrixbox_simulator.device import button_input, frame_bridge +from matrixbox_simulator.device import button_input, frame_bridge, local_repo from matrixbox_simulator.sizes import ( ROTATION_OVERRIDES, SIZE_CYCLE, @@ -722,6 +722,7 @@ def _run_kernel( sys.path.insert(0, str(staged_root)) sys.path.insert(0, str(staged_root / "lib")) sys.path.insert(0, str(STUB_DIR)) + _install_local_repositories(framework_root, staged_root, args) def cycle_size() -> NoReturn: _cycle_size(staged_root / "settings.txt") @@ -779,6 +780,72 @@ def cycle_size() -> NoReturn: print("\nmatrixbox-simulator: stopped") +def _wifi_connected() -> bool: + module = sys.modules.get("wifi") + + return bool(getattr(getattr(module, "radio", None), "connected", True)) + + +def _install_local_repositories( + framework_root: Path, staged_root: Path, args: argparse.Namespace +) -> None: + # Answers the Apps page's GitHub requests from local checkouts, so a + # change shows up in the store without pushing it first. + main_repository = None + if args.repo != "github": + main_root = ( + framework_root if args.repo is None else Path(args.repo).expanduser() + ) + if not main_root.is_dir(): + raise SystemExit(f"--repo: no such directory: {main_root}") + + main_repository = local_repo.LocalRepository(main_root) + + mapped_repositories = {} + for key, path in args.repo_map: + if not path.is_dir(): + raise SystemExit(f"--repo-map: no such directory: {path}") + + mapped_repositories[key] = local_repo.LocalRepository(path) + + if main_repository is None and not mapped_repositories: + return + + settings_path = staged_root / "settings.txt" + interceptor = local_repo.GitHubInterceptor( + settings_path, + main_repository, + mapped_repositories, + _wifi_connected, + ) + + # Patching the class, before main.py creates its session, covers every + # session the device makes. It lives in the checkout, so only now importable. + import adafruit_requests # ty: ignore[unresolved-import] + + interceptor.install(adafruit_requests.Session) + + if main_repository is not None: + identity = local_repo.repository_identity(interceptor.repository_url()) + name = "/".join(identity) if identity else "the main repo" + print( + f"matrixbox-simulator: serving {name} from {main_repository.root} " + "(--repo github for real GitHub)" + ) + + for key, repository in mapped_repositories.items(): + print(f"matrixbox-simulator: serving {key} from {repository.root}") + + +def _repository_mapping(value: str) -> tuple[str, Path]: + name, _, path = value.partition("=") + owner, _, repo = name.partition("/") + if not (owner and repo and path) or "/" in repo: + raise argparse.ArgumentTypeError(f"expected OWNER/REPO=PATH, got {value!r}") + + return local_repo.repository_key(owner, repo), Path(path).expanduser() + + # Set while stdin is in cbreak mode, so a SIGINT arriving mid-app can # still put the terminal back the way it found it — the hard exit below # can't rely on ordinary try/finally cleanup running. @@ -937,6 +1004,27 @@ def build_parser( "with 'n' after boot. Stays offline for the whole run" ), ) + parser.add_argument( + "--repo", + default=None, + metavar="PATH|github", + help=( + "where the main repo's GitHub requests (store, system update) are " + "answered from: a local checkout, including uncommitted changes. " + "Defaults to the checkout being run. `github` uses real GitHub" + ), + ) + parser.add_argument( + "--repo-map", + type=_repository_mapping, + action="append", + default=[], + metavar="OWNER/REPO=PATH", + help=( + "answer GitHub requests for another repo, such as a third-party " + "app, from a local directory. Repeatable" + ), + ) return parser diff --git a/tests/test_local_repo.py b/tests/test_local_repo.py new file mode 100644 index 0000000..4a1082e --- /dev/null +++ b/tests/test_local_repo.py @@ -0,0 +1,43 @@ +import json +from pathlib import Path + +import pytest + +from matrixbox_simulator.device import local_repo + +RAW = "https://raw.githubusercontent.com/MatrixBOX-dev/matrixbox" + + +@pytest.mark.parametrize( + ("repository_url", "request_url"), + [ + (f"{RAW}/refs/heads/main/", f"{RAW}/refs/heads/main/apps/clock/code.py"), + (f"{RAW}/refs/heads/main/", f"{RAW}/main/apps/clock/code.py"), + (f"{RAW}/refs/heads/main/", f"{RAW}/HEAD/apps/clock/code.py"), + ( + f"{RAW}/refs/heads/feat/store/", + f"{RAW}/refs/heads/feat/store/apps/clock/code.py", + ), + ], + ids=["refs-heads", "branch", "head", "branch-with-slash"], +) +def test_raw_url_serves_the_working_tree_file( + tmp_path: Path, repository_url: str, request_url: str +) -> None: + checkout = tmp_path / "matrixbox" + (checkout / "apps" / "clock").mkdir(parents=True) + (checkout / "apps" / "clock" / "code.py").write_text("print('clock')\n") + settings_path = tmp_path / "settings.txt" + settings_path.write_text(json.dumps({"repository_url": repository_url})) + interceptor = local_repo.GitHubInterceptor( + settings_path, + local_repo.LocalRepository(checkout), + {}, + lambda: True, + ) + + response = interceptor.respond("GET", request_url) + + assert response is not None + assert response.status_code == 200 + assert response.text == "print('clock')\n"