157 lines
6.5 KiB
Python
157 lines
6.5 KiB
Python
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()
|