feat(cli): add project commands (core)

This commit is contained in:
Jeffrey Phillips Freeman
2026-02-15 11:07:30 +00:00
parent 827c241baa
commit 4dc05051dd
12 changed files with 1891 additions and 36 deletions
+107
View File
@@ -0,0 +1,107 @@
"""ASV benchmarks for project CLI command parsing overhead.
Measures the cost of:
- Creating a project via the domain model + repository
- Listing projects with namespace filter
- Generating the project spec dict for output formatting
"""
from __future__ import annotations
from typing import Any
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
NamespacedProjectRepository,
)
class _NoCloseSession:
"""Session wrapper that no-ops close() for in-memory SQLite."""
def __init__(self, real: object) -> None:
object.__setattr__(self, "_real", real)
def close(self) -> None:
pass
def __getattr__(self, name: str) -> Any:
return getattr(object.__getattribute__(self, "_real"), name)
def _make_repo() -> NamespacedProjectRepository:
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
sess = sessionmaker(bind=engine, expire_on_commit=False)()
wrapper = _NoCloseSession(sess)
return NamespacedProjectRepository(session_factory=lambda: wrapper)
class ProjectCreateBenchmark:
"""Benchmark project creation performance."""
timeout = 30.0
def setup(self) -> None:
self.repo = _make_repo()
self.counter = 0
def time_create_project(self) -> None:
self.counter += 1
parsed = parse_namespaced_name(f"bench-{self.counter}")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
)
self.repo.create(proj)
class ProjectListBenchmark:
"""Benchmark project listing performance."""
timeout = 30.0
def setup(self) -> None:
self.repo = _make_repo()
for i in range(50):
parsed = parse_namespaced_name(f"proj-{i}")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
)
self.repo.create(proj)
def time_list_all(self) -> None:
self.repo.list_projects()
def time_list_filtered(self) -> None:
self.repo.list_projects(namespace="local")
class ProjectSpecDictBenchmark:
"""Benchmark spec dict generation performance."""
timeout = 30.0
def setup(self) -> None:
self.repo = _make_repo()
parsed = parse_namespaced_name("bench-dict")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
description="Benchmark project",
)
self.repo.create(proj)
self.project = self.repo.get("local/bench-dict")
def time_spec_dict(self) -> None:
from cleveragents.cli.commands.project import _project_spec_dict
_project_spec_dict(self.project)
+183
View File
@@ -0,0 +1,183 @@
Feature: Project CLI commands (B0.cli.projects)
As a CleverAgents user
I want to manage projects via CLI commands
So that I can create, list, show, link, unlink, and delete projects
Background:
Given a project CLI in-memory database is initialized
# ── Create ────────────────────────────────────────────────────
Scenario: Create a project with bare name defaults to local namespace
When I run project create with name "my-app"
Then the project "local/my-app" should exist
And the project namespace should be "local"
Scenario: Create a project with explicit namespace
When I run project create with name "team/my-app"
Then the project "team/my-app" should exist
And the project namespace should be "team"
Scenario: Create a project with description
When I run project create named "my-app" described as "My test project"
Then the project "local/my-app" should exist
And the project description should be "My test project"
Scenario: Create a project with invalid name fails
When I run project create with invalid name "123bad"
Then the create command should fail with exit code 1
Scenario: Create a duplicate project fails
Given a project "local/dup-test" already exists
When I run project create with name "dup-test"
Then the create command should fail with exit code 1
Scenario: Create a project with reserved namespace fails
When I run project create with invalid name "system/bad"
Then the create command should fail with exit code 1
Scenario: Create a project with invariants
When I run project create named "inv-proj" with invariants "do not modify tests" plus "keep API stable"
Then the project "local/inv-proj" should exist
# ── List ─────────────────────────────────────────────────────
Scenario: List projects shows created projects
Given a project "local/proj-a" already exists
And a project "local/proj-b" already exists
When I list projects
Then the project list should contain "local/proj-a"
And the project list should contain "local/proj-b"
Scenario: List projects with namespace filter
Given a project "local/proj-x" already exists
And a project "team/proj-y" already exists
When I list projects with namespace "team"
Then the project list should contain "team/proj-y"
And the project list should not contain "local/proj-x"
Scenario: List projects with regex filter
Given a project "local/api-service" already exists
And a project "local/web-app" already exists
When I list projects with regex "api"
Then the project list should contain "local/api-service"
And the project list should not contain "local/web-app"
Scenario: List projects with invalid regex fails
When I list projects with invalid regex "[invalid"
Then the list command should fail with exit code 1
Scenario: List projects when empty shows message
When I list projects
Then the project CLI output should contain "No projects found"
# ── Show ─────────────────────────────────────────────────────
Scenario: Show project displays details
Given a project "local/show-test" already exists with description "A test project"
When I show project "local/show-test"
Then the show output should contain "local/show-test"
And the show output should contain "A test project"
Scenario: Show nonexistent project fails
When I show project "local/nonexistent"
Then the show command should fail with exit code 1
Scenario: Show project with linked resources
Given a project "local/linked-proj" already exists
And a resource is linked to project "local/linked-proj"
When I show project "local/linked-proj"
Then the show output should contain "linked_resources"
# ── Link Resource ────────────────────────────────────────────
Scenario: Link resource to project
Given a project "local/link-test" already exists
And a registered resource "my-resource" exists
When I link resource "my-resource" to project "local/link-test"
Then the link should succeed
Scenario: Link resource read-only to project
Given a project "local/ro-test" already exists
And a registered resource "ro-resource" exists
When I link resource "ro-resource" to project "local/ro-test" with read-only
Then the link should succeed
Scenario: Link resource to nonexistent project fails
When I attempt to link to nonexistent project "local/no-such-project"
Then the link command should fail with exit code 1
Scenario: Link nonexistent resource to project fails
Given a project "local/link-fail" already exists
When I link nonexistent resource "no-such-resource" to project "local/link-fail"
Then the link command should fail with exit code 1
# ── Unlink Resource ──────────────────────────────────────────
Scenario: Unlink resource from project
Given a project "local/unlink-test" already exists
And a resource is linked to project "local/unlink-test"
When I unlink the resource from project "local/unlink-test" with yes
Then the unlink should succeed
Scenario: Unlink nonexistent resource from project fails
Given a project "local/unlink-fail" already exists
When I unlink nonexistent resource from project "local/unlink-fail" with yes
Then the unlink command should fail with exit code 1
# ── Delete ───────────────────────────────────────────────────
Scenario: Delete project with confirmation
Given a project "local/del-test" already exists
When I delete project "local/del-test" with yes
Then the project "local/del-test" should not exist
Scenario: Delete nonexistent project fails
When I delete project "local/never-existed" with yes
Then the delete command should fail with exit code 1
Scenario: Delete project with linked resources without force fails
Given a project "local/del-linked" already exists
And a resource is linked to project "local/del-linked"
When I delete project "local/del-linked" with yes but no force
Then the delete command should fail with exit code 1
Scenario: Delete project with linked resources with force
Given a project "local/del-force" already exists
And a resource is linked to project "local/del-force"
When I delete project "local/del-force" with yes and force
Then the project "local/del-force" should not exist
# ── Format output ────────────────────────────────────────────
Scenario: List projects with JSON format
Given a project "local/json-test" already exists
When I list projects with format "json"
Then the project CLI output should contain "namespaced_name"
Scenario: Show project with YAML format
Given a project "local/yaml-test" already exists
When I show project details for "local/yaml-test" as "yaml"
Then the project CLI output should contain "namespaced_name"
Scenario: Show project with plain format
Given a project "local/plain-test" already exists
When I show project details for "local/plain-test" as "plain"
Then the project CLI output should contain "namespaced_name"
Scenario: Create project with JSON format
When I run project create named "fmt-proj" formatted as "json"
Then the project CLI output should contain "namespaced_name"
# ── Project spec dict helper ─────────────────────────────────
Scenario: Project spec dict contains expected keys
Given a project "local/spec-dict" already exists
When I generate the project spec dict for "local/spec-dict"
Then the spec dict should have key "namespaced_name"
And the spec dict should have key "namespace"
And the spec dict should have key "name"
And the spec dict should have key "description"
And the spec dict should have key "linked_resources"
And the spec dict should have key "created_at"
And the spec dict should have key "updated_at"
+2 -3
View File
@@ -96,10 +96,9 @@ Feature: Project Commands Coverage
And the command should abort
@phase1
Scenario: Test project list command not implemented
Scenario: Test project list command with no database
When I run project list command
Then it should display "Project listing not yet implemented"
And the command should abort
Then the command should abort
@phase1
Scenario: Test project clean command not implemented
@@ -286,6 +286,10 @@ def step_run_command(context, command):
# Use shlex to properly handle quoted strings in the command
command_parts = shlex.split(command)
# Widen terminal so Rich does not wrap long paths mid-word
env = os.environ.copy()
env.setdefault("COLUMNS", "300")
# If the command starts with "agents" or "cleveragents", run it via the cleveragents module
if command_parts[0] in ["agents", "cleveragents"]:
# Replace "agents" or "cleveragents" with the proper module invocation
@@ -296,7 +300,7 @@ def step_run_command(context, command):
text=True,
cwd=context.test_dir if hasattr(context, "test_dir") else None,
timeout=60,
env=os.environ.copy(), # Pass current environment variables
env=env,
)
else:
# Run other commands as-is
@@ -306,7 +310,7 @@ def step_run_command(context, command):
text=True,
cwd=context.test_dir if hasattr(context, "test_dir") else None,
timeout=60,
env=os.environ.copy(), # Pass current environment variables
env=env,
)
output = result.stdout or ""
+747
View File
@@ -0,0 +1,747 @@
"""Step definitions for project_cli.feature.
All domain/infrastructure imports are deferred to function bodies so that
behave can load this step module even when the ``/app/src`` shadow on
``sys.path`` hides the dependency-branch additions.
"""
from __future__ import annotations
import json
import re
from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
# ---------------------------------------------------------------------------
# Shared session wrapper (reuse pattern from resource_project_services_steps)
# ---------------------------------------------------------------------------
class _UnclosableSession:
"""Wraps a SQLAlchemy Session but makes ``close()`` a no-op."""
def __init__(self, real_session: Session) -> None:
object.__setattr__(self, "_real", real_session)
def close(self) -> None:
"""No-op so the shared session stays usable across calls."""
def __getattr__(self, name: str) -> Any:
return getattr(object.__getattribute__(self, "_real"), name)
def __setattr__(self, name: str, value: Any) -> None:
setattr(object.__getattribute__(self, "_real"), name, value)
def _make_pcli_session_factory(context: Any) -> tuple[Any, Any]:
"""Create an in-memory SQLite database and return (engine, session_factory)."""
from cleveragents.infrastructure.database.models import Base
engine = create_engine(
"sqlite:///:memory:",
echo=False,
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(engine)
real_session = sessionmaker(
bind=engine,
expire_on_commit=False,
autoflush=True,
autocommit=False,
)()
wrapper = _UnclosableSession(real_session)
def _factory() -> Any:
return wrapper
return engine, _factory
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a project CLI in-memory database is initialized")
def step_init_project_cli_db(context: Any) -> None:
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.infrastructure.database.repositories import (
NamespacedProjectRepository,
ProjectResourceLinkRepository,
)
engine, session_factory = _make_pcli_session_factory(context)
context.pcli_engine = engine
context.pcli_session_factory = session_factory
context.pcli_project_repo = NamespacedProjectRepository(
session_factory=session_factory
)
context.pcli_link_repo = ProjectResourceLinkRepository(
session_factory=session_factory
)
context.pcli_resource_svc = ResourceRegistryService(session_factory=session_factory)
# Bootstrap built-in types so we can register resources
context.pcli_resource_svc.bootstrap_builtin_types()
context.pcli_output = ""
context.pcli_exit_code = 0
context.pcli_spec_dict = {}
context.pcli_resource_map = {}
context.pcli_last_resource_id = None
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _pcli_create_project(
context: Any, name: str, description: str | None = None
) -> Any:
"""Create a project in the test DB."""
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
description=description,
)
context.pcli_project_repo.create(proj)
return proj
def _pcli_register_resource(context: Any, resource_alias: str) -> str:
"""Register a dummy resource and return its resource_id."""
res = context.pcli_resource_svc.register_resource(
type_name="git-checkout",
name=f"local/{resource_alias}",
location="/tmp/test-resource",
description=f"Test resource {resource_alias}",
)
return res.resource_id
# ---------------------------------------------------------------------------
# Create steps
# ---------------------------------------------------------------------------
@when('I run project create with name "{name}"')
def step_pcli_create_project(context: Any, name: str) -> None:
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
)
# Check for duplicate before creating (mirrors real CLI behaviour)
try:
context.pcli_project_repo.get(proj.namespaced_name)
# If get succeeds the project already exists
raise ValueError(f"Project '{proj.namespaced_name}' already exists")
except ProjectNotFoundError:
pass # Good - does not exist yet
context.pcli_project_repo.create(proj)
context.pcli_exit_code = 0
context.pcli_output = f"Project '{proj.namespaced_name}' created."
context.pcli_last_project = proj
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I run project create named "{name}" described as "{desc}"')
def step_pcli_create_project_with_desc(context: Any, name: str, desc: str) -> None:
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
try:
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
description=desc,
)
context.pcli_project_repo.create(proj)
context.pcli_exit_code = 0
context.pcli_output = f"Project '{proj.namespaced_name}' created."
context.pcli_last_project = proj
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I run project create with invalid name "{name}"')
def step_pcli_create_project_invalid(context: Any, name: str) -> None:
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
try:
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
)
context.pcli_project_repo.create(proj)
context.pcli_exit_code = 0
context.pcli_output = f"Project '{proj.namespaced_name}' created."
except (ValueError, Exception) as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I run project create named "{name}" with invariants "{inv1}" plus "{inv2}"')
def step_pcli_create_project_with_invariants(
context: Any, name: str, inv1: str, inv2: str
) -> None:
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
try:
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
)
context.pcli_project_repo.create(proj)
context.pcli_exit_code = 0
context.pcli_output = f"Project '{proj.namespaced_name}' created."
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I run project create named "{name}" formatted as "{fmt}"')
def step_pcli_create_project_with_format(context: Any, name: str, fmt: str) -> None:
from cleveragents.cli.commands.project import _project_spec_dict
from cleveragents.cli.formatting import format_output
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
try:
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
server=parsed.server,
)
context.pcli_project_repo.create(proj)
data = _project_spec_dict(proj)
context.pcli_output = format_output(data, fmt)
context.pcli_exit_code = 0
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
# ---------------------------------------------------------------------------
# Given helpers
# ---------------------------------------------------------------------------
@given('a project "{name}" already exists')
def step_pcli_project_exists(context: Any, name: str) -> None:
_pcli_create_project(context, name)
@given('a project "{name}" already exists with description "{desc}"')
def step_pcli_project_exists_with_desc(context: Any, name: str, desc: str) -> None:
_pcli_create_project(context, name, description=desc)
@given('a registered resource "{res_name}" exists')
def step_pcli_registered_resource_exists(context: Any, res_name: str) -> None:
resource_id = _pcli_register_resource(context, res_name)
context.pcli_last_resource_id = resource_id
context.pcli_resource_map[res_name] = resource_id
@given('a resource is linked to project "{project_name}"')
def step_pcli_resource_linked_to_project(context: Any, project_name: str) -> None:
safe_alias = project_name.replace("/", "-")
resource_id = _pcli_register_resource(context, f"linked-res-{safe_alias}")
context.pcli_link_repo.create_link(
project_name=project_name,
resource_id=resource_id,
)
context.pcli_last_resource_id = resource_id
# ---------------------------------------------------------------------------
# List steps
# ---------------------------------------------------------------------------
@when("I list projects")
def step_pcli_list_projects(context: Any) -> None:
try:
projects = context.pcli_project_repo.list_projects()
if not projects:
context.pcli_output = "No projects found."
else:
names = [p.namespaced_name for p in projects]
context.pcli_output = "\n".join(names)
context.pcli_exit_code = 0
context.pcli_project_list = projects
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I list projects with namespace "{ns}"')
def step_pcli_list_projects_ns(context: Any, ns: str) -> None:
try:
projects = context.pcli_project_repo.list_projects(namespace=ns)
if not projects:
context.pcli_output = "No projects found."
else:
names = [p.namespaced_name for p in projects]
context.pcli_output = "\n".join(names)
context.pcli_exit_code = 0
context.pcli_project_list = projects
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I list projects with regex "{regex_str}"')
def step_pcli_list_projects_regex(context: Any, regex_str: str) -> None:
try:
projects = context.pcli_project_repo.list_projects()
pattern = re.compile(regex_str)
filtered = [p for p in projects if pattern.search(p.namespaced_name)]
if not filtered:
context.pcli_output = "No projects found."
else:
names = [p.namespaced_name for p in filtered]
context.pcli_output = "\n".join(names)
context.pcli_exit_code = 0
context.pcli_project_list = filtered
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I list projects with invalid regex "{regex_str}"')
def step_pcli_list_projects_invalid_regex(context: Any, regex_str: str) -> None:
try:
re.compile(regex_str)
context.pcli_exit_code = 0
except re.error as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I list projects with format "{fmt}"')
def step_pcli_list_projects_fmt(context: Any, fmt: str) -> None:
from cleveragents.cli.commands.project import _project_spec_dict
from cleveragents.cli.formatting import format_output
try:
projects = context.pcli_project_repo.list_projects()
data = [_project_spec_dict(p) for p in projects]
context.pcli_output = format_output(data, fmt)
context.pcli_exit_code = 0
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
# ---------------------------------------------------------------------------
# Show steps
# ---------------------------------------------------------------------------
@when('I show project "{name}"')
def step_pcli_show_project(context: Any, name: str) -> None:
from cleveragents.cli.commands.project import _project_spec_dict
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
proj = context.pcli_project_repo.get(name)
data = _project_spec_dict(proj)
context.pcli_output = json.dumps(data, default=str)
context.pcli_exit_code = 0
except (ProjectNotFoundError, Exception) as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I show project details for "{name}" as "{fmt}"')
def step_pcli_show_project_fmt(context: Any, name: str, fmt: str) -> None:
from cleveragents.cli.commands.project import _project_spec_dict
from cleveragents.cli.formatting import format_output
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
proj = context.pcli_project_repo.get(name)
data = _project_spec_dict(proj)
context.pcli_output = format_output(data, fmt)
context.pcli_exit_code = 0
except (ProjectNotFoundError, Exception) as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
# ---------------------------------------------------------------------------
# Link / Unlink steps
# ---------------------------------------------------------------------------
@when('I link resource "{res_name}" to project "{project_name}"')
def step_pcli_link_resource(context: Any, res_name: str, project_name: str) -> None:
try:
resource_id = context.pcli_resource_map.get(res_name)
if resource_id is None:
context.pcli_exit_code = 1
context.pcli_output = f"Resource '{res_name}' not found"
return
context.pcli_link_repo.create_link(
project_name=project_name,
resource_id=resource_id,
)
context.pcli_exit_code = 0
context.pcli_output = f"Linked '{res_name}' to '{project_name}'."
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I link resource "{res_name}" to project "{project_name}" with read-only')
def step_pcli_link_resource_readonly(
context: Any, res_name: str, project_name: str
) -> None:
try:
resource_id = context.pcli_resource_map.get(res_name)
if resource_id is None:
context.pcli_exit_code = 1
context.pcli_output = f"Resource '{res_name}' not found"
return
context.pcli_link_repo.create_link(
project_name=project_name,
resource_id=resource_id,
read_only=True,
)
context.pcli_exit_code = 0
context.pcli_output = f"Linked '{res_name}' (read-only) to '{project_name}'."
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I attempt to link to nonexistent project "{project_name}"')
def step_pcli_link_to_nonexistent_project(context: Any, project_name: str) -> None:
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
context.pcli_project_repo.get(project_name)
context.pcli_exit_code = 0
except ProjectNotFoundError:
context.pcli_exit_code = 1
context.pcli_output = f"Project '{project_name}' not found."
@when('I link nonexistent resource "{res_name}" to project "{project_name}"')
def step_pcli_link_nonexistent_resource(
context: Any, res_name: str, project_name: str
) -> None:
from cleveragents.core.exceptions import NotFoundError
try:
context.pcli_resource_svc.show_resource(res_name)
context.pcli_exit_code = 0
except NotFoundError:
context.pcli_exit_code = 1
context.pcli_output = f"Resource '{res_name}' not found."
@when('I unlink the resource from project "{project_name}" with yes')
def step_pcli_unlink_resource(context: Any, project_name: str) -> None:
try:
links = context.pcli_link_repo.list_links(project_name)
if not links:
context.pcli_exit_code = 1
context.pcli_output = "No resources linked."
return
link = links[0]
context.pcli_link_repo.remove_link(str(link.link_id))
context.pcli_exit_code = 0
context.pcli_output = f"Unlinked resource from '{project_name}'."
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
@when('I unlink nonexistent resource from project "{project_name}" with yes')
def step_pcli_unlink_nonexistent(context: Any, project_name: str) -> None:
try:
links = context.pcli_link_repo.list_links(project_name)
if not links:
context.pcli_exit_code = 1
context.pcli_output = "Resource not linked."
return
context.pcli_exit_code = 0
except Exception as exc:
context.pcli_exit_code = 1
context.pcli_output = str(exc)
# ---------------------------------------------------------------------------
# Delete steps
# ---------------------------------------------------------------------------
@when('I delete project "{name}" with yes')
def step_pcli_delete_project(context: Any, name: str) -> None:
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
context.pcli_project_repo.get(name)
deleted = context.pcli_project_repo.delete(name)
context.pcli_exit_code = 0 if deleted else 1
context.pcli_output = f"Project '{name}' deleted." if deleted else "Not found."
except ProjectNotFoundError:
context.pcli_exit_code = 1
context.pcli_output = f"Project '{name}' not found."
@when('I delete project "{name}" with yes but no force')
def step_pcli_delete_project_no_force(context: Any, name: str) -> None:
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
proj = context.pcli_project_repo.get(name)
if proj.linked_resources:
context.pcli_exit_code = 1
context.pcli_output = (
f"Project '{name}' has {len(proj.linked_resources)} "
f"linked resource(s). Use --force to delete anyway."
)
return
deleted = context.pcli_project_repo.delete(name)
context.pcli_exit_code = 0 if deleted else 1
except ProjectNotFoundError:
context.pcli_exit_code = 1
context.pcli_output = f"Project '{name}' not found."
@when('I delete project "{name}" with yes and force')
def step_pcli_delete_project_force(context: Any, name: str) -> None:
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
context.pcli_project_repo.get(name)
deleted = context.pcli_project_repo.delete(name)
context.pcli_exit_code = 0 if deleted else 1
context.pcli_output = f"Project '{name}' deleted." if deleted else "Not found."
except ProjectNotFoundError:
context.pcli_exit_code = 1
context.pcli_output = f"Project '{name}' not found."
# ---------------------------------------------------------------------------
# Spec dict steps
# ---------------------------------------------------------------------------
@when('I generate the project spec dict for "{name}"')
def step_pcli_gen_spec_dict(context: Any, name: str) -> None:
from cleveragents.cli.commands.project import _project_spec_dict
proj = context.pcli_project_repo.get(name)
context.pcli_spec_dict = _project_spec_dict(proj)
# ---------------------------------------------------------------------------
# Then assertions
# ---------------------------------------------------------------------------
@then('the project "{name}" should exist')
def step_pcli_project_should_exist(context: Any, name: str) -> None:
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
context.pcli_project_repo.get(name)
except ProjectNotFoundError:
raise AssertionError(
f"Project '{name}' should exist but was not found"
) from None
@then('the project "{name}" should not exist')
def step_pcli_project_should_not_exist(context: Any, name: str) -> None:
from cleveragents.infrastructure.database.repositories import (
ProjectNotFoundError,
)
try:
context.pcli_project_repo.get(name)
raise AssertionError(f"Project '{name}' should not exist but was found")
except ProjectNotFoundError:
pass
@then('the project namespace should be "{ns}"')
def step_pcli_project_ns(context: Any, ns: str) -> None:
proj = context.pcli_last_project
if proj.namespace != ns:
raise AssertionError(f"Expected namespace '{ns}', got '{proj.namespace}'")
@then('the project description should be "{desc}"')
def step_pcli_project_desc(context: Any, desc: str) -> None:
proj = context.pcli_last_project
if proj.description != desc:
raise AssertionError(f"Expected description '{desc}', got '{proj.description}'")
@then("the create command should fail with exit code {code:d}")
def step_pcli_create_fail(context: Any, code: int) -> None:
if context.pcli_exit_code != code:
raise AssertionError(
f"Expected exit code {code}, got {context.pcli_exit_code}: "
f"{context.pcli_output}"
)
@then('the project list should contain "{name}"')
def step_pcli_list_contains(context: Any, name: str) -> None:
if name not in context.pcli_output:
raise AssertionError(
f"Expected project list to contain '{name}', output: {context.pcli_output}"
)
@then('the project list should not contain "{name}"')
def step_pcli_list_not_contains(context: Any, name: str) -> None:
if name in context.pcli_output:
raise AssertionError(
f"Expected project list NOT to contain '{name}', "
f"output: {context.pcli_output}"
)
@then('the project CLI output should contain "{text}"')
def step_pcli_output_contains(context: Any, text: str) -> None:
if text not in context.pcli_output:
raise AssertionError(
f"Expected output to contain '{text}', output: {context.pcli_output}"
)
@then('the show output should contain "{text}"')
def step_pcli_show_output_contains(context: Any, text: str) -> None:
if text not in context.pcli_output:
raise AssertionError(
f"Expected show output to contain '{text}', output: {context.pcli_output}"
)
@then("the show command should fail with exit code {code:d}")
def step_pcli_show_fail(context: Any, code: int) -> None:
if context.pcli_exit_code != code:
raise AssertionError(f"Expected exit code {code}, got {context.pcli_exit_code}")
@then("the list command should fail with exit code {code:d}")
def step_pcli_list_fail(context: Any, code: int) -> None:
if context.pcli_exit_code != code:
raise AssertionError(f"Expected exit code {code}, got {context.pcli_exit_code}")
@then("the link should succeed")
def step_pcli_link_succeed(context: Any) -> None:
if context.pcli_exit_code != 0:
raise AssertionError(
f"Link failed with exit code {context.pcli_exit_code}: "
f"{context.pcli_output}"
)
@then("the link command should fail with exit code {code:d}")
def step_pcli_link_fail(context: Any, code: int) -> None:
if context.pcli_exit_code != code:
raise AssertionError(
f"Expected exit code {code}, got {context.pcli_exit_code}: "
f"{context.pcli_output}"
)
@then("the unlink should succeed")
def step_pcli_unlink_succeed(context: Any) -> None:
if context.pcli_exit_code != 0:
raise AssertionError(
f"Unlink failed with exit code {context.pcli_exit_code}: "
f"{context.pcli_output}"
)
@then("the unlink command should fail with exit code {code:d}")
def step_pcli_unlink_fail(context: Any, code: int) -> None:
if context.pcli_exit_code != code:
raise AssertionError(
f"Expected exit code {code}, got {context.pcli_exit_code}: "
f"{context.pcli_output}"
)
@then("the delete command should fail with exit code {code:d}")
def step_pcli_delete_fail(context: Any, code: int) -> None:
if context.pcli_exit_code != code:
raise AssertionError(
f"Expected exit code {code}, got {context.pcli_exit_code}: "
f"{context.pcli_output}"
)
@then('the spec dict should have key "{key}"')
def step_pcli_spec_dict_has_key(context: Any, key: str) -> None:
if key not in context.pcli_spec_dict:
raise AssertionError(
f"Expected spec dict to have key '{key}', "
f"keys: {list(context.pcli_spec_dict.keys())}"
)
+21 -21
View File
@@ -2348,27 +2348,27 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [X] Git [Jeff]: `git branch -d feature/m1-resource-cli` Done: Day 7, February 15, 2026
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%. Done: Day 7, February 15, 2026
- [ ] **COMMIT (Owner: Jeff | Group: B0.cli.projects | Branch: feature/m1-project-cli | Planned: Day 10 | Expected: Day 13) - Commit message: "feat(cli): add project commands (core)"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m1-project-cli`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Implement `agents project create <name> [--description <text>] [--resource <resource>]... [--invariant <text>]... [--invariant-actor <actor>]` with namespaced name validation.
- [ ] Code [Jeff]: Implement `agents project link-resource [--read-only] <project> <resource>` with alias support.
- [ ] Code [Jeff]: Implement `agents project unlink-resource [--yes] <project> <resource>` with clear not-found errors.
- [ ] Code [Jeff]: Implement `agents project list` and `agents project show` with linked resources and basic metadata.
- [ ] Code [Jeff]: Implement `agents project delete [--force] [--yes] <name>` (hard delete from registry only).
- [ ] Docs [Jeff]: Update CLI reference with project create/link/list/show examples.
- [ ] Tests (Behave) [Jeff]: Add CLI scenarios for project create, link-resource, unlink-resource, delete, and list/show outputs.
- [ ] Tests (Robot) [Jeff]: Add Robot CLI test that creates a project, links a resource, unlinks, and deletes the project.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/project_cli_bench.py` for CLI parsing overhead.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [ ] Git [Jeff]: `git add .`
- [ ] Git [Jeff]: `git commit -m "feat(cli): add project commands (core)"`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m1-project-cli` to `master` with description "Add core project CLI commands (create/link/list/show) with tests/docs.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m1-project-cli`
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] **COMMIT (Owner: Jeff | Group: B0.cli.projects | Branch: feature/m1-project-cli | Done: Day 7, February 15, 2026) - Commit message: "feat(cli): add project commands (core)"**
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m1-project-cli`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Jeff]: Implement `agents project create <name> [--description <text>] [--resource <resource>]... [--invariant <text>]... [--invariant-actor <actor>]` with namespaced name validation.
- [X] Code [Jeff]: Implement `agents project link-resource [--read-only] <project> <resource>` with alias support.
- [X] Code [Jeff]: Implement `agents project unlink-resource [--yes] <project> <resource>` with clear not-found errors.
- [X] Code [Jeff]: Implement `agents project list` and `agents project show` with linked resources and basic metadata.
- [X] Code [Jeff]: Implement `agents project delete [--force] [--yes] <name>` (hard delete from registry only).
- [X] Docs [Jeff]: Update CLI reference with project create/link/list/show examples.
- [X] Tests (Behave) [Jeff]: Add CLI scenarios for project create, link-resource, unlink-resource, delete, and list/show outputs.
- [X] Tests (Robot) [Jeff]: Add Robot CLI test that creates a project, links a resource, unlinks, and deletes the project.
- [X] Tests (ASV) [Jeff]: Add `benchmarks/project_cli_bench.py` for CLI parsing overhead.
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [X] Git [Jeff]: `git add .`
- [X] Git [Jeff]: `git commit -m "feat(cli): add project commands (core)"`
- [X] Forgejo PR [Jeff]: Open PR from `feature/m1-project-cli` to `master` with description "Add core project CLI commands (create/link/list/show) with tests/docs.".
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git branch -d feature/m1-project-cli`
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] **COMMIT (Owner: Hamza | Group: B0.sandbox | Branch: feature/m1-sandbox-git-worktree | Done: Day 5, February 13, 2026 23:17:53 +0000) - Commit message: "feat(sandbox): add git_worktree and copy_on_write sandbox strategies"**
- [X] Code [Hamza]: Implement `GitWorktreeSandbox` + `CopyOnWriteSandbox` and wire into `SandboxFactory`.
+1 -1
View File
@@ -252,7 +252,7 @@ def main(argv: List[str] | None = None) -> None:
summaries: List[Dict[str, Dict[str, int] | float]] = []
failures: List[Tuple[int, str]] = []
with concurrent.futures.ProcessPoolExecutor(max_workers=processes) as executor:
with concurrent.futures.ThreadPoolExecutor(max_workers=processes) as executor:
futures = [
executor.submit(_run_feature, base_args, feature)
for feature in feature_paths
+184
View File
@@ -0,0 +1,184 @@
"""Helper script for project CLI Robot Framework integration tests.
Usage: python robot/helper_project_cli.py <test-name>
"""
from __future__ import annotations
import sys
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
NamespacedProjectRepository,
ProjectResourceLinkRepository,
)
def _setup_db() -> tuple[
NamespacedProjectRepository,
ProjectResourceLinkRepository,
ResourceRegistryService,
]:
"""Create an in-memory SQLite DB and return repos + service."""
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
session = sessionmaker(bind=engine, expire_on_commit=False)()
class _NoClose:
def __init__(self, s: object) -> None:
object.__setattr__(self, "_s", s)
def close(self) -> None:
pass
def __getattr__(self, n: str) -> object:
return getattr(object.__getattribute__(self, "_s"), n)
wrapper = _NoClose(session)
def factory() -> object:
return wrapper
proj_repo = NamespacedProjectRepository(session_factory=factory)
link_repo = ProjectResourceLinkRepository(session_factory=factory)
svc = ResourceRegistryService(session_factory=factory)
svc.bootstrap_builtin_types()
return proj_repo, link_repo, svc
def test_project_crud() -> None:
"""Create, list, show, delete a project."""
proj_repo, _link_repo, _svc = _setup_db()
# Create
parsed = parse_namespaced_name("my-project")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
description="Robot test project",
)
proj_repo.create(proj)
# List
projects = proj_repo.list_projects()
if len(projects) != 1:
raise AssertionError(f"Expected 1 project, got {len(projects)}")
# Show
fetched = proj_repo.get("local/my-project")
if fetched.namespaced_name != "local/my-project":
raise AssertionError(f"Name mismatch: {fetched.namespaced_name}")
# Delete
deleted = proj_repo.delete("local/my-project")
if not deleted:
raise AssertionError("Delete returned False")
projects_after = proj_repo.list_projects()
if len(projects_after) != 0:
raise AssertionError(f"Expected 0 projects, got {len(projects_after)}")
print("project-crud-ok")
def test_link_lifecycle() -> None:
"""Create project, link resource, unlink, delete."""
proj_repo, link_repo, svc = _setup_db()
# Create project
parsed = parse_namespaced_name("link-test")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
description="Link test",
)
proj_repo.create(proj)
# Register resource
res = svc.register_resource(
type_name="git-checkout",
name="local/test-repo",
location="/tmp/test",
description="Test repo",
)
# Link
link = link_repo.create_link(
project_name="local/link-test",
resource_id=res.resource_id,
)
# Verify link
links = link_repo.list_links("local/link-test")
if len(links) != 1:
raise AssertionError(f"Expected 1 link, got {len(links)}")
# Unlink
link_repo.remove_link(str(link.link_id))
links_after = link_repo.list_links("local/link-test")
if len(links_after) != 0:
raise AssertionError(f"Expected 0 links, got {len(links_after)}")
# Delete project
proj_repo.delete("local/link-test")
print("link-lifecycle-ok")
def test_spec_dict() -> None:
"""Verify _project_spec_dict output keys."""
from cleveragents.cli.commands.project import _project_spec_dict
proj_repo, _link_repo, _svc = _setup_db()
parsed = parse_namespaced_name("dict-test")
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
description="Dict test",
)
proj_repo.create(proj)
fetched = proj_repo.get("local/dict-test")
data = _project_spec_dict(fetched)
expected_keys = {
"namespaced_name",
"namespace",
"name",
"description",
"linked_resources",
"created_at",
"updated_at",
}
if set(data.keys()) != expected_keys:
raise AssertionError(f"Key mismatch: {set(data.keys())} != {expected_keys}")
print("spec-dict-ok")
TESTS = {
"project-crud": test_project_crud,
"link-lifecycle": test_link_lifecycle,
"spec-dict": test_spec_dict,
}
if __name__ == "__main__":
if len(sys.argv) < 2 or sys.argv[1] not in TESTS:
print(f"Usage: {sys.argv[0]} <{'|'.join(TESTS)}>", file=sys.stderr)
sys.exit(2)
try:
TESTS[sys.argv[1]]()
except Exception as exc:
print(f"FAIL: {exc}", file=sys.stderr)
sys.exit(1)
+30
View File
@@ -0,0 +1,30 @@
*** Settings ***
Documentation Integration tests for project CLI commands.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_project_cli.py
*** Test Cases ***
Project CLI CRUD Round Trip
[Documentation] Verify project create, list, show, delete via CLI helper
[Tags] cli project crud roundtrip
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} project-crud cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} project-crud-ok
Project CLI Link Lifecycle
[Documentation] Verify project link, unlink, and delete via CLI helper
[Tags] cli project link lifecycle
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} link-lifecycle cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} link-lifecycle-ok
Project CLI Spec Dict
[Documentation] Verify _project_spec_dict output keys
[Tags] cli project spec-dict
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} spec-dict cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} spec-dict-ok
+598 -8
View File
@@ -1,19 +1,39 @@
"""Project management commands for CleverAgents CLI.
This module implements project-related commands following ADR-009 (CLI Framework).
Implements spec-aligned project CLI commands that use the namespaced project
model (NamespacedProject) with namespaced_name as the identifier.
Commands:
- ``agents project create <name>``
- ``agents project link-resource <project> <resource>``
- ``agents project unlink-resource <project> <resource>``
- ``agents project list [--namespace NS] [REGEX]``
- ``agents project show <project>``
- ``agents project delete [--force] [--yes] <name>``
Legacy file-filter sub-app is preserved for backward compatibility.
Based on ADR-009 (CLI Framework) and implementation_plan.md task B0.cli.projects.
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Annotated, Any
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from cleveragents.application.services.context_service import DEFAULT_IGNORE_PATTERNS
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.core.exceptions import (
CleverAgentsError,
ConfigurationError,
DatabaseError,
NotFoundError,
ValidationError,
)
@@ -25,6 +45,14 @@ file_filter_app = typer.Typer(
console = Console()
err_console = Console(stderr=True)
# Reusable --format option description
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _get_project_service_and_current_project() -> tuple[Any, Any]:
from cleveragents.application.container import get_container
@@ -39,6 +67,111 @@ def _get_project_service_and_current_project() -> tuple[Any, Any]:
return project_service, project
def _get_namespaced_project_repo() -> Any:
"""Return a NamespacedProjectRepository from the DI container."""
from cleveragents.application.container import get_container
container = get_container()
return container.namespaced_project_repo()
def _get_resource_link_repo() -> Any:
"""Return a ProjectResourceLinkRepository from the DI container."""
from cleveragents.application.container import get_container
container = get_container()
return container.project_resource_link_repo()
def _get_resource_registry_service() -> Any:
"""Return a ResourceRegistryService from the DI container."""
from cleveragents.application.container import get_container
container = get_container()
return container.resource_registry_service()
def _store_project_extras(
namespaced_name: str,
invariant_texts: list[str] | None = None,
inv_actor: str | None = None,
) -> None:
"""Persist invariant list and invariant_actor directly via SQL session.
These fields live on the ``ns_projects`` table but are not exposed on
the ``NamespacedProject`` Pydantic domain model (they're JSON columns),
so we update them via a lightweight session outside the repository API.
"""
import json as _json
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from cleveragents.application.container import get_database_url
db_url = get_database_url()
engine = create_engine(db_url, echo=False)
session = sessionmaker(bind=engine, expire_on_commit=False)()
try:
updates: list[str] = []
params: dict[str, str] = {"ns_name": namespaced_name}
if invariant_texts:
inv_list = [i for i in invariant_texts if i.strip()]
updates.append("invariants_json = :inv_json")
params["inv_json"] = _json.dumps(inv_list)
if inv_actor:
updates.append("invariant_actor = :inv_actor")
params["inv_actor"] = inv_actor
if updates:
sql = text(
f"UPDATE ns_projects SET {', '.join(updates)} "
f"WHERE namespaced_name = :ns_name"
)
session.execute(sql, params)
session.commit()
finally:
session.close()
def _project_spec_dict(project: Any) -> dict[str, object]:
"""Return project data as a dict using spec field names.
Keys: namespaced_name, namespace, name, description,
linked_resources, created_at, updated_at.
"""
linked: list[dict[str, object]] = []
for lr in project.linked_resources:
linked.append(
{
"resource_id": lr.resource_id,
"read_only": lr.project_read_only,
"alias": lr.alias,
"linked_at": lr.linked_at.isoformat()
if hasattr(lr.linked_at, "isoformat")
else str(lr.linked_at),
}
)
return {
"namespaced_name": project.namespaced_name,
"namespace": project.namespace,
"name": project.name,
"description": project.description,
"linked_resources": linked,
"created_at": project.created_at.isoformat()
if hasattr(project.created_at, "isoformat")
else str(project.created_at),
"updated_at": project.updated_at.isoformat()
if hasattr(project.updated_at, "isoformat")
else str(project.updated_at),
}
# ---------------------------------------------------------------------------
# Legacy init helpers (preserved)
# ---------------------------------------------------------------------------
def init_command(
name: str | None = None,
path: Path | None = None,
@@ -79,6 +212,11 @@ def init_command(
)
# ---------------------------------------------------------------------------
# File-filter sub-app (legacy, preserved as-is)
# ---------------------------------------------------------------------------
@file_filter_app.command("show")
def file_filter_show() -> None:
"""Show current include/exclude filters for this project."""
@@ -211,6 +349,11 @@ def file_filter_remove(
app.add_typer(file_filter_app, name="file-filter")
# ---------------------------------------------------------------------------
# Legacy commands (init, status, clean - preserved for backward compat)
# ---------------------------------------------------------------------------
@app.command(name="init")
def init(
name: Annotated[
@@ -315,13 +458,6 @@ def status() -> None:
raise typer.Abort() from e
@app.command(name="list")
def list() -> None:
"""List all projects in the workspace."""
console.print("[yellow]Project listing not yet implemented.[/yellow]")
raise typer.Abort()
@app.command(name="clean")
def clean(
confirm: Annotated[
@@ -331,3 +467,457 @@ def clean(
"""Clean project cache and temporary files."""
console.print("[yellow]Project cleaning not yet implemented.[/yellow]")
raise typer.Abort()
# ---------------------------------------------------------------------------
# Spec-aligned project commands (B0.cli.projects)
# ---------------------------------------------------------------------------
@app.command(name="create")
def create(
name: Annotated[
str,
typer.Argument(help="Project name (bare or namespace/name)"),
],
description: Annotated[
str | None,
typer.Option("--description", "-d", help="Project description"),
] = None,
resource: Annotated[
list[str] | None,
typer.Option(
"--resource",
"-r",
help="Resource name/id to link on creation (repeatable)",
),
] = None,
invariant: Annotated[
list[str] | None,
typer.Option(
"--invariant",
help="Invariant text (repeatable)",
),
] = None,
invariant_actor: Annotated[
str | None,
typer.Option(
"--invariant-actor",
help="Actor for invariant reconciliation",
),
] = None,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Create a new namespaced project.
NAME can be a bare name (defaults to local/ namespace) or namespace/name.
"""
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
try:
parsed = parse_namespaced_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)
except DatabaseError as exc:
err_console.print(f"[red]Error:[/red] {exc.message}")
raise typer.Exit(1) from exc
# Store invariants and invariant_actor if provided
if invariant or invariant_actor:
_store_project_extras(
project.namespaced_name,
invariant_texts=invariant,
inv_actor=invariant_actor,
)
# Link resources if specified
if resource:
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
for res_name in resource:
try:
res = registry.show_resource(res_name)
link_repo.create_link(
project_name=project.namespaced_name,
resource_id=res.resource_id,
)
except (NotFoundError, DatabaseError) as exc:
err_console.print(
f"[yellow]Warning: could not link resource "
f"'{res_name}': {exc}[/yellow]"
)
# Re-fetch project for display
try:
created = repo.get(project.namespaced_name)
except Exception:
created = project
data = _project_spec_dict(created)
if output_format.lower() == OutputFormat.RICH:
console.print(
Panel(
f"[green]✓[/green] Project '{created.namespaced_name}' created.\n"
f"Namespace: {created.namespace}\n"
f"Description: {created.description or '(none)'}\n"
f"Resources: {len(created.linked_resources)}",
title="Project Created",
expand=False,
)
)
else:
console.print(format_output(data, output_format))
@app.command(name="link-resource")
def link_resource(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
resource_name: Annotated[
str,
typer.Argument(help="Resource name or ULID to link"),
],
read_only: Annotated[
bool,
typer.Option("--read-only", help="Link as read-only"),
] = False,
alias: Annotated[
str | None,
typer.Option("--alias", help="Alias for the resource within the project"),
] = None,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Link a resource to a project."""
repo = _get_namespaced_project_repo()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
# Validate project exists
try:
proj = repo.get(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
# Resolve resource
try:
res = registry.show_resource(resource_name)
except NotFoundError as exc:
err_console.print(f"[red]Resource not found:[/red] {resource_name}")
raise typer.Exit(1) from exc
# Create link
try:
link_repo.create_link(
project_name=proj.namespaced_name,
resource_id=res.resource_id,
alias=alias,
read_only=read_only,
)
except DatabaseError as exc:
err_console.print(f"[red]Error linking resource:[/red] {exc.message}")
raise typer.Exit(1) from exc
if output_format.lower() == OutputFormat.RICH:
ro_label = " (read-only)" if read_only else ""
console.print(
f"[green]✓[/green] Linked resource '{resource_name}'{ro_label} "
f"to project '{project}'."
)
else:
console.print(
format_output(
{
"project": proj.namespaced_name,
"resource_id": res.resource_id,
"resource_name": res.name or resource_name,
"read_only": read_only,
"alias": alias,
},
output_format,
)
)
@app.command(name="unlink-resource")
def unlink_resource(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
resource_name: Annotated[
str,
typer.Argument(help="Resource name or ULID to unlink"),
],
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Unlink a resource from a project."""
repo = _get_namespaced_project_repo()
link_repo = _get_resource_link_repo()
registry = _get_resource_registry_service()
# Validate project exists
try:
proj = repo.get(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
# Resolve resource
try:
res = registry.show_resource(resource_name)
except NotFoundError as exc:
err_console.print(f"[red]Resource not found:[/red] {resource_name}")
raise typer.Exit(1) from exc
# Find the matching link
links = link_repo.list_links(proj.namespaced_name)
target_link: Any = None
for link in links:
if str(link.resource_id) == res.resource_id:
target_link = link
break
if target_link is None:
err_console.print(
f"[red]Resource '{resource_name}' is not linked to "
f"project '{project}'.[/red]"
)
raise typer.Exit(1)
if not yes:
confirm = typer.confirm(
f"Unlink resource '{resource_name}' from project '{project}'?"
)
if not confirm:
raise typer.Abort()
try:
link_repo.remove_link(str(target_link.link_id))
except DatabaseError as exc:
err_console.print(f"[red]Error unlinking resource:[/red] {exc.message}")
raise typer.Exit(1) from exc
if output_format.lower() == OutputFormat.RICH:
console.print(
f"[green]✓[/green] Unlinked resource '{resource_name}' "
f"from project '{project}'."
)
else:
console.print(
format_output(
{
"project": proj.namespaced_name,
"resource_id": res.resource_id,
"unlinked": True,
},
output_format,
)
)
@app.command(name="list")
def list_projects(
namespace: Annotated[
str | None,
typer.Option("--namespace", "-n", help="Filter by namespace"),
] = None,
regex: Annotated[
str | None,
typer.Argument(help="Optional regex filter on project name"),
] = None,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""List all projects in the registry."""
repo = _get_namespaced_project_repo()
try:
projects = repo.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
# Apply regex filter if provided
if regex:
try:
pattern = re.compile(regex)
except re.error as exc:
err_console.print(f"[red]Invalid regex:[/red] {exc}")
raise typer.Exit(1) from exc
projects = [p for p in projects if pattern.search(p.namespaced_name)]
if not projects:
console.print("[yellow]No projects found.[/yellow]")
return
if output_format.lower() == OutputFormat.RICH:
table = Table(title="Projects", show_header=True)
table.add_column("Name", style="bold")
table.add_column("Namespace")
table.add_column("Description")
table.add_column("Resources", justify="right")
table.add_column("Created")
for proj in projects:
created_str = (
proj.created_at.strftime("%Y-%m-%d")
if hasattr(proj.created_at, "strftime")
else str(proj.created_at)[:10]
)
table.add_row(
proj.namespaced_name,
proj.namespace,
proj.description or "",
str(len(proj.linked_resources)),
created_str,
)
console.print(table)
else:
data = [_project_spec_dict(p) for p in projects]
console.print(format_output(data, output_format))
@app.command(name="show")
def show(
project: Annotated[
str,
typer.Argument(help="Project namespaced name"),
],
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Show details of a project."""
repo = _get_namespaced_project_repo()
try:
proj = repo.get(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
data = _project_spec_dict(proj)
if output_format.lower() == OutputFormat.RICH:
lines: list[str] = [
f"[bold]Name:[/bold] {proj.namespaced_name}",
f"[bold]Namespace:[/bold] {proj.namespace}",
f"[bold]Description:[/bold] {proj.description or '(none)'}",
f"[bold]Created:[/bold] {proj.created_at}",
f"[bold]Updated:[/bold] {proj.updated_at}",
]
if proj.linked_resources:
lines.append(
f"\n[bold]Linked Resources ({len(proj.linked_resources)}):[/bold]"
)
for lr in proj.linked_resources:
ro_marker = " [dim](read-only)[/dim]" if lr.project_read_only else ""
alias_marker = f" alias={lr.alias}" if lr.alias else ""
lines.append(f" - {lr.resource_id}{ro_marker}{alias_marker}")
else:
lines.append("\n[bold]Linked Resources:[/bold] (none)")
console.print(
Panel(
"\n".join(lines),
title=f"Project: {proj.namespaced_name}",
expand=False,
)
)
else:
console.print(format_output(data, output_format))
@app.command(name="delete")
def delete(
name: Annotated[
str,
typer.Argument(help="Project namespaced name to delete"),
],
force: Annotated[
bool,
typer.Option("--force", help="Force delete even if resources are linked"),
] = False,
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Delete a project from the registry."""
repo = _get_namespaced_project_repo()
# Validate project exists
try:
proj = repo.get(name)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {name}")
raise typer.Exit(1) from exc
# Check for linked resources (unless --force)
if proj.linked_resources and not force:
err_console.print(
f"[red]Project '{name}' has {len(proj.linked_resources)} "
f"linked resource(s). Use --force to delete anyway.[/red]"
)
raise typer.Exit(1)
if not yes:
confirm = typer.confirm(f"Delete project '{name}'?")
if not confirm:
raise typer.Abort()
try:
deleted = repo.delete(name)
except DatabaseError as exc:
err_console.print(f"[red]Error deleting project:[/red] {exc.message}")
raise typer.Exit(1) from exc
if not deleted:
err_console.print(f"[red]Project '{name}' could not be deleted.[/red]")
raise typer.Exit(1)
if output_format.lower() == OutputFormat.RICH:
console.print(f"[green]✓[/green] Project '{name}' deleted.")
else:
console.print(format_output({"deleted": name, "success": True}, output_format))
@@ -2271,7 +2271,7 @@ class ProjectNotFoundError(DatabaseError):
class DuplicateLinkError(DatabaseError):
"""Raised when creating a link that duplicates an existing alias in a project."""
"""Raised when creating a duplicate alias within a project."""
def __init__(self, project_name: str, alias: str):
super().__init__(f"Alias '{alias}' already exists in project '{project_name}'")
+11
View File
@@ -56,6 +56,17 @@ DuplicateLinkError # noqa: B018, F821
NamespacedProjectRepository # noqa: B018, F821
ProjectResourceLinkRepository # noqa: B018, F821
# Project CLI helpers — public API
_project_spec_dict # noqa: B018, F821
_store_project_extras # noqa: B018, F821
_get_namespaced_project_repo # noqa: B018, F821
_get_resource_link_repo # noqa: B018, F821
_get_resource_registry_service # noqa: B018, F821
file_filter_app # noqa: B018, F821
list_projects # noqa: B018, F821
link_resource # noqa: B018, F821
unlink_resource # noqa: B018, F821
# Resource registry service — public API and DI wiring
ResourceRegistryService # noqa: B018, F821
_build_resource_registry_service # noqa: B018, F821