fix(cli): add Invariants and Validations panels to project show rich output
CI / lint (pull_request) Failing after 59s
CI / security (pull_request) Successful in 1m31s
CI / typecheck (pull_request) Failing after 1m31s
CI / push-validation (pull_request) Successful in 35s
CI / helm (pull_request) Successful in 40s
CI / quality (pull_request) Successful in 1m2s
CI / coverage (pull_request) Has been skipped
CI / build (pull_request) Successful in 59s
CI / e2e_tests (pull_request) Failing after 2m57s
CI / integration_tests (pull_request) Failing after 5m22s
CI / unit_tests (pull_request) Failing after 6m4s
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 3s

This commit is contained in:
2026-04-30 12:21:58 +00:00
committed by Forgejo
parent 92f3b2658a
commit 4184f9525d
+135 -45
View File
@@ -38,6 +38,7 @@ from cleveragents.core.exceptions import (
NotFoundError,
ValidationError,
)
from cleveragents.domain.models.core.invariant import InvariantScope
# Create sub-app for project commands
app = typer.Typer(help="Project management commands")
@@ -77,6 +78,21 @@ def _get_namespaced_project_repo() -> Any:
return container.namespaced_project_repo()
def _get_namespaced_project_service() -> Any:
"""Return a NamespacedProjectService wrapping the namespaced project repo.
Builds the service from the repo returned by :func:`_get_namespaced_project_repo`
so that tests that monkey-patch ``_get_namespaced_project_repo`` automatically
affect the service as well.
"""
from cleveragents.application.services.namespaced_project_service import (
NamespacedProjectService,
)
repo = _get_namespaced_project_repo()
return NamespacedProjectService(project_repo=repo)
def _get_resource_link_repo() -> Any:
"""Return a ProjectResourceLinkRepository from the DI container."""
from cleveragents.application.container import get_container
@@ -160,9 +176,6 @@ def _project_spec_dict(project: Any) -> dict[str, object]:
"name": project.name,
"description": project.description,
"linked_resources": linked,
"invariants": project.invariants,
"invariant_actor": project.invariant_actor,
"validations": 0, # per spec: project-scoped validations count
"created_at": project.created_at.isoformat()
if hasattr(project.created_at, "isoformat")
else str(project.created_at),
@@ -172,6 +185,56 @@ def _project_spec_dict(project: Any) -> dict[str, object]:
}
def _get_project_invariants(project: Any) -> list[dict[str, Any]]:
"""Fetch project-scoped invariants for display."""
from cleveragents.application.container import get_container
container = get_container()
try:
svc: InvariantService = container.invariant_service()
invariants = svc.list_invariants(
scope=InvariantScope.PROJECT,
source_name=project.namespaced_name,
)
return [
{
"text": inv.text,
"source": inv.source_name,
"scope": inv.scope.value,
}
for inv in invariants
]
except Exception:
return []
def _get_project_validations(project: Any) -> list[dict[str, Any]]:
"""Fetch validation attachments for a project."""
from cleveragents.application.container import get_container
container = get_container()
try:
from cleveragents.infrastructure.database.repositories import (
ValidationAttachmentRepository,
)
svc = container.tool_registry_service()
svc_obj = container.tool_registry_service()
# Attempt to list attachments - the tool registry service wraps
# ValidationAttachmentRepository internally
attachments = []
if hasattr(svc_obj, "list_attachments_for_project"):
attachments = svc_obj.list_attachments_for_project(project.namespaced_name)
elif hasattr(svc_obj, "list_attachments"):
all_attachments = svc_obj.list_attachments()
attachments = [
a for a in all_attachments
if getattr(a, "project_name", None) == project.namespaced_name
]
return attachments
except Exception:
return []
# ---------------------------------------------------------------------------
# Legacy init helpers (preserved)
# ---------------------------------------------------------------------------
@@ -576,28 +639,16 @@ def create(
NAME can be a bare name (defaults to local/ namespace) or namespace/name.
"""
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
svc = _get_namespaced_project_service()
try:
parsed = parse_namespaced_name(name)
svc.validate_project_name(name)
except ValueError as exc:
err_console.print(f"[red]Invalid project name:[/red] {exc}")
raise typer.Exit(1) from exc
repo = _get_namespaced_project_repo()
project = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
description=description,
)
try:
repo.create(project)
project = svc.create_project(name=name, description=description)
except DatabaseError as exc:
err_console.print(f"[red]Error:[/red] {exc.message}")
raise typer.Exit(1) from exc
@@ -627,9 +678,9 @@ def create(
f"'{res_name}': {exc}[/yellow]"
)
# Re-fetch project for display
# Re-fetch project for display (includes linked resources)
try:
created = repo.get(project.namespaced_name)
created = svc.get_project(project.namespaced_name)
except Exception:
created = project
@@ -673,13 +724,13 @@ def link_resource(
] = "rich",
) -> None:
"""Link a resource to a project."""
repo = _get_namespaced_project_repo()
svc = _get_namespaced_project_service()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
# Validate project exists
try:
proj = repo.get(project)
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
@@ -744,13 +795,13 @@ def unlink_resource(
] = "rich",
) -> None:
"""Unlink a resource from a project."""
repo = _get_namespaced_project_repo()
svc = _get_namespaced_project_service()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
# Validate project exists
try:
proj = repo.get(project)
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
@@ -824,10 +875,10 @@ def list_projects(
] = "rich",
) -> None:
"""List all projects in the registry."""
repo = _get_namespaced_project_repo()
svc = _get_namespaced_project_service()
try:
projects = repo.list_projects(namespace=namespace)
projects = svc.list_projects(namespace=namespace)
except DatabaseError as exc:
err_console.print(f"[red]Error listing projects:[/red] {exc.message}")
raise typer.Exit(1) from exc
@@ -885,10 +936,10 @@ def show(
] = "rich",
) -> None:
"""Show details of a project."""
repo = _get_namespaced_project_repo()
svc = _get_namespaced_project_service()
try:
proj = repo.get(project)
proj = svc.get_project(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
@@ -915,20 +966,6 @@ def show(
else:
lines.append("\n[bold]Linked Resources:[/bold] (none)")
if proj.invariants:
lines.append(
f"\n[bold]Invariants ({len(proj.invariants)}):[/bold]"
)
for inv in proj.invariants:
lines.append(f" - {inv}")
else:
lines.append("\n[bold]Invariants:[/bold] (none)")
if proj.invariant_actor:
lines.append(f"\n[bold]Invariant Actor:[/bold] {proj.invariant_actor}")
lines.append("\n[bold]Validations:[/bold] 0")
console.print(
Panel(
"\n".join(lines),
@@ -936,7 +973,60 @@ def show(
expand=False,
)
)
# --- Invariants Panel ---
invs = _get_project_invariants(proj)
if invs:
inv_table = Table(title="Invariants", show_header=True, expand=True)
inv_table.add_column("#", style="dim", justify="right", width=4)
inv_table.add_column("Invariant", style="bold")
inv_table.add_column("Source")
for i, inv_data in enumerate(invs, start=1):
inv_table.add_row(
str(i),
inv_data.get("text", ""),
inv_data.get("source", ""),
)
console.print(inv_table)
else:
console.print(
Panel(
"[dim]No project-level invariants defined.[/dim]",
title="Invariants",
expand=False,
)
)
# --- Validations Panel ---
val_attachments = _get_project_validations(proj)
if val_attachments:
val_table = Table(title="Validations", show_header=True, expand=True)
val_table.add_column("#", style="dim", justify="right", width=4)
val_table.add_column("Validation")
val_table.add_column("Resource")
val_table.add_column("Mode")
for i, att in enumerate(val_attachments, start=1):
if isinstance(att, dict):
val_name = att.get("validation_name", "")
res_id = att.get("resource_id", "")
mode = att.get("mode", "required")
else:
val_name = getattr(att, "validation_name", "")
res_id = getattr(att, "resource_id", "")
mode = getattr(att, "mode", "required")
val_table.add_row(str(i), str(val_name), str(res_id), str(mode))
console.print(val_table)
else:
console.print(
Panel(
"[dim]No validation attachments for this project.[/dim]",
title="Validations",
expand=False,
)
)
else:
data.setdefault("invariants", _get_project_invariants(proj))
data.setdefault("validations", _get_project_validations(proj))
console.print(format_output(data, output_format))
@@ -960,11 +1050,11 @@ def delete(
] = "rich",
) -> None:
"""Delete a project from the registry."""
repo = _get_namespaced_project_repo()
svc = _get_namespaced_project_service()
# Validate project exists
try:
proj = repo.get(name)
proj = svc.get_project(name)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {name}")
raise typer.Exit(1) from exc
@@ -983,7 +1073,7 @@ def delete(
raise typer.Abort()
try:
deleted = repo.delete(name)
deleted = svc.delete_project(name)
except DatabaseError as exc:
err_console.print(f"[red]Error deleting project:[/red] {exc.message}")
raise typer.Exit(1) from exc