85 lines
4.6 KiB
Python
85 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from mais_humana.repository_mesh import MeshEnvironment, MeshEnvironmentKind, RepositoryTarget, build_mesh_report
|
|
from mais_humana.repository_mesh_reconciliation import build_reconciliation_plan
|
|
from mais_humana.repository_mesh_runtime import acquire_lock, build_runtime_cycle, cron_scheduler_spec, release_lock, windows_scheduler_spec
|
|
from mais_humana.repository_mesh_semantic import write_repository_mesh_semantic_state
|
|
from mais_humana.repository_mesh_readiness import (
|
|
MeshGateStatus,
|
|
build_mesh_readiness_report,
|
|
readiness_csv,
|
|
readiness_markdown,
|
|
readiness_pending_items,
|
|
write_readiness_artifacts,
|
|
)
|
|
from tests.helpers import make_tmp
|
|
from tests.test_repository_mesh import FakeGit, make_repo
|
|
|
|
|
|
class RepositoryMeshReadinessTests(unittest.TestCase):
|
|
def build_bundle(self, *, fetch: bool = False, dirty: bool = False):
|
|
tmp = make_tmp()
|
|
eco = tmp / "eco"
|
|
project = tmp / "project"
|
|
central = tmp / "central"
|
|
eco.mkdir()
|
|
project.mkdir()
|
|
central.mkdir()
|
|
repo = make_repo(eco, "alpha")
|
|
fake = FakeGit()
|
|
status = (" M README.md",) if dirty else ()
|
|
fake.set_repo(repo, remote="https://git.ami.app.br/admin/alpha.git", status=status)
|
|
target = RepositoryTarget("alpha", "admin/alpha", "alpha", "01_alpha")
|
|
env = MeshEnvironment("primary", MeshEnvironmentKind.WINDOWS_PRIMARY, str(eco), "test")
|
|
report = build_mesh_report(eco, targets=(target,), environments=(env,), runner=fake, fetch=fetch)
|
|
plan = build_reconciliation_plan(report)
|
|
lock = acquire_lock(project / "dados" / "lock.json", owner="readiness-test")
|
|
cycle = build_runtime_cycle(report, plan, lock=lock)
|
|
specs = (
|
|
windows_scheduler_spec(python_exe="python.exe", project_root=project, ecosystem_root=eco, central_platform_folder=central),
|
|
cron_scheduler_spec(python_exe="python", project_root=project, ecosystem_root=eco, central_platform_folder=central),
|
|
)
|
|
counts = write_repository_mesh_semantic_state(central / "controle-semantico.sqlite", report=report, plan=plan, cycle=cycle, schedulers=specs)
|
|
return tmp, project, central, report, plan, cycle, specs, counts, lock
|
|
|
|
def test_readiness_report_warns_when_fetch_not_attempted(self) -> None:
|
|
tmp, project, central, report, plan, cycle, specs, counts, lock = self.build_bundle(fetch=False)
|
|
readiness = build_mesh_readiness_report(report, plan, cycle, specs, counts)
|
|
self.assertIn(readiness.status, {MeshGateStatus.WARN, MeshGateStatus.FAIL})
|
|
self.assertTrue(any(gate.gate_id == "inventory.fetch-attempted" and gate.status == MeshGateStatus.WARN for gate in readiness.gates))
|
|
release_lock(lock)
|
|
|
|
def test_readiness_report_classifies_dirty_tree_as_blocked(self) -> None:
|
|
tmp, project, central, report, plan, cycle, specs, counts, lock = self.build_bundle(fetch=True, dirty=True)
|
|
readiness = build_mesh_readiness_report(report, plan, cycle, specs, counts)
|
|
self.assertEqual(readiness.status, MeshGateStatus.FAIL)
|
|
self.assertTrue(any(gate.gate_id == "safety.dirty-blocked" for gate in readiness.gates))
|
|
self.assertTrue(readiness_pending_items(readiness))
|
|
release_lock(lock)
|
|
|
|
def test_readiness_csv_and_markdown_include_gate_details(self) -> None:
|
|
tmp, project, central, report, plan, cycle, specs, counts, lock = self.build_bundle(fetch=True)
|
|
readiness = build_mesh_readiness_report(report, plan, cycle, specs, counts)
|
|
self.assertIn("gate_id,status,severity", readiness_csv(readiness))
|
|
md = readiness_markdown(readiness)
|
|
self.assertIn("Repository Mesh Readiness", md)
|
|
self.assertIn("inventory.targets-declared", md)
|
|
release_lock(lock)
|
|
|
|
def test_write_readiness_artifacts_creates_project_and_central_records(self) -> None:
|
|
tmp, project, central, report, plan, cycle, specs, counts, lock = self.build_bundle(fetch=True)
|
|
readiness = build_mesh_readiness_report(report, plan, cycle, specs, counts)
|
|
records = write_readiness_artifacts(readiness, project, central_platform_folder=central)
|
|
self.assertTrue((project / "dados" / "repository-mesh-readiness.json").exists())
|
|
self.assertTrue((project / "matrizes" / "repository-mesh-readiness.csv").exists())
|
|
self.assertTrue((project / "ecossistema" / "REPOSITORY-MESH-READINESS.md").exists())
|
|
self.assertTrue((central / "reports" / "EXECUTADO__repository-mesh-readiness.md").exists())
|
|
self.assertGreaterEqual(len(records), 5)
|
|
release_lock(lock)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|