import json import os import re import shutil import subprocess import sys import tarfile import tempfile import unittest import zipfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] PORTABLE_DIR = ROOT / "packaging" / "portable" if str(PORTABLE_DIR) not in sys.path: sys.path.insert(0, str(PORTABLE_DIR)) import portable_installer as portable def _project_version() -> str: text = (ROOT / "pyproject.toml").read_text(encoding="utf-8") match = re.search(r'(?m)^version\s*=\s*"([^"]+)"\s*$', text) if not match: raise RuntimeError("project version not found") return match.group(1) def _write_file(path: Path, content: str, *, mode: int | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") if mode is not None: path.chmod(mode) def _build_fake_wheel(root: Path, version: str) -> Path: root.mkdir(parents=True, exist_ok=True) wheel_path = root / f"aman-{version}-py3-none-any.whl" dist_info = f"aman-{version}.dist-info" module_code = f'VERSION = "{version}"\n\ndef main():\n print(VERSION)\n return 0\n' with zipfile.ZipFile(wheel_path, "w") as archive: archive.writestr("portable_test_app.py", module_code) archive.writestr( f"{dist_info}/METADATA", "\n".join( [ "Metadata-Version: 2.1", "Name: aman", f"Version: {version}", "Summary: portable bundle test wheel", "", ] ), ) archive.writestr( f"{dist_info}/WHEEL", "\n".join( [ "Wheel-Version: 1.0", "Generator: test_portable_bundle", "Root-Is-Purelib: true", "Tag: py3-none-any", "", ] ), ) archive.writestr( f"{dist_info}/entry_points.txt", "[console_scripts]\naman=portable_test_app:main\n", ) archive.writestr(f"{dist_info}/RECORD", "") return wheel_path def _bundle_dir(root: Path, version: str) -> Path: bundle_dir = root / f"bundle-{version}" (bundle_dir / "wheelhouse" / "common").mkdir(parents=True, exist_ok=True) (bundle_dir / "requirements").mkdir(parents=True, exist_ok=True) for tag in portable.SUPPORTED_PYTHON_TAGS: (bundle_dir / "wheelhouse" / tag).mkdir(parents=True, exist_ok=True) (bundle_dir / "requirements" / f"{tag}.txt").write_text("", encoding="utf-8") (bundle_dir / "systemd").mkdir(parents=True, exist_ok=True) shutil.copy2(PORTABLE_DIR / "install.sh", bundle_dir / "install.sh") shutil.copy2(PORTABLE_DIR / "uninstall.sh", bundle_dir / "uninstall.sh") shutil.copy2(PORTABLE_DIR / "portable_installer.py", bundle_dir / "portable_installer.py") shutil.copy2(PORTABLE_DIR / "systemd" / "aman.service.in", bundle_dir / "systemd" / "aman.service.in") portable.write_manifest(version, bundle_dir / "manifest.json") payload = json.loads((bundle_dir / "manifest.json").read_text(encoding="utf-8")) payload["smoke_check_code"] = "import portable_test_app" (bundle_dir / "manifest.json").write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) shutil.copy2(_build_fake_wheel(root / "wheelhouse", version), bundle_dir / "wheelhouse" / "common") for name in ("install.sh", "uninstall.sh", "portable_installer.py"): (bundle_dir / name).chmod(0o755) return bundle_dir def _systemctl_env(home: Path, *, extra_path: list[Path] | None = None, fail_match: str | None = None) -> tuple[dict[str, str], Path]: fake_bin = home / "test-bin" fake_bin.mkdir(parents=True, exist_ok=True) log_path = home / "systemctl.log" script_path = fake_bin / "systemctl" _write_file( script_path, "\n".join( [ "#!/usr/bin/env python3", "import os", "import sys", "from pathlib import Path", "log_path = Path(os.environ['SYSTEMCTL_LOG'])", "log_path.parent.mkdir(parents=True, exist_ok=True)", "command = ' '.join(sys.argv[1:])", "with log_path.open('a', encoding='utf-8') as handle:", " handle.write(command + '\\n')", "fail_match = os.environ.get('SYSTEMCTL_FAIL_MATCH', '')", "if fail_match and fail_match in command:", " print(f'forced failure: {command}', file=sys.stderr)", " raise SystemExit(1)", "raise SystemExit(0)", "", ] ), mode=0o755, ) search_path = [ str(home / ".local" / "bin"), *(str(path) for path in (extra_path or [])), str(fake_bin), os.environ["PATH"], ] env = os.environ.copy() env["HOME"] = str(home) env["PATH"] = os.pathsep.join(search_path) env["SYSTEMCTL_LOG"] = str(log_path) env["AMAN_PORTABLE_TEST_PYTHON_TAG"] = "cp311" if fail_match: env["SYSTEMCTL_FAIL_MATCH"] = fail_match else: env.pop("SYSTEMCTL_FAIL_MATCH", None) return env, log_path def _run_script(bundle_dir: Path, script_name: str, env: dict[str, str], *args: str, check: bool = True) -> subprocess.CompletedProcess[str]: return subprocess.run( ["bash", str(bundle_dir / script_name), *args], cwd=bundle_dir, env=env, text=True, capture_output=True, check=check, ) def _manifest_with_supported_tags(bundle_dir: Path, tags: list[str]) -> None: manifest_path = bundle_dir / "manifest.json" payload = json.loads(manifest_path.read_text(encoding="utf-8")) payload["supported_python_tags"] = tags manifest_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") def _installed_version(home: Path) -> str: installed_python = home / ".local" / "share" / "aman" / "current" / "venv" / "bin" / "python" result = subprocess.run( [str(installed_python), "-c", "import portable_test_app; print(portable_test_app.VERSION)"], text=True, capture_output=True, check=True, ) return result.stdout.strip() class PortableBundleTests(unittest.TestCase): def test_package_portable_builds_bundle_and_checksum(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) dist_dir = tmp_path / "dist" build_dir = tmp_path / "build" stale_build_module = build_dir / "lib" / "desktop_wayland.py" test_wheelhouse = tmp_path / "wheelhouse" for tag in portable.SUPPORTED_PYTHON_TAGS: target_dir = test_wheelhouse / tag target_dir.mkdir(parents=True, exist_ok=True) _write_file(target_dir / f"{tag}-placeholder.whl", "placeholder\n") _write_file(stale_build_module, "stale = True\n") env = os.environ.copy() env["DIST_DIR"] = str(dist_dir) env["BUILD_DIR"] = str(build_dir) env["AMAN_PORTABLE_TEST_WHEELHOUSE_ROOT"] = str(test_wheelhouse) env["UV_CACHE_DIR"] = str(tmp_path / ".uv-cache") env["PIP_CACHE_DIR"] = str(tmp_path / ".pip-cache") subprocess.run( ["bash", "./scripts/package_portable.sh"], cwd=ROOT, env=env, text=True, capture_output=True, check=True, ) version = _project_version() tarball = dist_dir / f"aman-x11-linux-{version}.tar.gz" checksum = dist_dir / f"aman-x11-linux-{version}.tar.gz.sha256" wheel_path = dist_dir / f"aman-{version}-py3-none-any.whl" self.assertTrue(tarball.exists()) self.assertTrue(checksum.exists()) self.assertTrue(wheel_path.exists()) prefix = f"aman-x11-linux-{version}" with zipfile.ZipFile(wheel_path) as archive: wheel_names = set(archive.namelist()) metadata_path = f"aman-{version}.dist-info/METADATA" metadata = archive.read(metadata_path).decode("utf-8") self.assertNotIn("desktop_wayland.py", wheel_names) self.assertNotIn("Requires-Dist: pillow", metadata) self.assertNotIn("Requires-Dist: PyGObject", metadata) self.assertNotIn("Requires-Dist: python-xlib", metadata) with tarfile.open(tarball, "r:gz") as archive: names = set(archive.getnames()) requirements_path = f"{prefix}/requirements/cp311.txt" requirements_member = archive.extractfile(requirements_path) if requirements_member is None: self.fail(f"missing {requirements_path} in portable archive") requirements_text = requirements_member.read().decode("utf-8") self.assertIn(f"{prefix}/install.sh", names) self.assertIn(f"{prefix}/uninstall.sh", names) self.assertIn(f"{prefix}/portable_installer.py", names) self.assertIn(f"{prefix}/manifest.json", names) self.assertIn(f"{prefix}/wheelhouse/common", names) self.assertIn(f"{prefix}/wheelhouse/cp310", names) self.assertIn(f"{prefix}/wheelhouse/cp311", names) self.assertIn(f"{prefix}/wheelhouse/cp312", names) self.assertIn(f"{prefix}/requirements/cp310.txt", names) self.assertIn(f"{prefix}/requirements/cp311.txt", names) self.assertIn(f"{prefix}/requirements/cp312.txt", names) self.assertIn(f"{prefix}/systemd/aman.service.in", names) self.assertNotIn("pygobject", requirements_text.lower()) self.assertNotIn("python-xlib", requirements_text.lower()) def test_fresh_install_creates_managed_paths_and_starts_service(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) home = tmp_path / "home" bundle_dir = _bundle_dir(tmp_path, "0.1.0") env, log_path = _systemctl_env(home) result = _run_script(bundle_dir, "install.sh", env) self.assertIn("installed aman 0.1.0", result.stdout) current_link = home / ".local" / "share" / "aman" / "current" self.assertTrue(current_link.is_symlink()) self.assertEqual(current_link.resolve().name, "0.1.0") self.assertEqual(_installed_version(home), "0.1.0") shim_path = home / ".local" / "bin" / "aman" service_path = home / ".config" / "systemd" / "user" / "aman.service" state_path = home / ".local" / "share" / "aman" / "install-state.json" self.assertIn(portable.MANAGED_MARKER, shim_path.read_text(encoding="utf-8")) service_text = service_path.read_text(encoding="utf-8") self.assertIn(portable.MANAGED_MARKER, service_text) self.assertIn(str(current_link / "venv" / "bin" / "aman"), service_text) payload = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(payload["version"], "0.1.0") commands = log_path.read_text(encoding="utf-8") self.assertIn("--user daemon-reload", commands) self.assertIn("--user enable --now aman", commands) def test_upgrade_preserves_config_and_cache_and_prunes_old_version(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) home = tmp_path / "home" env, _log_path = _systemctl_env(home) bundle_v1 = _bundle_dir(tmp_path / "v1", "0.1.0") bundle_v2 = _bundle_dir(tmp_path / "v2", "0.2.0") _run_script(bundle_v1, "install.sh", env) config_path = home / ".config" / "aman" / "config.json" cache_path = home / ".cache" / "aman" / "models" / "cached.bin" _write_file(config_path, '{"config_version": 1}\n') _write_file(cache_path, "cache\n") _run_script(bundle_v2, "install.sh", env) current_link = home / ".local" / "share" / "aman" / "current" self.assertEqual(current_link.resolve().name, "0.2.0") self.assertEqual(_installed_version(home), "0.2.0") self.assertFalse((home / ".local" / "share" / "aman" / "0.1.0").exists()) self.assertTrue(config_path.exists()) self.assertTrue(cache_path.exists()) def test_unmanaged_shim_conflict_fails_before_mutation(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) home = tmp_path / "home" bundle_dir = _bundle_dir(tmp_path, "0.1.0") env, _log_path = _systemctl_env(home) _write_file(home / ".local" / "bin" / "aman", "#!/usr/bin/env bash\necho nope\n", mode=0o755) result = _run_script(bundle_dir, "install.sh", env, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("unmanaged shim", result.stderr) self.assertFalse((home / ".local" / "share" / "aman" / "install-state.json").exists()) def test_manifest_supported_tag_mismatch_fails_before_mutation(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) home = tmp_path / "home" bundle_dir = _bundle_dir(tmp_path, "0.1.0") _manifest_with_supported_tags(bundle_dir, ["cp399"]) env, _log_path = _systemctl_env(home) result = _run_script(bundle_dir, "install.sh", env, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("unsupported python3 version", result.stderr) self.assertFalse((home / ".local" / "share" / "aman").exists()) def test_uninstall_preserves_config_and_cache_by_default(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) home = tmp_path / "home" bundle_dir = _bundle_dir(tmp_path, "0.1.0") env, log_path = _systemctl_env(home) _run_script(bundle_dir, "install.sh", env) _write_file(home / ".config" / "aman" / "config.json", '{"config_version": 1}\n') _write_file(home / ".cache" / "aman" / "models" / "cached.bin", "cache\n") result = _run_script(bundle_dir, "uninstall.sh", env) self.assertIn("uninstalled aman portable bundle", result.stdout) self.assertFalse((home / ".local" / "share" / "aman").exists()) self.assertFalse((home / ".local" / "bin" / "aman").exists()) self.assertFalse((home / ".config" / "systemd" / "user" / "aman.service").exists()) self.assertTrue((home / ".config" / "aman" / "config.json").exists()) self.assertTrue((home / ".cache" / "aman" / "models" / "cached.bin").exists()) commands = log_path.read_text(encoding="utf-8") self.assertIn("--user disable --now aman", commands) def test_uninstall_purge_removes_config_and_cache(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) home = tmp_path / "home" bundle_dir = _bundle_dir(tmp_path, "0.1.0") env, _log_path = _systemctl_env(home) _run_script(bundle_dir, "install.sh", env) _write_file(home / ".config" / "aman" / "config.json", '{"config_version": 1}\n') _write_file(home / ".cache" / "aman" / "models" / "cached.bin", "cache\n") _run_script(bundle_dir, "uninstall.sh", env, "--purge") self.assertFalse((home / ".config" / "aman").exists()) self.assertFalse((home / ".cache" / "aman").exists()) def test_upgrade_rolls_back_when_service_restart_fails(self): with tempfile.TemporaryDirectory() as tmp: tmp_path = Path(tmp) home = tmp_path / "home" bundle_v1 = _bundle_dir(tmp_path / "v1", "0.1.0") bundle_v2 = _bundle_dir(tmp_path / "v2", "0.2.0") good_env, _ = _systemctl_env(home) failing_env, _ = _systemctl_env(home, fail_match="enable --now aman") _run_script(bundle_v1, "install.sh", good_env) result = _run_script(bundle_v2, "install.sh", failing_env, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("forced failure", result.stderr) self.assertEqual((home / ".local" / "share" / "aman" / "current").resolve().name, "0.1.0") self.assertEqual(_installed_version(home), "0.1.0") self.assertFalse((home / ".local" / "share" / "aman" / "0.2.0").exists()) payload = json.loads( (home / ".local" / "share" / "aman" / "install-state.json").read_text(encoding="utf-8") ) self.assertEqual(payload["version"], "0.1.0") if __name__ == "__main__": unittest.main()