mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-19 02:28:59 +00:00
Fix registry audit repo identity in worktrees
This commit is contained in:
rcourtman
parent
fbbf1daf4c
commit
df55901611
4 changed files with 140 additions and 9 deletions
|
|
@ -20,7 +20,7 @@ from canonical_completion_guard import (
|
|||
subsystem_matches_path,
|
||||
)
|
||||
from control_plane import DEFAULT_CONTROL_PLANE
|
||||
from repo_file_io import load_repo_json
|
||||
from repo_file_io import canonical_repo_id, canonical_workspace_repos_root, load_repo_json
|
||||
from status_audit import load_status_payload
|
||||
|
||||
|
||||
|
|
@ -62,12 +62,13 @@ def sorted_casefold(values: list[str]) -> list[str]:
|
|||
|
||||
|
||||
def tracked_repo_files() -> set[str]:
|
||||
return tracked_workspace_files(active_repos=[REPO_ROOT.name], local_repo=REPO_ROOT.name)
|
||||
local_repo = canonical_repo_id(REPO_ROOT)
|
||||
return tracked_workspace_files(active_repos=[local_repo], local_repo=local_repo)
|
||||
|
||||
|
||||
def tracked_workspace_files(*, active_repos: list[str], local_repo: str) -> set[str]:
|
||||
files: set[str] = set()
|
||||
repos_root = REPO_ROOT.parent
|
||||
repos_root = canonical_workspace_repos_root(REPO_ROOT)
|
||||
for repo_id in active_repos:
|
||||
repo_root = REPO_ROOT if repo_id == local_repo else repos_root / repo_id
|
||||
if not repo_root.exists():
|
||||
|
|
@ -633,7 +634,8 @@ def main(argv: list[str] | None = None) -> int:
|
|||
repo_id
|
||||
for repo_id in scope.get("active_repos", [])
|
||||
if isinstance(repo_id, str) and repo_id.strip()
|
||||
] or [REPO_ROOT.name]
|
||||
] or [canonical_repo_id(REPO_ROOT)]
|
||||
local_repo = canonical_repo_id(REPO_ROOT)
|
||||
lane_ids = {
|
||||
lane.get("id")
|
||||
for lane in status_payload.get("lanes", [])
|
||||
|
|
@ -641,7 +643,7 @@ def main(argv: list[str] | None = None) -> int:
|
|||
}
|
||||
report = audit_registry_payload(
|
||||
load_registry_payload(staged=args.staged),
|
||||
tracked_files=tracked_workspace_files(active_repos=active_repos, local_repo=REPO_ROOT.name),
|
||||
tracked_files=tracked_workspace_files(active_repos=active_repos, local_repo=local_repo),
|
||||
status_lane_ids=lane_ids,
|
||||
schema_contract=registry_schema_contract(staged=args.staged),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,14 +1,77 @@
|
|||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from registry_audit import audit_registry_payload, parse_args
|
||||
import registry_audit
|
||||
from repo_file_io import canonical_repo_id
|
||||
from registry_audit import audit_registry_payload, parse_args, tracked_workspace_files
|
||||
|
||||
|
||||
class RegistryAuditTest(unittest.TestCase):
|
||||
def git(self, repo_root: Path, *args: str) -> subprocess.CompletedProcess:
|
||||
env = os.environ.copy()
|
||||
env.pop("GIT_INDEX_FILE", None)
|
||||
return subprocess.run(
|
||||
["git", *args],
|
||||
cwd=repo_root,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
)
|
||||
|
||||
def test_parse_args_accepts_staged_flag(self) -> None:
|
||||
args = parse_args(["--check", "--staged"])
|
||||
self.assertTrue(args.check)
|
||||
self.assertTrue(args.staged)
|
||||
|
||||
def test_tracked_workspace_files_uses_linked_worktree_as_canonical_local_repo(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
workspace = Path(tmpdir) / "workspace"
|
||||
repo_root = workspace / "repos" / "pulse"
|
||||
linked_worktree = workspace / ".worktrees" / "pulse-first-session-onboarding-parity"
|
||||
repo_root.mkdir(parents=True)
|
||||
linked_worktree.parent.mkdir(parents=True)
|
||||
|
||||
self.git(repo_root, "init")
|
||||
(repo_root / "internal").mkdir()
|
||||
(repo_root / "internal" / "existing.go").write_text("package internal\n", encoding="utf-8")
|
||||
self.git(repo_root, "add", "internal/existing.go")
|
||||
self.git(
|
||||
repo_root,
|
||||
"-c",
|
||||
"user.name=Pulse Test",
|
||||
"-c",
|
||||
"user.email=pulse-test@example.invalid",
|
||||
"commit",
|
||||
"-m",
|
||||
"initial",
|
||||
)
|
||||
self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD")
|
||||
|
||||
(linked_worktree / "internal" / "staged.go").write_text("package internal\n", encoding="utf-8")
|
||||
self.git(linked_worktree, "add", "internal/staged.go")
|
||||
contracts_dir = linked_worktree / "docs" / "release-control" / "v6" / "internal" / "subsystems"
|
||||
contracts_dir.mkdir(parents=True)
|
||||
|
||||
with patch("registry_audit.REPO_ROOT", linked_worktree), patch(
|
||||
"registry_audit.DEFAULT_CONTROL_PLANE",
|
||||
{
|
||||
**registry_audit.DEFAULT_CONTROL_PLANE,
|
||||
"subsystems_dir_path": str(contracts_dir),
|
||||
},
|
||||
):
|
||||
files = tracked_workspace_files(
|
||||
active_repos=["pulse"],
|
||||
local_repo=canonical_repo_id(linked_worktree),
|
||||
)
|
||||
|
||||
self.assertIn("internal/staged.go", files)
|
||||
self.assertNotIn("pulse:internal/staged.go", files)
|
||||
|
||||
def test_audit_registry_payload_accepts_valid_minimal_registry(self) -> None:
|
||||
payload = {
|
||||
"version": 13,
|
||||
|
|
|
|||
|
|
@ -14,13 +14,45 @@ REPO_ROOT = Path(__file__).resolve().parents[2]
|
|||
DEFAULT_REPO_ROOT = REPO_ROOT
|
||||
|
||||
|
||||
def git_env() -> dict[str, str]:
|
||||
def git_env(repo_root: Path | None = None) -> dict[str, str]:
|
||||
env = os.environ.copy()
|
||||
if REPO_ROOT != DEFAULT_REPO_ROOT:
|
||||
root = repo_root or REPO_ROOT
|
||||
if root != DEFAULT_REPO_ROOT:
|
||||
env.pop("GIT_INDEX_FILE", None)
|
||||
return env
|
||||
|
||||
|
||||
def git_common_dir(repo_root: Path | None = None) -> Path:
|
||||
root = (repo_root or REPO_ROOT).resolve()
|
||||
result = subprocess.run(
|
||||
["git", "rev-parse", "--path-format=absolute", "--git-common-dir"],
|
||||
cwd=root,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=git_env(root),
|
||||
)
|
||||
common_dir = Path(result.stdout.strip())
|
||||
if not common_dir.is_absolute():
|
||||
common_dir = root / common_dir
|
||||
return common_dir.resolve()
|
||||
|
||||
|
||||
def canonical_repo_root(repo_root: Path | None = None) -> Path:
|
||||
common_dir = git_common_dir(repo_root)
|
||||
if common_dir.name == ".git":
|
||||
return common_dir.parent.resolve()
|
||||
return (repo_root or REPO_ROOT).resolve()
|
||||
|
||||
|
||||
def canonical_repo_id(repo_root: Path | None = None) -> str:
|
||||
return canonical_repo_root(repo_root).name
|
||||
|
||||
|
||||
def canonical_workspace_repos_root(repo_root: Path | None = None) -> Path:
|
||||
return canonical_repo_root(repo_root).parent
|
||||
|
||||
|
||||
def repo_relative_path(path: str | Path) -> str:
|
||||
candidate = Path(path)
|
||||
if candidate.is_absolute():
|
||||
|
|
|
|||
|
|
@ -6,7 +6,14 @@ import unittest
|
|||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from repo_file_io import load_repo_json, missing_staged_repo_paths, read_repo_text
|
||||
from repo_file_io import (
|
||||
canonical_repo_id,
|
||||
canonical_repo_root,
|
||||
canonical_workspace_repos_root,
|
||||
load_repo_json,
|
||||
missing_staged_repo_paths,
|
||||
read_repo_text,
|
||||
)
|
||||
|
||||
|
||||
class RepoFileIoTest(unittest.TestCase):
|
||||
|
|
@ -72,6 +79,33 @@ class RepoFileIoTest(unittest.TestCase):
|
|||
with patch("repo_file_io.REPO_ROOT", repo_root):
|
||||
self.assertEqual(missing_staged_repo_paths([staged_rel, missing_rel]), [missing_rel])
|
||||
|
||||
def test_canonical_repo_identity_uses_git_common_dir_for_linked_worktree(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
workspace = Path(tmpdir) / "workspace"
|
||||
repo_root = workspace / "repos" / "pulse"
|
||||
linked_worktree = workspace / ".worktrees" / "pulse-first-session-onboarding-parity"
|
||||
repo_root.mkdir(parents=True)
|
||||
linked_worktree.parent.mkdir(parents=True)
|
||||
|
||||
self.git(repo_root, "init")
|
||||
(repo_root / "README.md").write_text("pulse\n", encoding="utf-8")
|
||||
self.git(repo_root, "add", "README.md")
|
||||
self.git(
|
||||
repo_root,
|
||||
"-c",
|
||||
"user.name=Pulse Test",
|
||||
"-c",
|
||||
"user.email=pulse-test@example.invalid",
|
||||
"commit",
|
||||
"-m",
|
||||
"initial",
|
||||
)
|
||||
self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD")
|
||||
|
||||
self.assertEqual(canonical_repo_root(linked_worktree), repo_root.resolve())
|
||||
self.assertEqual(canonical_repo_id(linked_worktree), "pulse")
|
||||
self.assertEqual(canonical_workspace_repos_root(linked_worktree), (workspace / "repos").resolve())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue