weichen
93ac63fb3d
Why: Replace repeated submodule source pointers with consumer-owned language versions and fixed application sources, keeping independent repository edits and existing branches intact. Evidence: SELF task receipts contain startup/sync evidence, real Go and npm consumption, Native overlay/unlink, safe transition/ref preservation, credential boundary and historical build regressions. Exact committed full composition and RDM checks follow this checkpoint; formal acceptance and GATE CI remain pending. Impact: Add isolated build/test composition, source provenance and safe owner refresh, adapt governance and service build callers, remove replaced index links while retaining original Git metadata, and adopt reusable dependency standards. Refs: Documents approved dependency-workspace r1/R1-R15 and U1-U7 implementation plan; feature/dependency-workspace task card.
404 lines
24 KiB
Python
404 lines
24 KiB
Python
"""Task cutover protections with real repositories, gitlinks and worktrees."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
SOURCE = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class WorkspaceCutoverTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory(prefix="workspace-cutover-")
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.base = Path(self.tmp.name)
|
|
self.repo = self.base / "repo"
|
|
self.env = os.environ.copy()
|
|
for key in list(self.env):
|
|
if key.startswith("GIT_CONFIG_") or key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GOWORK", "PRECHECK_RECEIPT", "RECEIPT", "CMD", "IMPORT_RECEIPT", "COMPOSITION_SOURCE_MAP", "CI_DEPENDENCY_MIRRORS"):
|
|
self.env.pop(key, None)
|
|
self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null",
|
|
"GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0",
|
|
"GIT_AUTHOR_NAME": "Cutover Test", "GIT_AUTHOR_EMAIL": "test@example.invalid",
|
|
"GIT_COMMITTER_NAME": "Cutover Test", "GIT_COMMITTER_EMAIL": "test@example.invalid",
|
|
"XDG_CACHE_HOME": str(self.base / "cache")})
|
|
self.call(self.base, "git", "init", "--initial-branch=main", str(self.repo))
|
|
shutil.copytree(SOURCE / "scripts", self.repo / "scripts")
|
|
(self.repo / "Makefile").write_text("precheck:\n\t@bash scripts/task.sh precheck\n")
|
|
(self.repo / ".gitignore").write_text(".workspace/\ngo.work\ngo.work.sum\nmodules/\n")
|
|
(self.repo / "tracked").write_text("root\n")
|
|
self.git(self.repo, "add", ".")
|
|
self.git(self.repo, "commit", "-m", "Old project")
|
|
self.remote = self.base / "project.git"
|
|
self.call(self.base, "git", "init", "--bare", "--initial-branch=main", str(self.remote))
|
|
self.git(self.remote, "config", "gc.auto", "0")
|
|
self.git(self.remote, "config", "receive.autogc", "false")
|
|
self.git(self.repo, "remote", "add", "origin", str(self.remote))
|
|
self.git(self.repo, "push", "-u", "origin", "main")
|
|
self.app, self.app_sha = self.component("App", {"content": "app v1\n", ".gitignore": "private/\n"})
|
|
self.rules, self.rules_sha = self.component("Rules", {"core/baseline.md": "# Baseline\n", "workflows/delivery.md": "# Delivery\n"})
|
|
|
|
def call(self, cwd, *args, check=True, env=None):
|
|
result = subprocess.run(args, cwd=cwd, env=env or self.env, text=True,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60)
|
|
if check:
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
return result
|
|
|
|
def git(self, cwd, *args, **kwargs):
|
|
return self.call(cwd, "git", *args, **kwargs)
|
|
|
|
def task(self, cwd, command, *args, **kwargs):
|
|
return self.call(cwd, "bash", str(cwd / "scripts/task.sh"), command, *args, **kwargs)
|
|
|
|
def component(self, name, files):
|
|
path = self.base / name
|
|
self.call(self.base, "git", "init", "--initial-branch=main", str(path))
|
|
for relative, content in files.items():
|
|
target = path / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(content)
|
|
self.git(path, "add", ".")
|
|
self.git(path, "commit", "-m", name + " v1")
|
|
return path, self.git(path, "rev-parse", "HEAD").stdout.strip()
|
|
|
|
def adopt(self):
|
|
records = [{"id": name, "repo": str(path), "path": ".workspace/" + name,
|
|
"legacy_path": "modules/" + name, "role": role, "modules": [], "target": "main"}
|
|
for name, path, role in (("App", self.app, "application"), ("Rules", self.rules, "guidelines"))]
|
|
(self.repo / "components.json").write_text(json.dumps({"schema": 1, "components": records}))
|
|
(self.repo / "components.lock.json").write_text(json.dumps({"schema": 1, "components": {"App": {"commit": self.app_sha}, "Rules": {"commit": self.rules_sha}}}))
|
|
self.git(self.repo, "add", "components.json", "components.lock.json")
|
|
self.git(self.repo, "commit", "-m", "Adopt source manifests")
|
|
self.git(self.repo, "push", "origin", "main")
|
|
|
|
def worktree(self, name="feature/old"):
|
|
path = self.base / "task"
|
|
self.git(self.repo, "worktree", "add", "-b", name, str(path), "origin/main")
|
|
return path
|
|
|
|
def workspace(self, root, *args):
|
|
return self.call(root, "python3", "scripts/workspace.py", "--root", str(root), *args)
|
|
|
|
def test_precheck_prepares_fixed_rules_outside_project_before_ready(self):
|
|
self.adopt()
|
|
result = self.task(self.repo, "precheck")
|
|
self.assertIn("[verdict] READY", result.stdout)
|
|
self.assertFalse((self.repo / ".workspace").exists())
|
|
self.assertIn("[rules]", result.stdout)
|
|
reading = next(line for line in result.stdout.splitlines() if line.startswith("[rules]"))
|
|
self.assertIn(self.rules_sha, reading)
|
|
self.assertTrue(list((self.base / "cache").rglob("core/baseline.md")))
|
|
self.assertTrue(list((self.base / "cache").rglob("workflows/delivery.md")))
|
|
|
|
def test_precheck_never_claims_ready_for_unavailable_fixed_rules(self):
|
|
self.adopt()
|
|
(self.rules / "core/baseline.md").unlink()
|
|
self.git(self.rules, "commit", "-am", "Rules missing baseline")
|
|
document = json.loads((self.repo / "components.lock.json").read_text())
|
|
document["components"]["Rules"]["commit"] = self.git(self.rules, "rev-parse", "HEAD").stdout.strip()
|
|
(self.repo / "components.lock.json").write_text(json.dumps(document))
|
|
self.git(self.repo, "commit", "-am", "Pin invalid rules")
|
|
self.git(self.repo, "push", "origin", "main")
|
|
result = self.task(self.repo, "precheck", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertNotIn("[verdict] READY", result.stdout)
|
|
self.assertFalse((self.repo / ".workspace").exists())
|
|
|
|
def test_start_prepares_fixed_sources_without_recursive_submodule_init(self):
|
|
nested, _ = self.component("Nested", {"value": "nested\n"})
|
|
self.git(self.app, "submodule", "add", str(nested), "deps/Nested")
|
|
self.git(self.app, "commit", "-am", "Pin nested source")
|
|
self.app_sha = self.git(self.app, "rev-parse", "HEAD").stdout.strip()
|
|
self.adopt()
|
|
receipt = self.base / "precheck.json"
|
|
self.call(self.repo, "bash", "scripts/evidence.sh", env=self.env | {"RECEIPT": str(receipt), "CMD": "make precheck"})
|
|
self.task(self.repo, "start", "feature/new", env=self.env | {"PRECHECK_RECEIPT": str(receipt)})
|
|
wt = self.base / "repo-wt-feature-new"
|
|
self.assertTrue((wt / ".workspace/App/.git").is_dir(), "fixed App checkout is missing")
|
|
self.assertEqual(self.git(wt / ".workspace/App", "rev-parse", "HEAD").stdout.strip(), self.app_sha)
|
|
self.assertEqual(self.git(wt / ".workspace/Rules", "rev-parse", "HEAD").stdout.strip(), self.rules_sha)
|
|
self.assertFalse((wt / ".workspace/App/deps/Nested/.git").exists())
|
|
self.assertTrue((wt / ".task/feature-new.md").exists())
|
|
|
|
def test_old_task_sync_adopts_new_sources_and_reports_explicit_transition(self):
|
|
self.git(self.repo, "-c", "core.excludesfile=/dev/null", "submodule", "add", "--force", str(self.app), "modules/App")
|
|
self.git(self.repo, "commit", "-m", "Pin old source")
|
|
self.git(self.repo, "push", "origin", "main")
|
|
wt = self.worktree()
|
|
self.git(wt, "submodule", "update", "--init")
|
|
legacy = wt / "modules/App"
|
|
self.git(legacy, "switch", "-c", "feature/owned")
|
|
before = self.git(legacy, "show-ref").stdout
|
|
self.git(self.repo, "rm", "--cached", "modules/App")
|
|
self.git(self.repo, "rm", ".gitmodules")
|
|
self.git(self.repo, "commit", "-m", "Retire gitlink without removing checkout")
|
|
self.adopt()
|
|
result = self.task(wt, "sync")
|
|
self.assertIn("[transition]", result.stdout)
|
|
self.assertEqual(self.git(legacy, "show-ref").stdout, before)
|
|
self.assertEqual(self.git(legacy, "branch", "--show-current").stdout.strip(), "feature/owned")
|
|
self.assertTrue((wt / ".workspace/App/.git").is_dir())
|
|
self.assertEqual((legacy / "content").read_text(), "app v1\n")
|
|
|
|
def test_sync_preserves_edit_branch_and_refuses_changed_declared_source(self):
|
|
self.adopt()
|
|
wt = self.worktree()
|
|
self.workspace(wt, "edit", "--name", "App", "--branch", "feature/owned")
|
|
edited = wt / ".workspace/App"
|
|
(edited / "content").write_text("local unsaved\n")
|
|
self.task(wt, "sync")
|
|
(self.app / "content").write_text("app v2\n")
|
|
self.git(self.app, "commit", "-am", "App v2")
|
|
lock = json.loads((self.repo / "components.lock.json").read_text())
|
|
lock["components"]["App"]["commit"] = self.git(self.app, "rev-parse", "HEAD").stdout.strip()
|
|
(self.repo / "components.lock.json").write_text(json.dumps(lock))
|
|
self.git(self.repo, "commit", "-am", "Update application lock")
|
|
self.git(self.repo, "push", "origin", "main")
|
|
result = self.task(wt, "sync", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertNotIn("[synced]", result.stdout)
|
|
self.assertEqual((edited / "content").read_text(), "local unsaved\n")
|
|
self.assertEqual(self.git(edited, "branch", "--show-current").stdout.strip(), "feature/owned")
|
|
|
|
def test_finish_preserves_external_worktree_of_retired_legacy_checkout(self):
|
|
self.adopt()
|
|
wt = self.worktree()
|
|
legacy = wt / "modules/App"
|
|
self.git(wt, "clone", str(self.app), str(legacy))
|
|
external = self.base / "external"
|
|
self.git(legacy, "worktree", "add", "-b", "feature/external", str(external))
|
|
(external / "content").write_text("external unique\n")
|
|
self.git(external, "add", "content")
|
|
before = self.git(external, "status", "--porcelain").stdout
|
|
result = self.task(wt, "finish", check=False, env=self.env | {"TASK_WRITERS_STOPPED": "1"})
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("registered component worktree", result.stdout)
|
|
self.assertEqual(self.git(external, "status", "--porcelain").stdout, before)
|
|
self.assertEqual((external / "content").read_text(), "external unique\n")
|
|
self.assertTrue(wt.exists())
|
|
|
|
def test_verify_detects_active_git_operation_in_nested_retired_root(self):
|
|
self.adopt()
|
|
wt = self.worktree()
|
|
legacy = wt / "modules/App"
|
|
self.git(wt, "clone", str(self.app), str(legacy))
|
|
nested = legacy / "retired/nested"
|
|
self.git(wt, "clone", str(self.rules), str(nested))
|
|
lock = nested / ".git/index.lock"
|
|
lock.write_text("another writer\n")
|
|
result = self.call(wt, "python3", "scripts/check-worktree.py", "verify", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("active Git operation/lock", result.stdout)
|
|
self.assertEqual(lock.read_text(), "another writer\n")
|
|
|
|
def test_make_workspace_commands_prepare_edit_unlink_and_report(self):
|
|
self.adopt()
|
|
shutil.copyfile(SOURCE / "Makefile", self.repo / "Makefile")
|
|
self.call(self.repo, "make", "workspace-prepare", "NAME=App")
|
|
source = self.repo / ".workspace/App"
|
|
self.assertEqual((source / "content").read_text(), "app v1\n")
|
|
self.call(self.repo, "make", "component-edit", "NAME=App", "BRANCH=feature/from-make")
|
|
self.assertEqual(self.git(source, "branch", "--show-current").stdout.strip(), "feature/from-make")
|
|
self.call(self.repo, "make", "component-unlink", "NAME=App")
|
|
self.assertTrue(source.is_dir())
|
|
status = self.call(self.repo, "make", "workspace-status", "NAME=App")
|
|
self.assertIn('"feature/from-make"', status.stdout)
|
|
transition = self.call(self.repo, "make", "workspace-transition", "NAME=App")
|
|
self.assertIn('"legacy"', transition.stdout)
|
|
self.assertFalse((self.repo / "modules/App").exists())
|
|
|
|
def test_finish_preserves_absorbed_metadata_after_legacy_checkout_disappears(self):
|
|
self.git(self.repo, "submodule", "add", "--force", str(self.app), "modules/App")
|
|
self.git(self.repo, "commit", "-m", "Old gitlink")
|
|
self.git(self.repo, "push", "origin", "main")
|
|
wt = self.worktree()
|
|
self.git(wt, "submodule", "update", "--init")
|
|
legacy = wt / "modules/App"
|
|
external = self.base / "external"
|
|
self.git(legacy, "worktree", "add", "-b", "feature/external", str(external))
|
|
(external / "content").write_text("independent owner\n")
|
|
self.git(external, "add", "content")
|
|
before = self.git(external, "status", "--porcelain").stdout
|
|
self.git(self.repo, "rm", "--cached", "modules/App")
|
|
self.git(self.repo, "rm", ".gitmodules")
|
|
self.git(self.repo, "commit", "-m", "Remove legacy index entries")
|
|
self.adopt()
|
|
self.task(wt, "sync")
|
|
shutil.rmtree(legacy) # Simulate an absent checkout while its Git metadata remains.
|
|
result = self.task(wt, "finish", check=False, env=self.env | {"TASK_WRITERS_STOPPED": "1"})
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("retained metadata", result.stdout)
|
|
self.assertEqual(self.git(external, "status", "--porcelain").stdout, before)
|
|
self.assertEqual((external / "content").read_text(), "independent owner\n")
|
|
|
|
def test_finish_preserves_nonbranch_legacy_refs_without_remote_backup(self):
|
|
self.adopt()
|
|
wt = self.worktree()
|
|
legacy = wt / "modules/App"
|
|
self.git(wt, "clone", str(self.app), str(legacy))
|
|
(legacy / "content").write_text("unpublished\n")
|
|
self.git(legacy, "commit", "-am", "Unique migration object")
|
|
unique = self.git(legacy, "rev-parse", "HEAD").stdout.strip()
|
|
self.git(legacy, "update-ref", "refs/migration/unique", unique)
|
|
self.git(legacy, "reset", "--hard", "origin/main")
|
|
self.git(legacy, "reflog", "expire", "--expire=now", "--all")
|
|
result = self.task(wt, "finish", check=False, env=self.env | {"TASK_WRITERS_STOPPED": "1"})
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("unpreserved local refs", result.stdout)
|
|
self.assertEqual(self.git(legacy, "rev-parse", "refs/migration/unique").stdout.strip(), unique)
|
|
|
|
@unittest.skipUnless(shutil.which("go"), "Go is required for architecture gate")
|
|
def test_architecture_gate_preserves_edges_under_workspace_paths_and_new_module_identity(self):
|
|
self.adopt()
|
|
policy = self.repo / "policy"
|
|
policy.mkdir(exist_ok=True)
|
|
(policy / "architecture.yaml").write_text("backend:\n root: .workspace/Agent-System-Backend/internal\n import_prefix: git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git/internal/\nfrontend:\n root: .workspace/Agent-System-Frontend/src\n groups: components\n")
|
|
(policy / "architecture-baseline.json").write_text('["backend/app backend/store"]')
|
|
backend = self.repo / ".workspace/Agent-System-Backend"
|
|
(backend / "internal/app").mkdir(parents=True)
|
|
(backend / "internal/store").mkdir()
|
|
(backend / "go.mod").write_text("module git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git\n\ngo 1.23\n")
|
|
source = backend / "internal/app/app.go"
|
|
source.write_text('package app\nimport _ "git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git/internal/store"\n')
|
|
(self.repo / ".workspace/Agent-System-Frontend/src").mkdir(parents=True)
|
|
self.call(self.repo, "bash", "scripts/lint-architecture.sh")
|
|
(backend / "internal/server").mkdir()
|
|
source.write_text(source.read_text() + 'import _ "git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git/internal/server"\n')
|
|
result = self.call(self.repo, "bash", "scripts/lint-architecture.sh", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("backend/app backend/server", result.stdout)
|
|
self.assertEqual(json.loads((policy / "architecture-baseline.json").read_text()), ["backend/app backend/store"])
|
|
|
|
@unittest.skipUnless(shutil.which("go"), "Go is required for boundary gates")
|
|
def test_boundary_gate_uses_isolated_map_and_rejects_new_backend_internal_import(self):
|
|
self.adopt()
|
|
names = ("Agent-System-Backend", "Agent-System-Frontend", "Agent-Intent", "Agent-Native", "Agent-Squad")
|
|
modules = {name: "git-ssh.u1680832.nyat.app/weichen/" + name + ".git" for name in names}
|
|
entries = [{"id": name, "repo": str(self.app), "path": ".workspace/" + name,
|
|
"role": "application" if name in names[:2] else "go-library", "modules": [], "target": "main"} for name in names]
|
|
(self.repo / "components.json").write_text(json.dumps({"schema": 1, "components": entries}))
|
|
(self.repo / "components.lock.json").write_text(json.dumps({"schema": 1, "components": {name: {"commit": self.app_sha} for name in names[:2]}}))
|
|
policy = self.repo / "policy"
|
|
policy.mkdir()
|
|
(policy / "architecture.yaml").write_text("backend: .workspace/Agent-System-Backend/internal\nfrontend: .workspace/Agent-System-Frontend/src\ncomponents: {}\n")
|
|
(policy / "architecture-baseline.json").write_text("[]")
|
|
(policy / "lib-deps-allowlist.txt").write_text("# No sibling-library imports\n")
|
|
mapping = {"schema": 1, "root_commit": self.git(self.repo, "rev-parse", "HEAD").stdout.strip(), "components": {}}
|
|
for name in names:
|
|
path = self.base / "isolated" / name
|
|
path.mkdir(parents=True)
|
|
if name != "Agent-System-Frontend":
|
|
(path / "go.mod").write_text("module " + modules[name] + "\n\ngo 1.23\n")
|
|
(path / "source.go").write_text("package fixture\n")
|
|
mapping["components"][name] = {"path": str(path), "commit": self.app_sha, "repo": str(self.app)}
|
|
backend = self.base / "isolated/Agent-System-Backend"
|
|
(backend / "internal/store").mkdir(parents=True)
|
|
(backend / "internal/store/store.go").write_text("package store\n")
|
|
sdk = backend / "pkg/extension"
|
|
sdk.mkdir(parents=True)
|
|
source = sdk / "sdk.go"
|
|
source.write_text("package extension\n")
|
|
(self.base / "isolated/Agent-System-Frontend/src").mkdir()
|
|
source_map = self.base / "source-map.json"
|
|
source_map.write_text(json.dumps(mapping))
|
|
env = self.env | {"COMPOSITION_SOURCE_MAP": str(source_map)}
|
|
self.call(self.repo, "bash", "scripts/lint-boundaries.sh", env=env)
|
|
source.write_text('package extension\nimport _ "' + modules["Agent-System-Backend"] + '/internal/store"\n')
|
|
result = self.call(self.repo, "bash", "scripts/lint-boundaries.sh", env=env, check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("boundary: pkg/extension imports", result.stdout)
|
|
# A forged source identity cannot substitute for the locked application.
|
|
source.write_text("package extension\n")
|
|
mapping["components"]["Agent-System-Backend"]["commit"] = "0" * 40
|
|
source_map.write_text(json.dumps(mapping))
|
|
result = self.call(self.repo, "bash", "scripts/lint-boundaries.sh", env=env, check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("locked source", result.stdout)
|
|
|
|
def test_verify_detects_nested_git_operation_inside_new_workspace(self):
|
|
self.adopt()
|
|
wt = self.worktree()
|
|
self.workspace(wt, "prepare")
|
|
nested = wt / ".workspace/App/private/Rules"
|
|
self.git(wt, "clone", str(self.rules), str(nested))
|
|
lock = nested / ".git/index.lock"
|
|
lock.write_text("active nested writer\n")
|
|
result = self.call(wt, "python3", "scripts/check-worktree.py", "verify", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("active Git operation/lock", result.stdout)
|
|
self.assertEqual(lock.read_text(), "active nested writer\n")
|
|
|
|
def test_old_task_sync_preserves_unique_legacy_commit_staging_dirty_and_ignored_files(self):
|
|
self.git(self.repo, "submodule", "add", "--force", str(self.app), "modules/App")
|
|
self.git(self.repo, "commit", "-m", "Old gitlink")
|
|
self.git(self.repo, "push", "origin", "main")
|
|
wt = self.worktree()
|
|
self.git(wt, "submodule", "update", "--init")
|
|
legacy = wt / "modules/App"
|
|
self.git(legacy, "switch", "-c", "feature/unique")
|
|
(legacy / "content").write_text("unpublished commit\n")
|
|
self.git(legacy, "commit", "-am", "Unique component history")
|
|
(legacy / "content").write_text("staged component change\n")
|
|
self.git(legacy, "add", "content")
|
|
(legacy / "content").write_text("unstaged component change\n")
|
|
(legacy / "private").mkdir()
|
|
(legacy / "private/unique").write_text("ignored unique data\n")
|
|
before = (self.git(legacy, "show-ref").stdout, self.git(legacy, "ls-files", "--stage").stdout,
|
|
self.git(wt, "rev-parse", "HEAD").stdout)
|
|
self.git(self.repo, "rm", "--cached", "modules/App")
|
|
self.git(self.repo, "rm", ".gitmodules")
|
|
self.git(self.repo, "commit", "-m", "Retire old gitlink")
|
|
self.adopt()
|
|
result = self.task(wt, "sync", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertNotIn("[synced]", result.stdout)
|
|
self.assertEqual((self.git(legacy, "show-ref").stdout, self.git(legacy, "ls-files", "--stage").stdout,
|
|
self.git(wt, "rev-parse", "HEAD").stdout), before)
|
|
self.assertEqual((legacy / "content").read_text(), "unstaged component change\n")
|
|
self.assertEqual((legacy / "private/unique").read_text(), "ignored unique data\n")
|
|
|
|
def test_old_task_text_conflict_keeps_legacy_source_and_reports_no_success(self):
|
|
self.git(self.repo, "submodule", "add", "--force", str(self.app), "modules/App")
|
|
self.git(self.repo, "commit", "-m", "Old gitlink")
|
|
self.git(self.repo, "push", "origin", "main")
|
|
wt = self.worktree()
|
|
self.git(wt, "submodule", "update", "--init")
|
|
(wt / "tracked").write_text("task branch choice\n")
|
|
self.git(wt, "commit", "-am", "Task text change")
|
|
legacy = wt / "modules/App"
|
|
before = self.git(legacy, "show-ref").stdout
|
|
(self.repo / "tracked").write_text("main branch choice\n")
|
|
self.git(self.repo, "commit", "-am", "Main text change")
|
|
self.git(self.repo, "rm", "--cached", "modules/App")
|
|
self.git(self.repo, "rm", ".gitmodules")
|
|
self.git(self.repo, "commit", "-m", "Retire legacy index")
|
|
self.adopt()
|
|
result = self.task(wt, "sync", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertNotIn("[synced]", result.stdout)
|
|
self.assertTrue(self.git(wt, "ls-files", "-u").stdout.strip())
|
|
self.assertEqual(self.git(legacy, "show-ref").stdout, before)
|
|
self.assertEqual((legacy / "content").read_text(), "app v1\n")
|
|
|
|
def test_precheck_rejects_borrowed_rules_cache_symlink(self):
|
|
self.adopt()
|
|
self.task(self.repo, "precheck")
|
|
source = next(path for path in (self.base / "cache/agent-system/rules").iterdir() if path.is_dir())
|
|
borrowed = self.base / "borrowed-rules"
|
|
source.rename(borrowed)
|
|
source.symlink_to(borrowed, target_is_directory=True)
|
|
before = self.git(borrowed, "show-ref").stdout
|
|
result = self.task(self.repo, "precheck", check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertNotIn("[verdict] READY", result.stdout)
|
|
self.assertEqual(self.git(borrowed, "show-ref").stdout, before)
|
|
self.assertEqual((borrowed / "core/baseline.md").read_text(), "# Baseline\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|