"""Helper script for multi-project subplan Robot Framework tests. Usage: python helper_multi_project_subplan.py init-scopes python helper_multi_project_subplan.py resolve-alias python helper_multi_project_subplan.py validate-mixed python helper_multi_project_subplan.py record-changeset python helper_multi_project_subplan.py validate-cross python helper_multi_project_subplan.py cli-dict """ from __future__ import annotations import sys from pathlib import Path from unittest.mock import MagicMock # Ensure source tree is importable _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) from cleveragents.application.services.multi_project_service import ( # noqa: E402 MultiProjectService, ) from cleveragents.domain.models.core.multi_project import ( # noqa: E402 ChangeSetSummary, CrossProjectDependency, ProjectScopeResolver, ) from cleveragents.domain.models.core.plan import ( # noqa: E402 NamespacedName, Plan, PlanIdentity, PlanPhase, ProcessingState, ProjectLink, ) _PLAN_ID = "01HXAAAAAAAAAAAAAAAAAAAAAA" def _make_plan(links: list[ProjectLink]) -> Plan: """Build a minimal Plan.""" return Plan( identity=PlanIdentity(plan_id=_PLAN_ID), namespaced_name=NamespacedName.parse("local/robot-mp-test"), description="Robot multi-project test", action_name="local/test-action", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, project_links=links, ) def _mock_decision_service() -> MagicMock: mock = MagicMock() mock.list_by_type.return_value = [] return mock def cmd_init_scopes() -> None: """Initialize scopes and verify project count.""" svc = MultiProjectService(decision_service=_mock_decision_service()) plan = _make_plan( [ ProjectLink(project_name="api-service"), ProjectLink(project_name="shared-lib"), ] ) resources = { "api-service": ["res-api-1", "res-api-2"], "shared-lib": ["res-lib-1"], } updated = svc.initialize_scopes(plan, resources) mp = updated.multi_project_metadata assert mp is not None assert len(mp.project_scopes) == 2 api_scope = updated.get_project_scope("api-service") assert api_scope is not None assert len(api_scope.resource_ids) == 2 print("multi-project-init-ok") print(f"scope_count={len(mp.project_scopes)}") def cmd_resolve_alias() -> None: """Resolve a project alias.""" resolver = ProjectScopeResolver() links = [ ProjectLink(project_name="api-service", alias="api"), ProjectLink(project_name="shared-lib"), ] link = resolver.resolve_alias(links, "api") assert link is not None assert link.project_name == "api-service" link2 = resolver.resolve_alias(links, "shared-lib") assert link2 is not None assert link2.project_name == "shared-lib" link3 = resolver.resolve_alias(links, "nonexistent") assert link3 is None print("multi-project-alias-ok") def cmd_validate_mixed() -> None: """Validate mixed access mode detection.""" resolver = ProjectScopeResolver() links = [ ProjectLink(project_name="api-service", read_only=False), ProjectLink(project_name="shared-lib", read_only=True), ] errors = resolver.validate_access_modes(links, allow_mixed=False) assert len(errors) == 1, f"Expected 1 error, got {len(errors)}" errors2 = resolver.validate_access_modes(links, allow_mixed=True) assert len(errors2) == 0, f"Expected 0 errors, got {len(errors2)}" print("multi-project-mixed-ok") def cmd_record_changeset() -> None: """Record a changeset and verify it.""" svc = MultiProjectService(decision_service=_mock_decision_service()) plan = _make_plan( [ ProjectLink(project_name="api-service"), ProjectLink(project_name="shared-lib"), ] ) plan = svc.initialize_scopes( plan, { "api-service": ["res-1"], "shared-lib": ["res-2"], }, ) summary = ChangeSetSummary( project_name="api-service", files_changed=5, files_added=2, total_lines_changed=120, ) plan = svc.record_changeset(plan, "api-service", summary) scope = plan.get_project_scope("api-service") assert scope is not None assert scope.changeset_summary is not None assert scope.changeset_summary.files_changed == 5 print("multi-project-changeset-ok") def cmd_validate_cross() -> None: """Validate cross-project dependency constraints.""" svc = MultiProjectService(decision_service=_mock_decision_service()) plan = _make_plan( [ ProjectLink(project_name="api-service"), ProjectLink(project_name="shared-lib"), ] ) plan = svc.initialize_scopes( plan, { "api-service": ["res-1"], "shared-lib": ["res-2"], }, ) # Valid dependency dep = CrossProjectDependency( source_project="api-service", target_project="shared-lib", dependency_type="imports", ) mp = plan.multi_project_metadata assert mp is not None updated_mp = mp.model_copy(update={"cross_project_dependencies": [dep]}) plan = plan.model_copy(update={"multi_project_metadata": updated_mp}) errors = svc.validate_cross_project(plan) assert len(errors) == 0, f"Expected 0 errors, got: {errors}" # Invalid dependency (unknown target) bad_dep = CrossProjectDependency( source_project="api-service", target_project="unknown", dependency_type="imports", ) updated_mp2 = mp.model_copy(update={"cross_project_dependencies": [bad_dep]}) plan2 = plan.model_copy(update={"multi_project_metadata": updated_mp2}) errors2 = svc.validate_cross_project(plan2) assert len(errors2) > 0, "Expected error for unknown project" print("multi-project-cross-ok") def cmd_cli_dict() -> None: """Verify as_cli_dict includes multi-project section.""" svc = MultiProjectService(decision_service=_mock_decision_service()) plan = _make_plan( [ ProjectLink(project_name="api-service"), ProjectLink(project_name="shared-lib"), ] ) plan = svc.initialize_scopes( plan, { "api-service": ["res-1"], "shared-lib": ["res-2"], }, ) cli_dict = plan.as_cli_dict() assert "multi_project" in cli_dict mp_section = cli_dict["multi_project"] assert mp_section["total_projects"] == 2 assert len(mp_section["project_scopes"]) == 2 print("multi-project-cli-dict-ok") def main() -> None: """Dispatch subcommand.""" if len(sys.argv) < 2: print("Usage: helper_multi_project_subplan.py ") sys.exit(1) cmd = sys.argv[1] dispatch = { "init-scopes": cmd_init_scopes, "resolve-alias": cmd_resolve_alias, "validate-mixed": cmd_validate_mixed, "record-changeset": cmd_record_changeset, "validate-cross": cmd_validate_cross, "cli-dict": cmd_cli_dict, } handler = dispatch.get(cmd) if handler is None: print(f"Unknown command: {cmd}") sys.exit(1) handler() if __name__ == "__main__": main()