auto-sync: tudo-para-ia-mais-humana 2026-05-02 00:10:03
This commit is contained in:
156
tests/test_targeted_sync_audit.py
Normal file
156
tests/test_targeted_sync_audit.py
Normal file
@@ -0,0 +1,156 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from typing import Sequence
|
||||
|
||||
from mais_humana.cli import main
|
||||
from mais_humana.targeted_sync_audit import (
|
||||
GitCommandResult,
|
||||
SyncAuditStatus,
|
||||
SyncAuditTarget,
|
||||
TargetedSyncAuditReport,
|
||||
build_targeted_sync_audit,
|
||||
classify_observation,
|
||||
observe_repo,
|
||||
run_targeted_sync_audit,
|
||||
sync_audit_csv,
|
||||
sync_audit_markdown,
|
||||
)
|
||||
from tests.helpers import make_tmp
|
||||
|
||||
|
||||
class FakeGitRunner:
|
||||
def __init__(self) -> None:
|
||||
self.results: dict[tuple[str, str], GitCommandResult] = {}
|
||||
|
||||
def set_result(self, repo: Path, args: Sequence[str], output: str, exit_code: int = 0) -> None:
|
||||
self.results[(str(repo), " ".join(args))] = GitCommandResult(tuple(args), exit_code, output)
|
||||
|
||||
def run(self, repo_path: Path, args: Sequence[str]) -> GitCommandResult:
|
||||
return self.results.get((str(repo_path), " ".join(args)), GitCommandResult(tuple(args), 0, ""))
|
||||
|
||||
|
||||
def make_git_repo(root: Path, name: str) -> Path:
|
||||
repo = root / name
|
||||
(repo / ".git").mkdir(parents=True)
|
||||
return repo
|
||||
|
||||
|
||||
def target(path: Path, target_id: str = "repo") -> SyncAuditTarget:
|
||||
return SyncAuditTarget(target_id, str(path), "https://git.ami.app.br/admin/repo.git", "unit test repo")
|
||||
|
||||
|
||||
def configure_clean_runner(runner: FakeGitRunner, repo: Path, *, ahead_behind: str = "0 0") -> None:
|
||||
runner.set_result(repo, ("branch", "--show-current"), "main\n")
|
||||
runner.set_result(repo, ("rev-parse", "HEAD"), "a" * 40 + "\n")
|
||||
runner.set_result(repo, ("remote", "get-url", "origin"), "https://git.ami.app.br/admin/repo.git\n")
|
||||
runner.set_result(repo, ("status", "--short", "--branch"), "## main...origin/main\n")
|
||||
runner.set_result(repo, ("status", "--porcelain", "--untracked-files=all"), "")
|
||||
runner.set_result(repo, ("rev-list", "--left-right", "--count", "origin/main...main"), ahead_behind)
|
||||
|
||||
|
||||
class TargetedSyncAuditTests(unittest.TestCase):
|
||||
def test_classify_missing_and_not_git(self) -> None:
|
||||
status, blockers, decision = classify_observation(
|
||||
exists=False,
|
||||
is_git=False,
|
||||
dirty=False,
|
||||
ahead=None,
|
||||
behind=None,
|
||||
fetch_output="",
|
||||
command_outputs=(),
|
||||
)
|
||||
self.assertEqual(status, SyncAuditStatus.MISSING)
|
||||
self.assertIn("path_missing", blockers)
|
||||
self.assertIn("materializar", decision)
|
||||
|
||||
def test_observe_clean_repo_aligned(self) -> None:
|
||||
root = make_tmp()
|
||||
repo = make_git_repo(root, "repo")
|
||||
runner = FakeGitRunner()
|
||||
configure_clean_runner(runner, repo)
|
||||
observation = observe_repo(target(repo), runner=runner)
|
||||
self.assertEqual(observation.status, SyncAuditStatus.ALIGNED)
|
||||
self.assertEqual(observation.branch, "main")
|
||||
self.assertEqual(observation.short_head, "a" * 12)
|
||||
self.assertTrue(observation.clean_for_auto_sync)
|
||||
|
||||
def test_fetch_credential_error_blocks_sync(self) -> None:
|
||||
root = make_tmp()
|
||||
repo = make_git_repo(root, "repo")
|
||||
runner = FakeGitRunner()
|
||||
configure_clean_runner(runner, repo)
|
||||
runner.set_result(repo, ("fetch", "--all", "--prune"), "fatal: schannel SEC_E_NO_CREDENTIALS", 128)
|
||||
observation = observe_repo(target(repo), runner=runner, fetch=True)
|
||||
self.assertEqual(observation.status, SyncAuditStatus.CREDENTIAL_BLOCKED)
|
||||
self.assertIn("git_credentials_unavailable", observation.blockers)
|
||||
|
||||
def test_dirty_and_remote_ahead_is_diverged(self) -> None:
|
||||
root = make_tmp()
|
||||
repo = make_git_repo(root, "repo")
|
||||
runner = FakeGitRunner()
|
||||
configure_clean_runner(runner, repo, ahead_behind="2 0")
|
||||
runner.set_result(repo, ("status", "--porcelain", "--untracked-files=all"), " M README.md\n")
|
||||
observation = observe_repo(target(repo), runner=runner)
|
||||
self.assertEqual(observation.status, SyncAuditStatus.DIVERGED)
|
||||
self.assertIn("dirty_worktree_remote_ahead", observation.blockers)
|
||||
|
||||
def test_build_report_and_render_outputs(self) -> None:
|
||||
root = make_tmp()
|
||||
repo = make_git_repo(root, "repo")
|
||||
runner = FakeGitRunner()
|
||||
configure_clean_runner(runner, repo)
|
||||
report = build_targeted_sync_audit(targets=(target(repo),), runner=runner)
|
||||
self.assertIsInstance(report, TargetedSyncAuditReport)
|
||||
self.assertEqual(report.status, SyncAuditStatus.ALIGNED)
|
||||
self.assertIn("Targeted Git Sync Audit", sync_audit_markdown(report))
|
||||
self.assertIn("target_id,path,branch", sync_audit_csv(report))
|
||||
|
||||
def test_run_targeted_sync_audit_writes_project_and_central_artifacts(self) -> None:
|
||||
root = make_tmp()
|
||||
project = make_git_repo(root, "tudo-para-ia-mais-humana")
|
||||
mcp = make_git_repo(root, "tudo-para-ia-mcps-internos-plataform")
|
||||
central_repo = make_git_repo(root, "nucleo-gestao-operacional")
|
||||
central = root / "central" / "15_repo_tudo-para-ia-mais-humana-platform"
|
||||
runner = FakeGitRunner()
|
||||
for repo in (project, mcp, central_repo):
|
||||
configure_clean_runner(runner, repo)
|
||||
report, records = run_targeted_sync_audit(
|
||||
project_root=project,
|
||||
mcp_repo_root=mcp,
|
||||
central_repo_root=central_repo,
|
||||
central_platform_folder=central,
|
||||
runner=runner,
|
||||
)
|
||||
self.assertEqual(report.status, SyncAuditStatus.ALIGNED)
|
||||
self.assertTrue((project / "dados" / "targeted-sync-audit.json").exists())
|
||||
self.assertTrue((project / "matrizes" / "targeted-sync-audit.csv").exists())
|
||||
self.assertTrue((project / "ecossistema" / "TARGETED-SYNC-AUDIT.md").exists())
|
||||
self.assertTrue((central / "reports" / "EXECUTADO__targeted-sync-audit.md").exists())
|
||||
self.assertGreaterEqual(len(records), 4)
|
||||
|
||||
def test_cli_targeted_sync_audit_writes_payload(self) -> None:
|
||||
root = make_tmp()
|
||||
project = make_git_repo(root, "project")
|
||||
mcp = make_git_repo(root, "mcp")
|
||||
central = make_git_repo(root, "central")
|
||||
code = main(
|
||||
[
|
||||
"targeted-sync-audit",
|
||||
"--project-root",
|
||||
str(project),
|
||||
"--mcp-repo-root",
|
||||
str(mcp),
|
||||
"--central-repo-root",
|
||||
str(central),
|
||||
]
|
||||
)
|
||||
self.assertEqual(code, 0)
|
||||
payload = json.loads((project / "dados" / "targeted-sync-audit.json").read_text(encoding="utf-8"))
|
||||
self.assertEqual(len(payload["observations"]), 3)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
84
tests/test_workspace_hygiene.py
Normal file
84
tests/test_workspace_hygiene.py
Normal file
@@ -0,0 +1,84 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from mais_humana.cli import main
|
||||
from mais_humana.workspace_hygiene import (
|
||||
HygieneStatus,
|
||||
HygieneTarget,
|
||||
build_workspace_hygiene_report,
|
||||
hygiene_csv,
|
||||
hygiene_markdown,
|
||||
run_workspace_hygiene,
|
||||
safe_child_path,
|
||||
)
|
||||
from tests.helpers import make_tmp
|
||||
|
||||
|
||||
class WorkspaceHygieneTests(unittest.TestCase):
|
||||
def make_project_with_artifacts(self) -> Path:
|
||||
project = make_tmp()
|
||||
scratch = project / ".test-tmp" / "case"
|
||||
deps = project / "node_modules" / "pkg"
|
||||
scratch.mkdir(parents=True)
|
||||
deps.mkdir(parents=True)
|
||||
(scratch / "evidence.tmp").write_text("temporary evidence", encoding="utf-8")
|
||||
(deps / "package.json").write_text("{}", encoding="utf-8")
|
||||
return project
|
||||
|
||||
def test_safe_child_path_rejects_escape(self) -> None:
|
||||
project = make_tmp()
|
||||
with self.assertRaises(ValueError):
|
||||
safe_child_path(project, "..\\outside")
|
||||
|
||||
def test_dry_run_reports_present_artifacts_without_deleting(self) -> None:
|
||||
project = self.make_project_with_artifacts()
|
||||
report = build_workspace_hygiene_report(project, apply=False)
|
||||
self.assertEqual(report.status, HygieneStatus.PARTIAL)
|
||||
self.assertFalse(report.clean)
|
||||
self.assertTrue((project / ".test-tmp").exists())
|
||||
self.assertTrue((project / "node_modules").exists())
|
||||
self.assertIn("dry run", hygiene_markdown(report))
|
||||
self.assertIn("target_id,path,action,status", hygiene_csv(report))
|
||||
|
||||
def test_apply_removes_approved_artifacts(self) -> None:
|
||||
project = self.make_project_with_artifacts()
|
||||
report = build_workspace_hygiene_report(project, apply=True)
|
||||
self.assertEqual(report.status, HygieneStatus.PASSED)
|
||||
self.assertTrue(report.clean)
|
||||
self.assertFalse((project / ".test-tmp").exists())
|
||||
self.assertFalse((project / "node_modules").exists())
|
||||
self.assertTrue(all(action.deleted for action in report.actions))
|
||||
|
||||
def test_custom_unsafe_target_is_blocked_before_delete(self) -> None:
|
||||
project = make_tmp()
|
||||
target = HygieneTarget("unsafe", "..\\outside", "unit test")
|
||||
report = build_workspace_hygiene_report(project, apply=True, targets=(target,))
|
||||
self.assertEqual(report.status, HygieneStatus.BLOCKED)
|
||||
self.assertTrue(report.blockers)
|
||||
self.assertIn("escapes project root", report.actions[0].error)
|
||||
|
||||
def test_run_workspace_hygiene_writes_project_and_central_artifacts(self) -> None:
|
||||
project = self.make_project_with_artifacts()
|
||||
central = make_tmp() / "central" / "15_repo_tudo-para-ia-mais-humana-platform"
|
||||
report, records = run_workspace_hygiene(project_root=project, central_platform_folder=central, apply=True)
|
||||
self.assertEqual(report.status, HygieneStatus.PASSED)
|
||||
self.assertTrue((project / "dados" / "workspace-hygiene-report.json").exists())
|
||||
self.assertTrue((project / "matrizes" / "workspace-hygiene-targets.csv").exists())
|
||||
self.assertTrue((project / "ecossistema" / "WORKSPACE-HYGIENE-REPORT.md").exists())
|
||||
self.assertTrue((central / "reports" / "EXECUTADO__workspace-hygiene.md").exists())
|
||||
self.assertGreaterEqual(len(records), 4)
|
||||
|
||||
def test_cli_workspace_hygiene_writes_json_payload(self) -> None:
|
||||
project = self.make_project_with_artifacts()
|
||||
code = main(["workspace-hygiene", "--project-root", str(project), "--apply"])
|
||||
self.assertEqual(code, 0)
|
||||
payload = json.loads((project / "dados" / "workspace-hygiene-report.json").read_text(encoding="utf-8"))
|
||||
self.assertEqual(payload["status"], "passed")
|
||||
self.assertEqual(payload["clean"], True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user