mattrguerra/better-memory-pipelinemainPrivate on GitHub, published here in full

helpers.py

Language Python2.4 KB74 lines

tests/helpers.py

  1. """Shared test helpers: load a gate module fresh with its env, and copy the
  2. sample vaults into a temp dir so tests can edit, delete and symlink freely."""
  3. import contextlib
  4. import importlib.util
  5. import io
  6. import itertools
  7. import os
  8. import shutil
  9. import tempfile
  10. from pathlib import Path
  11. REPO = Path(__file__).resolve().parent.parent
  12. SAMPLE = REPO / "sample-vaults"
  13. _counter = itertools.count()
  14. def load(script: str, env: dict):
  15. """Import REPO/<script> as a fresh module with `env` applied. The gates read
  16. their configuration from the environment at import time."""
  17. old = {k: os.environ.get(k) for k in env}
  18. os.environ.update(env)
  19. try:
  20. spec = importlib.util.spec_from_file_location(
  21. f"_bm_{Path(script).stem}_{next(_counter)}", REPO / script)
  22. mod = importlib.util.module_from_spec(spec)
  23. spec.loader.exec_module(mod)
  24. return mod
  25. finally:
  26. for k, v in old.items():
  27. if v is None:
  28. os.environ.pop(k, None)
  29. else:
  30. os.environ[k] = v
  31. def run(mod) -> tuple[int, str]:
  32. """Run mod.main(), capturing stdout."""
  33. buf = io.StringIO()
  34. with contextlib.redirect_stdout(buf):
  35. code = mod.main()
  36. return code, buf.getvalue()
  37. class TempVaults:
  38. """A throwaway copy of the sample vaults plus a staging root."""
  39. def __init__(self):
  40. self.dir = Path(tempfile.mkdtemp(prefix="bm-test-"))
  41. self.vaults = self.dir / "vaults"
  42. shutil.copytree(SAMPLE, self.vaults)
  43. # A .git folder inside a nested vault cannot be committed to this repo,
  44. # so the fixture creates it here.
  45. git = self.vaults / "workbench" / "Efforts" / "garden-studio" / ".git"
  46. git.mkdir(parents=True)
  47. (git / "notes.md").write_text("# git internals\n\nWALLED at any depth.\n")
  48. self.workbench = self.vaults / "workbench"
  49. self.identity = self.vaults / "identity"
  50. self.staging_root = self.dir / "cowork-staging"
  51. @property
  52. def env(self) -> dict:
  53. return {"VAULT_ROOTS": f"{self.workbench}:{self.identity}",
  54. "STAGING_DIR": str(self.staging_root), "EXTRA_DENY": ""}
  55. @property
  56. def staged(self) -> list[str]:
  57. d = self.staging_root / "staging"
  58. return sorted(p.name for p in d.iterdir()) if d.exists() else []
  59. def gate(self):
  60. return load("gate.py", self.env)
  61. def cleanup(self):
  62. shutil.rmtree(self.dir, ignore_errors=True)