206 lines
10 KiB
Python
206 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from mais_humana.models import OrderType
|
|
from mais_humana.repository_mesh import MeshEnvironment, MeshEnvironmentKind, RepositoryTarget, build_mesh_report
|
|
from mais_humana.repository_mesh_reconciliation import (
|
|
ReconciliationStatus,
|
|
StatePrecedence,
|
|
apply_reconciliation_to_report,
|
|
build_reconciliation_plan,
|
|
candidate_states,
|
|
operations_csv,
|
|
reconciliation_csv,
|
|
reconciliation_markdown,
|
|
receipt_to_service_order,
|
|
select_latest_state,
|
|
service_orders_from_plan,
|
|
)
|
|
from tests.helpers import make_tmp
|
|
from tests.test_repository_mesh import FakeGit, make_repo
|
|
|
|
|
|
def one_env(root: Path) -> MeshEnvironment:
|
|
return MeshEnvironment("primary", MeshEnvironmentKind.WINDOWS_PRIMARY, str(root), "test")
|
|
|
|
|
|
def one_target(name: str = "alpha") -> RepositoryTarget:
|
|
return RepositoryTarget(name, f"admin/{name}", name, f"01_{name}")
|
|
|
|
|
|
class RepositoryMeshReconciliationTests(unittest.TestCase):
|
|
def test_single_hash_is_selected_as_latest_valid_state(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git", head="a" * 40)
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake)
|
|
latest = select_latest_state(report.summaries[0])
|
|
self.assertEqual(latest.precedence, StatePrecedence.SINGLE_HASH)
|
|
self.assertEqual(latest.selected_head, "a" * 40)
|
|
self.assertFalse(latest.blockers)
|
|
|
|
def test_dirty_tree_prevents_latest_state_selection(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git", status=(" M README.md",))
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake)
|
|
latest = select_latest_state(report.summaries[0])
|
|
self.assertEqual(latest.precedence, StatePrecedence.DIRTY_TREE)
|
|
self.assertIsNone(latest.selected_head)
|
|
self.assertTrue(latest.blockers)
|
|
|
|
def test_divergent_ahead_behind_requires_manual_precedence(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git", ahead_behind="2 2")
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake)
|
|
latest = select_latest_state(report.summaries[0])
|
|
self.assertEqual(latest.precedence, StatePrecedence.DIVERGENT)
|
|
self.assertIn("ahead=2 behind=2", " ".join(latest.blockers))
|
|
|
|
def test_local_ahead_is_candidate_but_still_needs_push_review(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git", ahead_behind="3 0")
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake)
|
|
latest = select_latest_state(report.summaries[0])
|
|
self.assertEqual(latest.precedence, StatePrecedence.SINGLE_HASH)
|
|
plan = build_reconciliation_plan(report)
|
|
self.assertEqual(plan.receipts[0].status, ReconciliationStatus.ALIGNED)
|
|
|
|
def test_missing_hash_creates_missing_materialization_receipt(self) -> None:
|
|
root = make_tmp()
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=FakeGit())
|
|
plan = build_reconciliation_plan(report)
|
|
receipt = plan.receipts[0]
|
|
self.assertEqual(receipt.status, ReconciliationStatus.MISSING_MATERIALIZATION)
|
|
self.assertFalse(receipt.safe_to_auto_sync)
|
|
self.assertTrue(receipt.pending_items)
|
|
|
|
def test_environment_blocked_receipt_has_blocker_operation(self) -> None:
|
|
root = make_tmp()
|
|
env = MeshEnvironment("server", MeshEnvironmentKind.CODEX_SERVER, str(root / "missing"), "remote", local=False)
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(env,), runner=FakeGit())
|
|
plan = build_reconciliation_plan(report)
|
|
receipt = plan.receipts[0]
|
|
self.assertEqual(receipt.status, ReconciliationStatus.ENVIRONMENT_BLOCKED)
|
|
self.assertTrue(receipt.blockers)
|
|
self.assertIn("ambiente declarado", receipt.blockers[0])
|
|
|
|
def test_credential_blocked_receipt_has_credential_operation(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(
|
|
repo,
|
|
remote="https://git.ami.app.br/admin/alpha.git",
|
|
fetch_error="fatal: SEC_E_NO_CREDENTIALS",
|
|
)
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake, fetch=True)
|
|
plan = build_reconciliation_plan(report)
|
|
receipt = plan.receipts[0]
|
|
self.assertEqual(receipt.status, ReconciliationStatus.CREDENTIAL_BLOCKED)
|
|
self.assertTrue(any("credencial" in item.lower() for item in receipt.pending_items))
|
|
|
|
def test_nominal_rename_ready_receipt_is_created_for_clean_alias(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha-old")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git")
|
|
target = RepositoryTarget(
|
|
"alpha-new",
|
|
"admin/alpha",
|
|
"alpha-new",
|
|
"01_alpha",
|
|
aliases=("alpha-old",),
|
|
canonical_name="alpha-new",
|
|
requires_nominal_reconciliation=True,
|
|
)
|
|
report = build_mesh_report(root, targets=(target,), environments=(one_env(root),), runner=fake)
|
|
plan = build_reconciliation_plan(report)
|
|
receipt = plan.receipts[0]
|
|
self.assertEqual(receipt.status, ReconciliationStatus.NOMINAL_RENAME_READY)
|
|
self.assertTrue(any("Rename-Item" in " ".join(operation.commands) for operation in receipt.operations))
|
|
|
|
def test_remote_mismatch_receipt_prioritizes_remote_operation(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/wrong.git")
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake)
|
|
plan = build_reconciliation_plan(report)
|
|
receipt = plan.receipts[0]
|
|
self.assertEqual(receipt.status, ReconciliationStatus.REMOTE_MISMATCH)
|
|
self.assertTrue(any("remote origin" in operation.title.lower() for operation in receipt.operations))
|
|
|
|
def test_candidate_states_extract_clean_commit_details(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git", head="b" * 40, ahead_behind="1 0")
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake)
|
|
candidates = candidate_states(report.summaries[0])
|
|
self.assertEqual(len(candidates), 1)
|
|
self.assertEqual(candidates[0].short_head, "b" * 12)
|
|
self.assertTrue(candidates[0].has_local_only_commits)
|
|
|
|
def test_reconciliation_markdown_and_csv_are_rendered(self) -> None:
|
|
root = make_tmp()
|
|
repo = make_repo(root, "alpha")
|
|
fake = FakeGit()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git")
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=fake)
|
|
plan = build_reconciliation_plan(report)
|
|
md = reconciliation_markdown(plan)
|
|
receipts_csv = reconciliation_csv(plan)
|
|
ops_csv = operations_csv(plan)
|
|
self.assertIn("Repository Mesh Reconciliation Plan", md)
|
|
self.assertIn("receipt_id,target_name", receipts_csv)
|
|
self.assertIn("operation_id,target_name", ops_csv)
|
|
|
|
def test_service_orders_from_plan_builds_executive_and_managerial_orders(self) -> None:
|
|
root = make_tmp()
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=FakeGit())
|
|
plan = build_reconciliation_plan(report)
|
|
orders = service_orders_from_plan(plan, start_executive=200, start_managerial=300)
|
|
self.assertEqual(len(orders), 2)
|
|
self.assertEqual(orders[0].order_type, OrderType.EXECUTIVE)
|
|
self.assertEqual(orders[1].order_type, OrderType.MANAGERIAL)
|
|
self.assertIn("repo-mesh", " ".join(orders[0].validations))
|
|
|
|
def test_receipt_to_service_order_preserves_paths_and_ready_criteria(self) -> None:
|
|
root = make_tmp()
|
|
report = build_mesh_report(root, targets=(one_target(),), environments=(one_env(root),), runner=FakeGit())
|
|
plan = build_reconciliation_plan(report)
|
|
order = receipt_to_service_order(plan.receipts[0], 222, OrderType.EXECUTIVE)
|
|
self.assertIn("G:/_codex-git/alpha", order.affected_paths)
|
|
self.assertTrue(order.ready_criteria)
|
|
self.assertEqual(order.status.value, "planejada")
|
|
|
|
def test_apply_reconciliation_writes_project_and_central_artifacts(self) -> None:
|
|
tmp = make_tmp()
|
|
ecosystem = tmp / "eco"
|
|
project = tmp / "human"
|
|
central = tmp / "central" / "projects" / "15_repo_tudo-para-ia-mais-humana-platform"
|
|
ecosystem.mkdir()
|
|
project.mkdir()
|
|
central.mkdir(parents=True)
|
|
report = build_mesh_report(ecosystem, targets=(one_target(),), environments=(one_env(ecosystem),), runner=FakeGit())
|
|
plan, records = apply_reconciliation_to_report(report, project, central_platform_folder=central)
|
|
self.assertTrue((project / "dados" / "repository-mesh-reconciliation.json").exists())
|
|
self.assertTrue((project / "matrizes" / "repository-mesh-operations.csv").exists())
|
|
self.assertTrue((project / "ecossistema" / "REPOSITORY-MESH-RECONCILIATION.md").exists())
|
|
self.assertTrue((central / "reports" / "EXECUTADO__repository-mesh-reconciliation.md").exists())
|
|
self.assertGreaterEqual(len(records), 7)
|
|
self.assertEqual(plan.report_id, report.report_id)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|