Merge branch 'master' into feat/issue-657-mcp-restart-path-inventory-guard
This commit is contained in:
@@ -0,0 +1,845 @@
|
||||
"""Integration tests for dirty same-claimant author-session rebind (#864).
|
||||
|
||||
Uses real temp git repos/worktrees and a temp GITEA_ISSUE_LOCK_DIR. Does not
|
||||
mutate any real #860/#864 worktree on disk.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import dirty_same_claimant_session_rebind as rebind # noqa: E402
|
||||
import issue_lock_provenance # noqa: E402
|
||||
import issue_lock_store as ils # noqa: E402
|
||||
import issue_lock_worktree # noqa: E402
|
||||
|
||||
ISSUE = 864
|
||||
BRANCH = f"fix/issue-{ISSUE}-dirty-same-claimant-session-rebind"
|
||||
REMOTE = "prgs"
|
||||
ORG = "Scaled-Tech-Consulting"
|
||||
REPO = "Gitea-Tools"
|
||||
IDENTITY = "jcwalker3"
|
||||
PROFILE = "prgs-author"
|
||||
|
||||
|
||||
def _git(cwd: str, *args: str, check: bool = True) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(
|
||||
["git", "-C", cwd, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=check,
|
||||
)
|
||||
|
||||
|
||||
def dead_pid() -> int:
|
||||
proc = subprocess.Popen([sys.executable, "-c", "pass"])
|
||||
proc.wait()
|
||||
return proc.pid
|
||||
|
||||
|
||||
def future_ts(hours: int = 4) -> str:
|
||||
return (
|
||||
(datetime.now(timezone.utc) + timedelta(hours=hours))
|
||||
.isoformat()
|
||||
.replace("+00:00", "Z")
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lock_dir(tmp_path, monkeypatch):
|
||||
d = tmp_path / "issue-locks"
|
||||
d.mkdir()
|
||||
monkeypatch.setenv("GITEA_ISSUE_LOCK_DIR", str(d))
|
||||
return str(d)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dirty_repo(tmp_path):
|
||||
"""Canonical repo root with branches/<name> worktree and dirty content."""
|
||||
root = tmp_path / "repo"
|
||||
root.mkdir()
|
||||
main = root / "main"
|
||||
main.mkdir()
|
||||
subprocess.run(["git", "init", "-q", str(main)], check=True, capture_output=True)
|
||||
_git(str(main), "config", "user.email", "t@t")
|
||||
_git(str(main), "config", "user.name", "t")
|
||||
(main / "README.md").write_text("base\n", encoding="utf-8")
|
||||
_git(str(main), "add", "README.md")
|
||||
_git(str(main), "commit", "-q", "-m", "base")
|
||||
_git(str(main), "branch", "-M", "master")
|
||||
|
||||
# Bare remote + origin tracking so remote head is observable offline.
|
||||
bare = tmp_path / "remote.git"
|
||||
subprocess.run(
|
||||
["git", "init", "--bare", "-q", str(bare)], check=True, capture_output=True
|
||||
)
|
||||
_git(str(main), "remote", "add", "origin", str(bare))
|
||||
_git(str(main), "push", "-q", "origin", "master:master")
|
||||
|
||||
branches = root / "branches"
|
||||
branches.mkdir()
|
||||
wt_name = f"fix-issue-{ISSUE}-dirty-same-claimant-session-rebind"
|
||||
wt = branches / wt_name
|
||||
_git(str(main), "worktree", "add", "-q", "-b", BRANCH, str(wt))
|
||||
_git(str(wt), "push", "-q", "-u", "origin", BRANCH)
|
||||
|
||||
# Seed committed files we will dirty.
|
||||
tracked = [
|
||||
"dirty_same_claimant_session_rebind.py",
|
||||
"issue_lock_provenance.py",
|
||||
"task_capability_map.py",
|
||||
]
|
||||
for rel in tracked:
|
||||
p = wt / rel
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text(f"seed {rel}\n", encoding="utf-8")
|
||||
_git(str(wt), "add", *tracked)
|
||||
_git(str(wt), "commit", "-q", "-m", "seed tracked")
|
||||
_git(str(wt), "push", "-q", "origin", BRANCH)
|
||||
|
||||
# Dirty tracked + untracked.
|
||||
for rel in tracked:
|
||||
(wt / rel).write_text(f"dirty {rel}\n", encoding="utf-8")
|
||||
untracked = [
|
||||
"tests/test_dirty_same_claimant_session_rebind.py",
|
||||
"docs/runbook-dirty-rebind.md",
|
||||
"scratch/notes-untracked.txt",
|
||||
"extra_untracked.txt",
|
||||
]
|
||||
for rel in untracked:
|
||||
p = wt / rel
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
p.write_text(f"untracked {rel}\n", encoding="utf-8")
|
||||
|
||||
inv = rebind.collect_dirty_inventory(str(wt))
|
||||
assert inv["ok"], inv.get("reasons")
|
||||
head = _git(str(wt), "rev-parse", "HEAD").stdout.strip()
|
||||
remote_head = _git(
|
||||
str(wt), "rev-parse", f"refs/remotes/origin/{BRANCH}"
|
||||
).stdout.strip()
|
||||
assert head == remote_head
|
||||
|
||||
return {
|
||||
"root": str(root),
|
||||
"main": str(main),
|
||||
"worktree": str(wt),
|
||||
"branch": BRANCH,
|
||||
"inventory": inv,
|
||||
"local_head": head,
|
||||
"remote_head": remote_head,
|
||||
"dirty_paths": list(inv["dirty_paths"]),
|
||||
"fingerprints": dict(inv["fingerprints"]),
|
||||
}
|
||||
|
||||
|
||||
def _make_lock(
|
||||
*,
|
||||
worktree: str,
|
||||
pid: int,
|
||||
lock_dir: str,
|
||||
identity: str = IDENTITY,
|
||||
profile: str = PROFILE,
|
||||
**overrides,
|
||||
) -> dict:
|
||||
lease = {
|
||||
"operation_type": ils.AUTHOR_ISSUE_WORK_LEASE,
|
||||
"issue_number": ISSUE,
|
||||
"branch": BRANCH,
|
||||
"worktree_path": worktree,
|
||||
"claimant": {"username": identity, "profile": profile},
|
||||
"created_at": "2026-01-01T00:00:00Z",
|
||||
"expires_at": future_ts(),
|
||||
"last_heartbeat_at": "2026-01-01T00:00:00Z",
|
||||
}
|
||||
lock = {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": worktree,
|
||||
"remote": REMOTE,
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"session_pid": pid,
|
||||
"pid": pid,
|
||||
"work_lease": lease,
|
||||
"lock_generation": 1,
|
||||
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
|
||||
tool="gitea_lock_issue",
|
||||
claimant={"username": identity, "profile": profile},
|
||||
),
|
||||
}
|
||||
lock.update(overrides)
|
||||
path = ils.lock_file_path(
|
||||
remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=lock_dir
|
||||
)
|
||||
lock["lock_file_path"] = path
|
||||
ils.save_lock_file(path, lock)
|
||||
# Stale session pointer for the dead owner.
|
||||
ptr = {
|
||||
"pid": pid,
|
||||
"lock_file_path": path,
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"remote": REMOTE,
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
}
|
||||
ils.save_lock_file(os.path.join(lock_dir, f"session-{pid}.json"), ptr)
|
||||
return ils.read_lock_file(path) or lock
|
||||
|
||||
|
||||
def _apply_kwargs(repo, lock, lock_dir, **overrides):
|
||||
kwargs = {
|
||||
"remote": REMOTE,
|
||||
"org": ORG,
|
||||
"repo": REPO,
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": repo["worktree"],
|
||||
"claimant_identity": IDENTITY,
|
||||
"claimant_profile": PROFILE,
|
||||
"old_pid": lock.get("session_pid") or lock.get("pid"),
|
||||
"expected_local_head": repo["local_head"],
|
||||
"expected_remote_head": repo["remote_head"],
|
||||
"expected_dirty_paths": repo["dirty_paths"],
|
||||
"expected_fingerprints": repo["fingerprints"],
|
||||
"existing_lock": lock,
|
||||
"current_identity": IDENTITY,
|
||||
"current_profile": PROFILE,
|
||||
"role_kind": "author",
|
||||
"current_pid": os.getpid(),
|
||||
"current_branch": BRANCH,
|
||||
"local_head": repo["local_head"],
|
||||
"remote_head": repo["remote_head"],
|
||||
"dirty_inventory": repo["inventory"],
|
||||
"competing_live_locks": [],
|
||||
"competing_sessions": [],
|
||||
"workflow_lease_active": False,
|
||||
"repo_root": repo["root"],
|
||||
"dry_run": False,
|
||||
"lock_dir": lock_dir,
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return kwargs
|
||||
|
||||
|
||||
# ── 1. Successful dead-PID same-claimant dirty rebind ───────────────────────
|
||||
|
||||
|
||||
def test_successful_dead_pid_same_claimant_dirty_rebind(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
|
||||
)
|
||||
assert result["success"], result
|
||||
assert result["outcome"] == rebind.REBIND_SANCTIONED
|
||||
assert result["old_pid"] == old
|
||||
assert result["new_pid"] == os.getpid()
|
||||
assert result["generation_after"] == result["generation_before"] + 1
|
||||
|
||||
rebound = ils.read_lock_file(result["lock_path"])
|
||||
assert rebound is not None
|
||||
assert int(rebound["session_pid"]) == os.getpid()
|
||||
assert int(rebound["pid"]) == os.getpid()
|
||||
assert (
|
||||
rebound.get("lock_provenance", {}).get("source")
|
||||
== issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND
|
||||
)
|
||||
assert rebound.get("rebind_record", {}).get("old_pid") == old
|
||||
|
||||
|
||||
# ── 2. Byte-for-byte preservation ───────────────────────────────────────────
|
||||
|
||||
|
||||
def test_byte_for_byte_preservation(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
before = {
|
||||
rel: rebind.content_fingerprint(os.path.join(dirty_repo["worktree"], rel))
|
||||
for rel in dirty_repo["dirty_paths"]
|
||||
}
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
|
||||
)
|
||||
assert result["success"], result
|
||||
after = {
|
||||
rel: rebind.content_fingerprint(os.path.join(dirty_repo["worktree"], rel))
|
||||
for rel in dirty_repo["dirty_paths"]
|
||||
}
|
||||
assert before == after
|
||||
assert result["fingerprints"] == before
|
||||
|
||||
|
||||
# ── 3. Exact dirty-path and fingerprint enforcement ─────────────────────────
|
||||
|
||||
|
||||
def test_extra_dirty_path_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
pins = list(dirty_repo["dirty_paths"])[:-1] # missing one observed path
|
||||
fps = {p: dirty_repo["fingerprints"][p] for p in pins}
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
expected_dirty_paths=pins,
|
||||
expected_fingerprints=fps,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("unexpected paths" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_missing_expected_dirty_path_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
pins = list(dirty_repo["dirty_paths"]) + ["not_really_dirty.txt"]
|
||||
fps = dict(dirty_repo["fingerprints"])
|
||||
fps["not_really_dirty.txt"] = "0" * 64
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
expected_dirty_paths=pins,
|
||||
expected_fingerprints=fps,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("missing expected" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_modified_fingerprint_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
fps = dict(dirty_repo["fingerprints"])
|
||||
victim = dirty_repo["dirty_paths"][0]
|
||||
fps[victim] = "f" * 64
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
expected_fingerprints=fps,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("fingerprint disagreement" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 4. Atomic session-pointer replacement ───────────────────────────────────
|
||||
|
||||
|
||||
def test_session_pointer_points_to_lock(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
|
||||
)
|
||||
assert result["success"], result
|
||||
new_ptr_path = os.path.join(lock_dir, f"session-{os.getpid()}.json")
|
||||
assert os.path.exists(new_ptr_path)
|
||||
ptr = ils.read_lock_file(new_ptr_path)
|
||||
assert ptr is not None
|
||||
assert os.path.realpath(ptr["lock_file_path"]) == os.path.realpath(
|
||||
result["lock_path"]
|
||||
)
|
||||
# Old pointer removed when it targeted this lock.
|
||||
old_ptr = os.path.join(lock_dir, f"session-{old}.json")
|
||||
assert not os.path.exists(old_ptr)
|
||||
assert result.get("removed_old_session_pointer") is True
|
||||
|
||||
|
||||
# ── 5. Retry after interruption (journal mid-state) ─────────────────────────
|
||||
|
||||
|
||||
def test_retry_after_journal_mid_state(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
jpath = rebind.journal_path(lock_dir, ISSUE)
|
||||
rebind._atomic_write_json(
|
||||
jpath,
|
||||
{
|
||||
"phase": rebind.JOURNAL_PHASE_PRE_BIND,
|
||||
"issue_number": ISSUE,
|
||||
"old_pid": old,
|
||||
"new_pid": os.getpid(),
|
||||
"expected_generation": 1,
|
||||
},
|
||||
)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
|
||||
)
|
||||
assert result["success"], result
|
||||
assert result["journal_phase"] == rebind.JOURNAL_PHASE_COMPLETE
|
||||
|
||||
# Second apply is already_rebound (retry-safe).
|
||||
rebound_lock = ils.read_lock_file(result["lock_path"])
|
||||
result2 = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
rebound_lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
existing_lock=rebound_lock,
|
||||
)
|
||||
)
|
||||
assert result2["success"], result2
|
||||
assert result2["already_rebound"] is True
|
||||
|
||||
|
||||
# ── 6. Active-PID refusal ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_active_pid_refused(dirty_repo, lock_dir):
|
||||
live = os.getpid()
|
||||
# Use a different "current" identity of session via fake current_pid...
|
||||
# Owner is live (this process). Rebind must refuse.
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=live, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=live,
|
||||
current_pid=live + 10_000_000, # distinct "new" session id for pin check
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("still alive" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 7. Foreign claimant refusal ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_foreign_claimant_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
current_identity="someone-else",
|
||||
claimant_identity="someone-else",
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("foreign claimant" in r or "does not match" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 8. Profile mismatch refusal ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_profile_mismatch_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
current_profile="other-profile",
|
||||
claimant_profile="other-profile",
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("profile" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 9. Competing session/lock/lease refusal ─────────────────────────────────
|
||||
|
||||
|
||||
def test_competing_live_lock_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
competing = [
|
||||
{
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": dirty_repo["worktree"] + "-other",
|
||||
"pid": os.getpid(),
|
||||
}
|
||||
]
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
competing_live_locks=competing,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("competing live lock" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_competing_session_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
competing_sessions=[
|
||||
{"pid": os.getpid(), "lock_file_path": lock["lock_file_path"], "live": True}
|
||||
],
|
||||
)
|
||||
)
|
||||
# current_pid is os.getpid(), so same session is skipped — use another live pid.
|
||||
# Spawn a long-lived process to act as competing live session.
|
||||
rival = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
|
||||
try:
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
competing_sessions=[
|
||||
{
|
||||
"pid": rival.pid,
|
||||
"lock_file_path": lock["lock_file_path"],
|
||||
"live": True,
|
||||
}
|
||||
],
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("competing live session" in r for r in result["reasons"])
|
||||
finally:
|
||||
rival.kill()
|
||||
rival.wait()
|
||||
|
||||
|
||||
def test_workflow_lease_active_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
workflow_lease_active=True,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("workflow lease" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 10. Local- and remote-head movement refusal ─────────────────────────────
|
||||
|
||||
|
||||
def test_local_head_movement_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
expected_local_head="b" * 40,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("local head" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_remote_head_movement_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
expected_remote_head="c" * 40,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("remote head" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 11. Path/symlink/registration mismatches ────────────────────────────────
|
||||
|
||||
|
||||
def test_worktree_not_under_branches_refused(dirty_repo, lock_dir, tmp_path):
|
||||
old = dead_pid()
|
||||
# Use a path outside branches/ as the declared worktree (still real dir).
|
||||
outside = tmp_path / "outside-wt"
|
||||
outside.mkdir()
|
||||
lock = _make_lock(worktree=str(outside), pid=old, lock_dir=lock_dir)
|
||||
# Inventory empty for outside path; use empty pins to hit path gate first
|
||||
# by providing matching empty-ish inventory after we force path checks.
|
||||
inv = {
|
||||
"dirty_paths": dirty_repo["dirty_paths"],
|
||||
"fingerprints": dirty_repo["fingerprints"],
|
||||
"ok": True,
|
||||
"reasons": [],
|
||||
}
|
||||
result = rebind.assess_dirty_same_claimant_session_rebind(
|
||||
remote=REMOTE,
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
issue_number=ISSUE,
|
||||
branch_name=BRANCH,
|
||||
worktree_path=str(outside),
|
||||
claimant_identity=IDENTITY,
|
||||
claimant_profile=PROFILE,
|
||||
old_pid=old,
|
||||
expected_local_head=dirty_repo["local_head"],
|
||||
expected_remote_head=dirty_repo["remote_head"],
|
||||
expected_dirty_paths=dirty_repo["dirty_paths"],
|
||||
expected_fingerprints=dirty_repo["fingerprints"],
|
||||
existing_lock=lock,
|
||||
current_identity=IDENTITY,
|
||||
current_profile=PROFILE,
|
||||
role_kind="author",
|
||||
current_pid=os.getpid(),
|
||||
current_branch=BRANCH,
|
||||
local_head=dirty_repo["local_head"],
|
||||
remote_head=dirty_repo["remote_head"],
|
||||
dirty_inventory=inv,
|
||||
repo_root=dirty_repo["root"],
|
||||
)
|
||||
assert not result["rebind_sanctioned"]
|
||||
assert any("branches/" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_lock_worktree_mismatch_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(
|
||||
worktree=dirty_repo["worktree"] + "-elsewhere",
|
||||
pid=old,
|
||||
lock_dir=lock_dir,
|
||||
)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("does not match declared" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 12. Malformed lock/session records ──────────────────────────────────────
|
||||
|
||||
|
||||
def test_malformed_lock_missing_pid_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
lock.pop("session_pid", None)
|
||||
lock.pop("pid", None)
|
||||
ils.save_lock_file(lock["lock_file_path"], lock)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("incomplete" in r or "session_pid" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_empty_old_pid_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=None)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("old_pid" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_reviewer_role_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, role_kind="reviewer")
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("reviewer" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_reconciler_without_authorize_refused(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(
|
||||
dirty_repo,
|
||||
lock,
|
||||
lock_dir,
|
||||
old_pid=old,
|
||||
role_kind="reconciler",
|
||||
authorize_reconciler_execute=False,
|
||||
)
|
||||
)
|
||||
assert not result["success"]
|
||||
assert any("authorize_reconciler_execute" in r for r in result["reasons"])
|
||||
|
||||
|
||||
# ── 13. No duplicate ownership after success or retry ───────────────────────
|
||||
|
||||
|
||||
def test_no_duplicate_ownership_after_success_or_retry(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
r1 = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
|
||||
)
|
||||
assert r1["success"], r1
|
||||
rebound = ils.read_lock_file(r1["lock_path"])
|
||||
r2 = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, rebound, lock_dir, old_pid=old, existing_lock=rebound)
|
||||
)
|
||||
assert r2["success"], r2
|
||||
assert r2["already_rebound"] is True
|
||||
|
||||
# Only one durable lock file for this issue; session pointer is current pid.
|
||||
# Skip session pointers and rebind journals (dotfiles / non-lock records).
|
||||
matching = []
|
||||
for p in ils.iter_lock_files(lock_dir):
|
||||
name = os.path.basename(p)
|
||||
if name.startswith(".") or name.startswith("session-"):
|
||||
continue
|
||||
rec = ils.read_lock_file(p)
|
||||
if not rec:
|
||||
continue
|
||||
if (
|
||||
rec.get("issue_number") == ISSUE
|
||||
and rec.get("remote") == REMOTE
|
||||
and rec.get("branch_name") == BRANCH
|
||||
and rec.get("session_pid") is not None
|
||||
):
|
||||
matching.append(rec)
|
||||
assert len(matching) == 1
|
||||
assert int(matching[0]["session_pid"]) == os.getpid()
|
||||
# No live session pointer for the dead old pid.
|
||||
assert not os.path.exists(os.path.join(lock_dir, f"session-{old}.json"))
|
||||
|
||||
|
||||
# ── 14. Ordinary dirty-worktree locking remains fail-closed ─────────────────
|
||||
|
||||
|
||||
def test_ordinary_dirty_lock_worktree_assessment_blocks(dirty_repo):
|
||||
porcelain = dirty_repo["inventory"]["porcelain_status"]
|
||||
assessment = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=dirty_repo["worktree"],
|
||||
current_branch=BRANCH,
|
||||
porcelain_status=porcelain,
|
||||
base_equivalent=False,
|
||||
)
|
||||
assert assessment["block"] is True
|
||||
assert any(
|
||||
"tracked file edits exist before issue lock" in r
|
||||
for r in assessment["reasons"]
|
||||
)
|
||||
|
||||
|
||||
# ── 15. Fixture matching #860 class with 7 fingerprint-pinned dirty paths ───
|
||||
|
||||
|
||||
def test_issue_860_regression_fixture_spec():
|
||||
spec = rebind.build_issue_860_regression_fixture_spec()
|
||||
assert spec["claimant_identity"] == "jcwalker3"
|
||||
assert spec["claimant_profile"] == "prgs-author"
|
||||
assert spec["old_pid_alive"] is False
|
||||
assert spec["live_session_pointer"] is None
|
||||
assert spec["dirty_path_count"] == 7
|
||||
assert len(spec["expected_dirty_paths"]) == 7
|
||||
assert len(spec["expected_fingerprints"]) == 7
|
||||
assert spec["expected_local_head"] == spec["expected_remote_head"]
|
||||
for path in spec["expected_dirty_paths"]:
|
||||
assert path in spec["expected_fingerprints"]
|
||||
assert len(spec["expected_fingerprints"][path]) == 64
|
||||
|
||||
|
||||
def test_dry_run_does_not_write(dirty_repo, lock_dir):
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
before = ils.read_lock_file(lock["lock_file_path"])
|
||||
result = rebind.apply_dirty_same_claimant_session_rebind(
|
||||
**_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dry_run=True)
|
||||
)
|
||||
assert result["success"], result
|
||||
assert result["dry_run"] is True
|
||||
after = ils.read_lock_file(lock["lock_file_path"])
|
||||
assert after["session_pid"] == before["session_pid"]
|
||||
assert not os.path.exists(os.path.join(lock_dir, f"session-{os.getpid()}.json"))
|
||||
|
||||
|
||||
def test_provenance_source_is_sanctioned():
|
||||
assert (
|
||||
issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND
|
||||
in issue_lock_provenance.SANCTIONED_LOCK_SOURCES
|
||||
)
|
||||
assessment = issue_lock_provenance.assess_lock_file_for_create_pr(
|
||||
{
|
||||
"work_lease": {"operation_type": "author_issue_work"},
|
||||
"lock_provenance": {
|
||||
"source": issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND,
|
||||
"written_by_tool": issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND,
|
||||
"written_at": "2026-01-01T00:00:00Z",
|
||||
},
|
||||
}
|
||||
)
|
||||
assert assessment["proven"] is True
|
||||
|
||||
|
||||
def test_permission_allowed_is_not_ownership_proof(dirty_repo, lock_dir):
|
||||
"""permission_allowed=True must not bypass foreign claimant refusal."""
|
||||
old = dead_pid()
|
||||
lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
|
||||
result = rebind.assess_dirty_same_claimant_session_rebind(
|
||||
remote=REMOTE,
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
issue_number=ISSUE,
|
||||
branch_name=BRANCH,
|
||||
worktree_path=dirty_repo["worktree"],
|
||||
claimant_identity=IDENTITY,
|
||||
claimant_profile=PROFILE,
|
||||
old_pid=old,
|
||||
expected_local_head=dirty_repo["local_head"],
|
||||
expected_remote_head=dirty_repo["remote_head"],
|
||||
expected_dirty_paths=dirty_repo["dirty_paths"],
|
||||
expected_fingerprints=dirty_repo["fingerprints"],
|
||||
existing_lock=lock,
|
||||
current_identity="intruder",
|
||||
current_profile=PROFILE,
|
||||
role_kind="author",
|
||||
current_pid=os.getpid(),
|
||||
current_branch=BRANCH,
|
||||
local_head=dirty_repo["local_head"],
|
||||
remote_head=dirty_repo["remote_head"],
|
||||
dirty_inventory=dirty_repo["inventory"],
|
||||
permission_allowed=True,
|
||||
repo_root=dirty_repo["root"],
|
||||
)
|
||||
assert not result["rebind_sanctioned"]
|
||||
assert any("does not match active identity" in r for r in result["reasons"])
|
||||
|
||||
|
||||
def test_content_fingerprint_stable(tmp_path):
|
||||
p = tmp_path / "f.txt"
|
||||
p.write_bytes(b"abc123")
|
||||
a = rebind.content_fingerprint(str(p))
|
||||
b = rebind.content_fingerprint(str(p))
|
||||
assert a == b
|
||||
assert len(a) == 64
|
||||
Reference in New Issue
Block a user