Compare commits

...

8 Commits

Author SHA1 Message Date
khyari hamza 659e6c00ac feat(service): add project service v3 (B5-ITEM-5)
- ProjectServicePersistent: create, list_all, list_by_namespace, show, delete
- Resource registration, link/unlink via ProjectResourceLinkRepository
- Validation attachment summary helpers
- Persistence-backed (repositories + session commit)
- 11/11 BDD scenarios passing, 0 regressions (197 total)
- pyright 0 errors, ruff all checks passed
2026-02-13 21:13:40 +00:00
khyari hamza 334e5559a7 feat(service): add resource registry service (B5-ITEM-4)
- ResourceRegistryService: register, remove, show, tree, link_child, auto_discover
- Name/ULID resolution for all operations
- Auto-discovery hook registers git-checkout resources (MVP)
- Wraps ResourceRepository with session-level commit management
- 11/11 BDD scenarios passing, 0 regressions (176 total)
- pyright 0 errors, ruff all checks passed
2026-02-13 21:13:37 +00:00
khyari hamza 13638b1862 feat(repo): add project repositories (B5-ITEM-3)
- Added get_by_namespace() and get_by_namespace_and_name() to ProjectRepository
  for namespace-based filtering and lookup
- Created ProjectResourceLinkRepository with link/unlink/list_links/
  list_links_with_details/get_validation_summary operations
- Duplicate link guard raises ValueError on re-link attempt
- 10/10 BDD scenarios passing, 0 regressions
- pyright 0 errors, ruff all checks passed
2026-02-13 21:13:33 +00:00
khyari hamza ed10d7d208 feat: add scenarios for parsing namespaced project names and name validation 2026-02-13 21:13:01 +00:00
khyari hamza 281936ddd9 refactor: standardize Field defaults in project model classes 2026-02-13 21:12:57 +00:00
khyari hamza 453b310460 refactor: fix linting issues 2026-02-13 21:12:54 +00:00
khyari hamza 8b2ad60c5b feat: Add ValidationConfig and ContextConfig models with step definitions
- Introduced ValidationConfig for project validation commands including test, lint, type-check, and build commands.
- Added ContextConfig for project context indexing and filtering, with support for ignore/include patterns and file size limits.
- Implemented step definitions for both models in project_config_model_steps.py.
- Enhanced Project model to include validation_config and context_config fields.
- Created comprehensive step definitions for Resource model, including validation and manipulation steps in resource_model_steps.py.
- Updated resource.py to define Resource model with necessary fields and validation.
- Refactored project.py to integrate new models and maintain backward compatibility.
2026-02-13 21:12:50 +00:00
khyari hamza 6200c9327a feat: add ResourceType and SandboxStrategy enums with associated tests 2026-02-13 21:12:45 +00:00
19 changed files with 4272 additions and 47 deletions
+130
View File
@@ -0,0 +1,130 @@
Feature: Project Configuration Models
As a developer
I want ValidationConfig and ContextConfig domain models
So that projects can define validation commands and context indexing behavior
# ValidationConfig Tests (B1.5)
# B1.5a - Model creation with defaults
Scenario: Create a default ValidationConfig
Given a default ValidationConfig
Then the validation test_command should be none
And the validation lint_command should be none
And the validation type_check_command should be none
And the validation build_command should be none
And the validation custom_commands should be empty
And the validation timeout_seconds should be 300
And the validation fail_on_lint_error should be true
# B1.5b - Fields with values
Scenario: Create a ValidationConfig with all commands
Given a ValidationConfig with test "pytest" lint "ruff check ." typecheck "pyright" build "python -m build"
Then the validation test_command should be "pytest"
And the validation lint_command should be "ruff check ."
And the validation type_check_command should be "pyright"
And the validation build_command should be "python -m build"
Scenario: Create a ValidationConfig with custom commands
Given a ValidationConfig with custom commands "security=bandit,docs=mkdocs build"
Then the validation custom_commands should contain "security" with value "bandit"
And the validation custom_commands should contain "docs" with value "mkdocs build"
Scenario: Create a ValidationConfig with custom timeout
Given a ValidationConfig with timeout 600
Then the validation timeout_seconds should be 600
Scenario: Create a ValidationConfig with fail_on_lint_error false
Given a ValidationConfig with fail_on_lint_error false
Then the validation fail_on_lint_error should be false
# B1.5c - Helper methods
Scenario: get_all_commands returns all configured commands
Given a ValidationConfig with test "pytest" lint "ruff check ." typecheck "pyright" build "python -m build"
Then get_all_commands should include "test" with value "pytest"
And get_all_commands should include "lint" with value "ruff check ."
And get_all_commands should include "type_check" with value "pyright"
And get_all_commands should include "build" with value "python -m build"
Scenario: get_all_commands includes custom commands
Given a ValidationConfig with test "pytest" and custom commands "security=bandit"
Then get_all_commands should include "test" with value "pytest"
And get_all_commands should include "security" with value "bandit"
Scenario: get_all_commands returns empty dict when no commands configured
Given a default ValidationConfig
Then get_all_commands should be empty
Scenario: has_any_validation is true when commands exist
Given a ValidationConfig with test "pytest" lint "ruff check ." typecheck "pyright" build "python -m build"
Then has_any_validation should be true
Scenario: has_any_validation is false when no commands configured
Given a default ValidationConfig
Then has_any_validation should be false
Scenario: has_any_validation is true with only custom commands
Given a ValidationConfig with custom commands "security=bandit,docs=mkdocs build"
Then has_any_validation should be true
# ContextConfig Tests (B1.6)
# B1.6a - Model creation with defaults
Scenario: Create a default ContextConfig
Given a default ContextConfig
Then the context max_file_size should be 1000000
And the context max_files should be 100000
And the context indexing_strategy should be "full_text"
And the context chunking_policy should be "smart"
And the context chunk_size should be 1000
And the context include_patterns should be none
Scenario: Default ContextConfig has default ignore patterns
Given a default ContextConfig
Then the context ignore_patterns should contain ".git/"
And the context ignore_patterns should contain "node_modules/"
And the context ignore_patterns should contain "__pycache__/"
And the context ignore_patterns should contain ".venv/"
And the context ignore_patterns should contain "*.pyc"
And the context ignore_patterns should contain ".DS_Store"
# B1.6b - Fields with values
Scenario: Create a ContextConfig with custom ignore patterns
Given a ContextConfig with ignore patterns "dist/,build/"
Then the context ignore_patterns should contain "dist/"
And the context ignore_patterns should contain "build/"
And the context ignore_patterns should contain ".git/"
Scenario: Create a ContextConfig with include patterns
Given a ContextConfig with include patterns "*.py,*.ts"
Then the context include_patterns should contain "*.py"
And the context include_patterns should contain "*.ts"
Scenario: Create a ContextConfig with custom max_file_size
Given a ContextConfig with max_file_size 2000000
Then the context max_file_size should be 2000000
Scenario: Create a ContextConfig with custom indexing strategy
Given a ContextConfig with indexing_strategy "semantic"
Then the context indexing_strategy should be "semantic"
Scenario: Create a ContextConfig with custom chunking policy
Given a ContextConfig with chunking_policy "fixed"
Then the context chunking_policy should be "fixed"
Scenario: Create a ContextConfig with custom chunk size
Given a ContextConfig with chunk_size 500
Then the context chunk_size should be 500
# B1.6c - Duplicate ignore patterns are deduplicated
Scenario: ContextConfig deduplicates ignore patterns that overlap with defaults
Given a ContextConfig with ignore patterns ".git/,node_modules/,custom/"
Then the context ignore_patterns should contain ".git/"
And the context ignore_patterns should contain "node_modules/"
And the context ignore_patterns should contain "custom/"
And the context ignore_patterns should not contain duplicate ".git/"
+76
View File
@@ -0,0 +1,76 @@
Feature: Project Repository Enhancements and Link Repository (B5 Item 3)
As a developer
I want namespace-based project filtering and resource link management
So that projects can be organized and linked to resources
Background:
Given a clean in-memory database for project link testing
# ProjectRepository namespace filtering
Scenario: List projects by namespace
Given a saved project "alpha" in namespace "team_one"
And a saved project "beta" in namespace "team_one"
And a saved project "gamma" in namespace "team_two"
When I list projects by namespace "team_one"
Then I should get 2 projects in the list
Scenario: List projects by namespace returns empty for unknown namespace
Given a saved project "solo" in namespace "team_x"
When I list projects by namespace "team_unknown"
Then I should get 0 projects in the list
Scenario: Get project by namespaced name
Given a saved project "myproject" in namespace "team_one"
When I get the project by namespaced name "team_one/myproject"
Then the project should have name "myproject"
And the project should have namespace "team_one"
Scenario: Get project by namespaced name returns None for missing
When I get the project by namespaced name "team_x/nope"
Then the namespaced project result should be None
# ProjectResourceLinkRepository
Scenario: Link a resource to a project
Given a saved project "link-proj" in namespace "local"
And a saved resource "link-res" for linking
When I link the resource to the project with alias "main"
And I list links for the project
Then the link list should have 1 entry
And the first link should have alias "main"
Scenario: Unlink a resource from a project
Given a saved project "unlink-proj" in namespace "local"
And a saved resource "unlink-res" for linking
And the resource is linked to the project with alias "temp"
When I unlink the resource from the project
And I list links for the project
Then the link list should have 0 entries
Scenario: List links returns resource details
Given a saved project "detail-proj" in namespace "local"
And a saved resource "detail-res" for linking
And the resource is linked to the project with alias "docs"
When I list links for the project with details
Then the link detail should include resource name "detail-res"
Scenario: Link with read-only flag
Given a saved project "ro-proj" in namespace "local"
And a saved resource "ro-res" for linking
When I link the resource to the project with alias "readonly" as read-only
And I list links for the project
Then the first link should be read-only
Scenario: Reject duplicate resource link to same project
Given a saved project "dup-proj" in namespace "local"
And a saved resource "dup-res" for linking
And the resource is linked to the project with alias "first"
When I try to link the same resource again with alias "second"
Then a link duplicate error should be raised
Scenario: Get validation attachment summary for linked resources
Given a saved project "val-proj" in namespace "local" with validation config
And a saved resource "val-res" for linking
And the resource is linked to the project with alias "code"
When I get the validation attachment summary for the project
Then the summary should include the project validation config
And the summary should list 1 linked resource
+141
View File
@@ -0,0 +1,141 @@
Feature: Project Domain Model Extensions
As a developer
I want the Project model to support ULID identifiers, namespaces, resources, and configs
So that projects can be properly categorized and managed
# B1.1a - Import Resource, verify Project has new fields
Scenario: Create a Project with new B1.1 fields
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" namespace "local"
Then the project project_id should be "01ARZ3NDEKTSV4RRFFQ69G5FAV"
And the project name should be "my-project"
And the project namespace should be "local"
# B1.1b - Identity fields
Scenario: Project has default namespace
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project namespace should be "local"
Scenario: Project description defaults to none
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project description should be none
Scenario: Create a Project with description
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" and description "A test project"
Then the project description should be "A test project"
# B1.1c - Categorization fields
Scenario: Project tags default to empty list
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project tags should be empty
Scenario: Create a Project with tags
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" and tags "python,web,api"
Then the project tags should contain "python"
And the project tags should contain "web"
And the project tags should contain "api"
Scenario: Project resources default to empty list
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project resources should be empty
Scenario: Project validation_config defaults to none
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project validation_config should be none
Scenario: Project context_config has defaults
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project context_config should not be none
# B1.1d - Timestamp fields
Scenario: Project has timestamps
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project created_at should be set
And the project updated_at should be set
# B1.1e - is_remote computed property
Scenario: Project with no resources is not remote
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
Then the project is_remote should be false
Scenario: Project with all remote resources is remote
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" and all remote resources
Then the project is_remote should be true
Scenario: Project with mixed resources is not remote
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" and mixed resources
Then the project is_remote should be false
# B1.1f - Namespace validator
Scenario: Namespace local is valid
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" namespace "local"
Then the project namespace should be "local"
Scenario: Namespace with valid pattern is accepted
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" namespace "team_alpha"
Then the project namespace should be "team_alpha"
Scenario: Namespace system is rejected
When I try to create a Project with namespace "system"
Then a namespace validation error should be raised
Scenario: Namespace with uppercase is rejected
When I try to create a Project with namespace "MyNamespace"
Then a namespace validation error should be raised
Scenario: Namespace starting with number is rejected
When I try to create a Project with namespace "1invalid"
Then a namespace validation error should be raised
# B1.1g - Helper methods
Scenario: namespaced_name returns correct format
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" namespace "team_alpha"
Then the project namespaced_name should be "team_alpha/my-project"
Scenario: namespaced_name with local namespace
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" namespace "local"
Then the project namespaced_name should be "local/my-project"
Scenario: add_resource adds a resource to the project
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
When I add a resource named "my-repo" to the project
Then the project should have 1 resource
And the project should have a resource named "my-repo"
Scenario: remove_resource removes a resource from the project
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" and a resource named "my-repo"
When I remove the resource named "my-repo" from the project
Then the project resources should be empty
Scenario: get_resource returns the correct resource
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-project" and a resource named "my-repo"
When I get the resource named "my-repo" from the project
Then the retrieved resource name should be "my-repo"
Scenario: get_resource returns none for unknown name
Given a Project with project_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and name "my-project"
When I get the resource named "nonexistent" from the project
Then the retrieved resource should be none
# B1.1h - parse_namespaced_name static method
Scenario: parse_namespaced_name splits valid namespaced string
When I parse the project namespaced name "team_alpha/my-project"
Then the parsed project namespace should be "team_alpha"
And the parsed project name should be "my-project"
Scenario: parse_namespaced_name rejects string without slash
When I try to parse the project namespaced name "no-slash-here"
Then a project namespaced name parse error should be raised
# B1.1i - Name validation edge cases
Scenario: Project rejects name with special characters
When I try to create a Project with invalid name "bad@name!"
Then a name validation error should be raised
@@ -0,0 +1,75 @@
Feature: Project Service v3 with Persistence (B5 Item 5)
As a developer
I want a persistence-backed project service
So that projects can be managed with durable storage and resource linking
Background:
Given a clean database for project service testing
# Create operations
Scenario: Create a project via service
When I create a project named "my-proj" in namespace "local"
Then the created project should have name "my-proj"
And the created project should have namespace "local"
Scenario: Create a project with description
When I create a described project named "desc-proj" in namespace "local" described as "A test project"
Then the created project should have description "A test project"
Scenario: Reject duplicate project name
Given an existing project "dup-proj" in namespace "local"
When I try to create a project named "dup-proj" in namespace "local"
Then a project already exists error should be raised
# List operations
Scenario: List all projects
Given an existing project "alpha-proj" in namespace "team_a"
And an existing project "beta-proj" in namespace "team_b"
When I list all projects via service
Then the service project list should have 2 entries
Scenario: List projects filtered by namespace
Given an existing project "ns-proj-a" in namespace "team_x"
And an existing project "ns-proj-b" in namespace "team_x"
And an existing project "ns-proj-c" in namespace "team_y"
When I list projects by namespace "team_x" via service
Then the service project list should have 2 entries
# Show operations
Scenario: Show a project by name
Given an existing project "show-proj" in namespace "local"
When I show the project "show-proj" via service
Then the shown project should have name "show-proj"
Scenario: Show non-existent project raises error
When I try to show the project "nope" via service
Then a persistent project not found error should be raised
# Delete operations
Scenario: Delete a project
Given an existing project "del-proj" in namespace "local"
When I delete the project "del-proj" via service
Then showing project "del-proj" via service should raise not found
# Link/Unlink operations
Scenario: Link a resource to a project via service
Given an existing project "link-proj" in namespace "local"
And a registered service resource "link-svc-res" of type "git_repository" at "/repos/main"
When I link resource "link-svc-res" to project "link-proj" with alias "main"
Then the project "link-proj" should have 1 linked resource
Scenario: Unlink a resource from a project via service
Given an existing project "unlink-proj" in namespace "local"
And a registered service resource "unlink-svc-res" of type "filesystem" at "/data"
And resource "unlink-svc-res" is linked to project "unlink-proj" with alias "data"
When I unlink resource "unlink-svc-res" from project "unlink-proj"
Then the project "unlink-proj" should have 0 linked resources
# Validation attachment helpers
Scenario: Get validation summary for a project via service
Given an existing project "val-svc-proj" in namespace "local" with validation
And a registered service resource "val-svc-res" of type "git_repository" at "/repos/code"
And resource "val-svc-res" is linked to project "val-svc-proj" with alias "code"
When I get the validation summary for project "val-svc-proj" via service
Then the service summary should include validation config
And the service summary should list 1 linked resource
+204
View File
@@ -0,0 +1,204 @@
Feature: Resource Domain Model
As a developer
I want resource types and sandbox strategies defined as domain enums
So that the system can classify project resources and determine sandboxing behavior
# ResourceType Enum Tests (B1.3)
Scenario: ResourceType has all required values
Then the resource types should be "git_repository, filesystem, database, api_endpoint, document_corpus, cloud_infrastructure"
Scenario: ResourceType GIT_REPOSITORY has correct value
When I access ResourceType.GIT_REPOSITORY
Then the resource type value should be "git_repository"
Scenario: ResourceType FILESYSTEM has correct value
When I access ResourceType.FILESYSTEM
Then the resource type value should be "filesystem"
Scenario: ResourceType DATABASE has correct value
When I access ResourceType.DATABASE
Then the resource type value should be "database"
Scenario: ResourceType API_ENDPOINT has correct value
When I access ResourceType.API_ENDPOINT
Then the resource type value should be "api_endpoint"
Scenario: ResourceType DOCUMENT_CORPUS has correct value
When I access ResourceType.DOCUMENT_CORPUS
Then the resource type value should be "document_corpus"
Scenario: ResourceType CLOUD_INFRASTRUCTURE has correct value
When I access ResourceType.CLOUD_INFRASTRUCTURE
Then the resource type value should be "cloud_infrastructure"
Scenario: ResourceType is a string enum
When I access ResourceType.GIT_REPOSITORY
Then the resource type should be a string
Scenario: ResourceType can be created from string value
When I create a ResourceType from string "filesystem"
Then the resource type value should be "filesystem"
Scenario: Invalid resource type string raises error
When I try to create a ResourceType from string "invalid_type"
Then a ValueError should be raised
# SandboxStrategy Enum Tests (B1.4)
Scenario: SandboxStrategy has all required values
Then the sandbox strategies should be "git_worktree, copy_on_write, overlay, transaction_rollback, versioning, none"
Scenario: SandboxStrategy GIT_WORKTREE has correct value
When I access SandboxStrategy.GIT_WORKTREE
Then the sandbox strategy value should be "git_worktree"
Scenario: SandboxStrategy COPY_ON_WRITE has correct value
When I access SandboxStrategy.COPY_ON_WRITE
Then the sandbox strategy value should be "copy_on_write"
Scenario: SandboxStrategy OVERLAY has correct value
When I access SandboxStrategy.OVERLAY
Then the sandbox strategy value should be "overlay"
Scenario: SandboxStrategy TRANSACTION_ROLLBACK has correct value
When I access SandboxStrategy.TRANSACTION_ROLLBACK
Then the sandbox strategy value should be "transaction_rollback"
Scenario: SandboxStrategy VERSIONING has correct value
When I access SandboxStrategy.VERSIONING
Then the sandbox strategy value should be "versioning"
Scenario: SandboxStrategy NONE has correct value
When I access SandboxStrategy.NONE
Then the sandbox strategy value should be "none"
Scenario: SandboxStrategy is a string enum
When I access SandboxStrategy.GIT_WORKTREE
Then the sandbox strategy should be a string
# SandboxStrategy.supports_rollback Tests (B1.4b)
Scenario: GIT_WORKTREE supports rollback
Then SandboxStrategy.GIT_WORKTREE should support rollback
Scenario: COPY_ON_WRITE supports rollback
Then SandboxStrategy.COPY_ON_WRITE should support rollback
Scenario: OVERLAY supports rollback
Then SandboxStrategy.OVERLAY should support rollback
Scenario: TRANSACTION_ROLLBACK supports rollback
Then SandboxStrategy.TRANSACTION_ROLLBACK should support rollback
Scenario: VERSIONING supports rollback
Then SandboxStrategy.VERSIONING should support rollback
Scenario: NONE does not support rollback
Then SandboxStrategy.NONE should not support rollback
# SandboxStrategy.is_copy_based Tests (B1.4b)
Scenario: COPY_ON_WRITE is copy based
Then SandboxStrategy.COPY_ON_WRITE should be copy based
Scenario: OVERLAY is copy based
Then SandboxStrategy.OVERLAY should be copy based
Scenario: GIT_WORKTREE is not copy based
Then SandboxStrategy.GIT_WORKTREE should not be copy based
Scenario: TRANSACTION_ROLLBACK is not copy based
Then SandboxStrategy.TRANSACTION_ROLLBACK should not be copy based
Scenario: VERSIONING is not copy based
Then SandboxStrategy.VERSIONING should not be copy based
Scenario: NONE is not copy based
Then SandboxStrategy.NONE should not be copy based
# Resource Pydantic Model Tests (B1.2)
# B1.2a - Model structure (frozen=True)
Scenario: Create a valid Resource with all required fields
Given a Resource with required fields id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo"
Then the resource name should be "my-repo"
And the resource type should be ResourceType.GIT_REPOSITORY
And the resource location should be "/tmp/repo"
Scenario: Resource model is frozen
Given I have a valid Resource
When I try to modify the resource name
Then a validation error should be raised
# B1.2b - All fields present with defaults
Scenario: Resource has correct default values
Given a Resource with required fields id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo"
Then the resource is_remote should be false
And the resource sandbox_strategy should be SandboxStrategy.NONE
And the resource read_only should be false
And the resource metadata should be empty
And the resource created_at should be set
Scenario: Create a remote Resource
Given a remote Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "api-svc" type "api_endpoint" location "https://api.example.com"
Then the resource is_remote should be true
Scenario: Create a read-only Resource
Given a read-only Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "docs" type "document_corpus" location "/data/docs"
Then the resource read_only should be true
Scenario: Create a Resource with custom sandbox strategy
Given a Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo" strategy "git_worktree"
Then the resource sandbox_strategy should be SandboxStrategy.GIT_WORKTREE
Scenario: Create a Resource with metadata
Given a Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo" metadata "branch=main,remote=origin"
Then the resource metadata should contain key "branch" with value "main"
And the resource metadata should contain key "remote" with value "origin"
# B1.2c - Validators
Scenario: Reject Resource with invalid ULID
When I try to create a Resource with id "not-a-ulid" name "my-repo" type "git_repository" location "/tmp/repo"
Then a validation error should be raised
Scenario: Reject Resource with empty name
When I try to create a Resource with an empty name
Then a validation error should be raised
Scenario: Reject Resource with empty location
When I try to create a Resource with an empty location
Then a validation error should be raised
Scenario: Resource name is lowercased
Given a Resource with required fields id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "My-Repo" type "git_repository" location "/tmp/repo"
Then the resource name should be "my-repo"
Scenario: Reject Resource with invalid name characters
When I try to create a Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my repo!!" type "git_repository" location "/tmp/repo"
Then a validation error should be raised
# B1.2d - Properties
Scenario: Resource with sandbox strategy supports sandbox
Given a Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo" strategy "git_worktree"
Then the resource supports_sandbox should be true
Scenario: Resource with NONE sandbox strategy does not support sandbox
Given a Resource with required fields id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo"
Then the resource supports_sandbox should be false
Scenario: Writable Resource can_write is true
Given a Resource with required fields id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo"
Then the resource can_write should be true
Scenario: Read-only Resource can_write is false
Given a read-only Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "docs" type "document_corpus" location "/data/docs"
Then the resource can_write should be false
Scenario: Resource get_sandbox_path returns correct path
Given a Resource with id "01ARZ3NDEKTSV4RRFFQ69G5FAV" name "my-repo" type "git_repository" location "/tmp/repo" strategy "git_worktree"
Then the resource get_sandbox_path with base "/sandbox" should end with "my-repo"
@@ -0,0 +1,68 @@
Feature: Resource Registry Service (B5 Item 4)
As a developer
I want a service layer for resource registration and management
So that resources can be registered, removed, inspected and traversed via a clean API
Background:
Given a clean database for resource registry testing
# Register operations
Scenario: Register a new resource
When I register a resource named "my-repo" of type "git_repository" at "/tmp/repo"
Then the registration result should have name "my-repo"
And the registration result should have a valid resource ID
Scenario: Register a resource with sandbox strategy
When I register a sandboxed resource named "sandbox-repo" of type "git_repository" at "/tmp/repo" with strategy "git_worktree"
Then the registration result should have sandbox strategy "git_worktree"
Scenario: Register a read-only resource
When I register a read-only resource named "docs" of type "document_corpus" at "/data/docs"
Then the registration result should be read-only
Scenario: Reject duplicate resource name
Given a registered resource named "dup-res" of type "filesystem" at "/data"
When I try to register a resource named "dup-res" of type "filesystem" at "/data2"
Then a duplicate resource error should be raised
# Remove operations
Scenario: Remove a registered resource
Given a registered resource named "temp-res" of type "filesystem" at "/tmp/data"
When I remove the resource named "temp-res"
Then showing resource "temp-res" should raise not found
Scenario: Remove non-existent resource raises error
When I try to remove the resource named "ghost"
Then a resource not found error should be raised
# Show operations
Scenario: Show a resource by name
Given a registered resource named "show-me" of type "api_endpoint" at "https://api.example.com"
When I show the resource named "show-me"
Then the shown resource should have type "api_endpoint"
And the shown resource should have location "https://api.example.com"
Scenario: Show a resource by ULID
Given a registered resource named "by-id-res" of type "database" at "postgres://localhost/db"
When I show the resource by its ULID
Then the shown resource should have name "by-id-res"
# Tree operations
Scenario: Tree with parent-child relationship
Given a registered resource named "parent-res" of type "git_repository" at "/repos/parent"
And a registered resource named "child-res" of type "filesystem" at "/repos/child"
And resource "child-res" is a child of "parent-res"
When I get the resource tree for "parent-res"
Then the tree should contain 2 resources
And the tree should include resource "child-res"
Scenario: Tree of single resource with no children
Given a registered resource named "lonely" of type "filesystem" at "/data/lonely"
When I get the resource tree for "lonely"
Then the tree should contain 1 resources
# Auto-discovery hook (git-checkout MVP)
Scenario: Auto-discover registers a git-checkout resource
When I auto-discover a resource at "/tmp/my-git-project" with name "auto-git"
Then the registration result should have name "auto-git"
And the registration result should have type "git_repository"
@@ -0,0 +1,313 @@
"""Step definitions for project configuration model tests (B1.5, B1.6)."""
from __future__ import annotations
from behave import given, then
from behave.runner import Context
from cleveragents.domain.models.core.project import ContextConfig, ValidationConfig
# ValidationConfig Steps (B1.5)
@given("a default ValidationConfig")
def step_default_validation_config(context: Context) -> None:
"""Create a ValidationConfig with all defaults."""
context.validation_config = ValidationConfig()
@given(
'a ValidationConfig with test "{test}" lint "{lint}" typecheck "{typecheck}" build "{build}"'
)
def step_validation_config_all_commands(
context: Context, test: str, lint: str, typecheck: str, build: str
) -> None:
"""Create a ValidationConfig with all standard commands."""
context.validation_config = ValidationConfig(
test_command=test,
lint_command=lint,
type_check_command=typecheck,
build_command=build,
)
@given('a ValidationConfig with custom commands "{commands_str}"')
def step_validation_config_custom_commands(context: Context, commands_str: str) -> None:
"""Create a ValidationConfig with custom commands parsed from key=value pairs."""
custom = {}
for pair in commands_str.split(","):
key, value = pair.strip().split("=", 1)
custom[key.strip()] = value.strip()
context.validation_config = ValidationConfig(custom_commands=custom)
@given("a ValidationConfig with timeout {timeout:d}")
def step_validation_config_timeout(context: Context, timeout: int) -> None:
"""Create a ValidationConfig with custom timeout."""
context.validation_config = ValidationConfig(timeout_seconds=timeout)
@given("a ValidationConfig with fail_on_lint_error false")
def step_validation_config_no_lint_fail(context: Context) -> None:
"""Create a ValidationConfig with fail_on_lint_error disabled."""
context.validation_config = ValidationConfig(fail_on_lint_error=False)
@given('a ValidationConfig with test "{test}" and custom commands "{commands_str}"')
def step_validation_config_test_and_custom(
context: Context, test: str, commands_str: str
) -> None:
"""Create a ValidationConfig with test command and custom commands."""
custom = {}
for pair in commands_str.split(","):
key, value = pair.strip().split("=", 1)
custom[key.strip()] = value.strip()
context.validation_config = ValidationConfig(
test_command=test, custom_commands=custom
)
# ValidationConfig field assertions
@then("the validation test_command should be none")
def step_check_test_command_none(context: Context) -> None:
"""Verify test_command is None."""
assert context.validation_config.test_command is None
@then('the validation test_command should be "{expected}"')
def step_check_test_command(context: Context, expected: str) -> None:
"""Verify test_command value."""
assert context.validation_config.test_command == expected, (
f"Expected '{expected}', got '{context.validation_config.test_command}'"
)
@then("the validation lint_command should be none")
def step_check_lint_command_none(context: Context) -> None:
"""Verify lint_command is None."""
assert context.validation_config.lint_command is None
@then('the validation lint_command should be "{expected}"')
def step_check_lint_command(context: Context, expected: str) -> None:
"""Verify lint_command value."""
assert context.validation_config.lint_command == expected, (
f"Expected '{expected}', got '{context.validation_config.lint_command}'"
)
@then("the validation type_check_command should be none")
def step_check_typecheck_command_none(context: Context) -> None:
"""Verify type_check_command is None."""
assert context.validation_config.type_check_command is None
@then('the validation type_check_command should be "{expected}"')
def step_check_typecheck_command(context: Context, expected: str) -> None:
"""Verify type_check_command value."""
assert context.validation_config.type_check_command == expected, (
f"Expected '{expected}', got '{context.validation_config.type_check_command}'"
)
@then("the validation build_command should be none")
def step_check_build_command_none(context: Context) -> None:
"""Verify build_command is None."""
assert context.validation_config.build_command is None
@then('the validation build_command should be "{expected}"')
def step_check_build_command(context: Context, expected: str) -> None:
"""Verify build_command value."""
assert context.validation_config.build_command == expected, (
f"Expected '{expected}', got '{context.validation_config.build_command}'"
)
@then("the validation custom_commands should be empty")
def step_check_custom_commands_empty(context: Context) -> None:
"""Verify custom_commands is empty dict."""
assert context.validation_config.custom_commands == {}
@then('the validation custom_commands should contain "{key}" with value "{value}"')
def step_check_custom_command(context: Context, key: str, value: str) -> None:
"""Verify custom_commands contains expected key-value pair."""
assert key in context.validation_config.custom_commands, (
f"Expected key '{key}' in custom_commands"
)
assert context.validation_config.custom_commands[key] == value, (
f"Expected custom_commands['{key}'] = '{value}', "
f"got '{context.validation_config.custom_commands[key]}'"
)
@then("the validation timeout_seconds should be {expected:d}")
def step_check_timeout(context: Context, expected: int) -> None:
"""Verify timeout_seconds value."""
assert context.validation_config.timeout_seconds == expected
@then("the validation fail_on_lint_error should be true")
def step_check_fail_on_lint_true(context: Context) -> None:
"""Verify fail_on_lint_error is True."""
assert context.validation_config.fail_on_lint_error is True
@then("the validation fail_on_lint_error should be false")
def step_check_fail_on_lint_false(context: Context) -> None:
"""Verify fail_on_lint_error is False."""
assert context.validation_config.fail_on_lint_error is False
# ValidationConfig method assertions
@then('get_all_commands should include "{key}" with value "{value}"')
def step_check_get_all_commands_key(context: Context, key: str, value: str) -> None:
"""Verify get_all_commands includes expected key-value pair."""
commands = context.validation_config.get_all_commands()
assert key in commands, (
f"Expected key '{key}' in get_all_commands(), got {commands}"
)
assert commands[key] == value, (
f"Expected commands['{key}'] = '{value}', got '{commands[key]}'"
)
@then("get_all_commands should be empty")
def step_check_get_all_commands_empty(context: Context) -> None:
"""Verify get_all_commands returns empty dict."""
commands = context.validation_config.get_all_commands()
assert commands == {}, f"Expected empty dict, got {commands}"
@then("has_any_validation should be true")
def step_check_has_validation_true(context: Context) -> None:
"""Verify has_any_validation returns True."""
assert context.validation_config.has_any_validation() is True
@then("has_any_validation should be false")
def step_check_has_validation_false(context: Context) -> None:
"""Verify has_any_validation returns False."""
assert context.validation_config.has_any_validation() is False
# ContextConfig Steps (B1.6)
@given("a default ContextConfig")
def step_default_context_config(context: Context) -> None:
"""Create a ContextConfig with all defaults."""
context.context_config = ContextConfig()
@given('a ContextConfig with ignore patterns "{patterns}"')
def step_context_config_ignore_patterns(context: Context, patterns: str) -> None:
"""Create a ContextConfig with custom ignore patterns."""
pattern_list = [p.strip() for p in patterns.split(",")]
context.context_config = ContextConfig(ignore_patterns=pattern_list)
@given('a ContextConfig with include patterns "{patterns}"')
def step_context_config_include_patterns(context: Context, patterns: str) -> None:
"""Create a ContextConfig with include patterns."""
pattern_list = [p.strip() for p in patterns.split(",")]
context.context_config = ContextConfig(include_patterns=pattern_list)
@given("a ContextConfig with max_file_size {size:d}")
def step_context_config_max_file_size(context: Context, size: int) -> None:
"""Create a ContextConfig with custom max_file_size."""
context.context_config = ContextConfig(max_file_size=size)
@given('a ContextConfig with indexing_strategy "{strategy}"')
def step_context_config_indexing_strategy(context: Context, strategy: str) -> None:
"""Create a ContextConfig with custom indexing strategy."""
context.context_config = ContextConfig(indexing_strategy=strategy)
@given('a ContextConfig with chunking_policy "{policy}"')
def step_context_config_chunking_policy(context: Context, policy: str) -> None:
"""Create a ContextConfig with custom chunking policy."""
context.context_config = ContextConfig(chunking_policy=policy)
@given("a ContextConfig with chunk_size {size:d}")
def step_context_config_chunk_size(context: Context, size: int) -> None:
"""Create a ContextConfig with custom chunk size."""
context.context_config = ContextConfig(chunk_size=size)
# ContextConfig field assertions
@then("the context max_file_size should be {expected:d}")
def step_check_max_file_size(context: Context, expected: int) -> None:
"""Verify max_file_size value."""
assert context.context_config.max_file_size == expected
@then("the context max_files should be {expected:d}")
def step_check_max_files(context: Context, expected: int) -> None:
"""Verify max_files value."""
assert context.context_config.max_files == expected
@then('the context indexing_strategy should be "{expected}"')
def step_check_indexing_strategy(context: Context, expected: str) -> None:
"""Verify indexing_strategy value."""
assert context.context_config.indexing_strategy == expected
@then('the context chunking_policy should be "{expected}"')
def step_check_chunking_policy(context: Context, expected: str) -> None:
"""Verify chunking_policy value."""
assert context.context_config.chunking_policy == expected
@then("the context chunk_size should be {expected:d}")
def step_check_chunk_size(context: Context, expected: int) -> None:
"""Verify chunk_size value."""
assert context.context_config.chunk_size == expected
@then("the context include_patterns should be none")
def step_check_include_patterns_none(context: Context) -> None:
"""Verify include_patterns is None."""
assert context.context_config.include_patterns is None
@then('the context ignore_patterns should contain "{pattern}"')
def step_check_ignore_pattern(context: Context, pattern: str) -> None:
"""Verify ignore_patterns contains expected pattern."""
assert pattern in context.context_config.ignore_patterns, (
f"Expected '{pattern}' in ignore_patterns, "
f"got {context.context_config.ignore_patterns}"
)
@then('the context include_patterns should contain "{pattern}"')
def step_check_include_pattern(context: Context, pattern: str) -> None:
"""Verify include_patterns contains expected pattern."""
assert context.context_config.include_patterns is not None, (
"include_patterns is None"
)
assert pattern in context.context_config.include_patterns, (
f"Expected '{pattern}' in include_patterns, "
f"got {context.context_config.include_patterns}"
)
@then('the context ignore_patterns should not contain duplicate "{pattern}"')
def step_check_no_duplicate_ignore_pattern(context: Context, pattern: str) -> None:
"""Verify ignore_patterns does not contain duplicates of the given pattern."""
count = context.context_config.ignore_patterns.count(pattern)
assert count == 1, (
f"Expected exactly 1 occurrence of '{pattern}' in ignore_patterns, "
f"found {count}: {context.context_config.ignore_patterns}"
)
@@ -0,0 +1,354 @@
"""Step definitions for Project Repository Enhancements and Link Repository."""
from datetime import datetime
from pathlib import Path
from typing import Any
from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.domain.models.core.project import (
ContextConfig,
Project,
ProjectSettings,
ValidationConfig,
)
from cleveragents.domain.models.core.resource import Resource, ResourceType
from cleveragents.infrastructure.database.models import (
ProjectModel,
ProjectResourceLinkModel,
)
from cleveragents.infrastructure.database.repositories import ProjectRepository
from cleveragents.infrastructure.database.resource_repository import (
ResourceRepository,
)
_ULID_POOL = [
"01ARZ3NDEKTSV4RRFFQ69G5FAV",
"01HN8Y1FMCEZ7V3KNRS9PGEDKW",
"01BX5ZZKBKACTAV9WEVGEMMVS0",
"01C3G8XQNB5VTFNHC2HQ7P9QFP",
"01D7BNHV7E3T1QZ4KT8S6XFWPV",
]
_ULID_IDX = 0
def _next_ulid() -> str:
"""Get next test ULID."""
global _ULID_IDX
ulid = _ULID_POOL[_ULID_IDX % len(_ULID_POOL)]
_ULID_IDX += 1
return ulid
def _make_project(name: str, namespace: str = "local", **kwargs: Any) -> Project:
"""Create a Project domain object for testing."""
return Project(
id=None,
name=name,
path=Path("/tmp/test"),
created_at=datetime.now(),
updated_at=datetime.now(),
current_plan_id=None,
settings=ProjectSettings(
auto_build=False,
auto_apply=False,
confirm_apply=True,
max_context_size=52428800,
default_model="mock-gpt",
),
namespace=namespace,
**kwargs,
)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a clean in-memory database for project link testing")
def step_clean_db(context: Context) -> None:
"""Set up a clean in-memory database with migrations."""
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
from cleveragents.infrastructure.database.migration_runner import MigrationRunner
global _ULID_IDX
_ULID_IDX = 0
database_url = "sqlite:///:memory:"
engine = create_engine(database_url, connect_args={"check_same_thread": False})
MEMORY_ENGINES[database_url] = engine
migration_runner = MigrationRunner(database_url)
migration_runner.run_migrations(engine=engine)
context.engine = engine
context.session_factory = sessionmaker(
bind=engine, autoflush=False, autocommit=False
)
context.db_session = context.session_factory()
context.project_repo = ProjectRepository(context.db_session)
context.resource_repo = ResourceRepository(context.db_session)
# Import link repository
from cleveragents.infrastructure.database.project_link_repository import (
ProjectResourceLinkRepository,
)
context.link_repo = ProjectResourceLinkRepository(context.db_session)
# ---------------------------------------------------------------------------
# Project namespace steps
# ---------------------------------------------------------------------------
@given('a saved project "{name}" in namespace "{namespace}"')
def step_saved_project_ns(context: Context, name: str, namespace: str) -> None:
"""Create and save a project with a specific namespace."""
project = _make_project(name=name, namespace=namespace)
project = context.project_repo.create(project)
context.db_session.commit()
context.project = project
@given('a saved project "{name}" in namespace "{namespace}" with validation config')
def step_saved_project_ns_vc(context: Context, name: str, namespace: str) -> None:
"""Create and save a project with namespace and validation config."""
project = _make_project(
name=name,
namespace=namespace,
validation_config=ValidationConfig(
test_command="pytest",
lint_command="ruff check .",
),
)
project = context.project_repo.create(project)
context.db_session.commit()
context.project = project
@when('I list projects by namespace "{namespace}"')
def step_list_by_namespace(context: Context, namespace: str) -> None:
"""List projects filtered by namespace."""
context.project_list = context.project_repo.get_by_namespace(namespace)
@then("I should get {count:d} projects in the list")
def step_check_project_count(context: Context, count: int) -> None:
"""Verify the number of projects returned."""
assert len(context.project_list) == count, (
f"Expected {count} projects, got {len(context.project_list)}"
)
@when('I get the project by namespaced name "{namespaced_name}"')
def step_get_by_namespaced(context: Context, namespaced_name: str) -> None:
"""Get a project by its namespaced name."""
namespace, name = namespaced_name.split("/", 1)
context.namespaced_result = context.project_repo.get_by_namespace_and_name(
namespace, name
)
@then('the project should have name "{name}"')
def step_check_project_name(context: Context, name: str) -> None:
"""Verify the project name."""
assert context.namespaced_result is not None, "No project found"
assert context.namespaced_result.name == name, (
f"Expected '{name}', got '{context.namespaced_result.name}'"
)
@then('the project should have namespace "{namespace}"')
def step_check_project_namespace(context: Context, namespace: str) -> None:
"""Verify the project namespace."""
assert context.namespaced_result is not None, "No project found"
assert context.namespaced_result.namespace == namespace, (
f"Expected '{namespace}', got '{context.namespaced_result.namespace}'"
)
@then("the namespaced project result should be None")
def step_check_namespaced_none(context: Context) -> None:
"""Verify no project was found."""
assert context.namespaced_result is None, (
f"Expected None, got {context.namespaced_result}"
)
# ---------------------------------------------------------------------------
# Resource link steps
# ---------------------------------------------------------------------------
@given('a saved resource "{name}" for linking')
def step_saved_resource_for_link(context: Context, name: str) -> None:
"""Create and save a resource for linking."""
resource = Resource(
resource_id=_next_ulid(),
name=name,
type=ResourceType.GIT_REPOSITORY,
location="/tmp/repo",
)
resource = context.resource_repo.create(resource)
context.db_session.commit()
context.link_resource = resource
@when('I link the resource to the project with alias "{alias}"')
def step_link_resource(context: Context, alias: str) -> None:
"""Link a resource to the project."""
context.link_repo.link(
project_id=context.project.id,
resource_id=context.link_resource.resource_id,
alias=alias,
)
context.db_session.commit()
@when('I link the resource to the project with alias "{alias}" as read-only')
def step_link_resource_ro(context: Context, alias: str) -> None:
"""Link a resource to the project as read-only."""
context.link_repo.link(
project_id=context.project.id,
resource_id=context.link_resource.resource_id,
alias=alias,
read_only=True,
)
context.db_session.commit()
@given('the resource is linked to the project with alias "{alias}"')
def step_linked_resource(context: Context, alias: str) -> None:
"""Link a resource (given variant)."""
context.link_repo.link(
project_id=context.project.id,
resource_id=context.link_resource.resource_id,
alias=alias,
)
context.db_session.commit()
@when("I unlink the resource from the project")
def step_unlink_resource(context: Context) -> None:
"""Unlink a resource from the project."""
context.link_repo.unlink(
project_id=context.project.id,
resource_id=context.link_resource.resource_id,
)
context.db_session.commit()
@when("I list links for the project")
def step_list_links(context: Context) -> None:
"""List all resource links for the project."""
context.link_list = context.link_repo.list_links(context.project.id)
@when("I list links for the project with details")
def step_list_links_detail(context: Context) -> None:
"""List links with resource details."""
context.link_details = context.link_repo.list_links_with_details(context.project.id)
@then("the link list should have {count:d} entry")
def step_check_link_count_singular(context: Context, count: int) -> None:
"""Verify link count (singular)."""
assert len(context.link_list) == count, (
f"Expected {count} links, got {len(context.link_list)}"
)
@then("the link list should have {count:d} entries")
def step_check_link_count_plural(context: Context, count: int) -> None:
"""Verify link count (plural)."""
assert len(context.link_list) == count, (
f"Expected {count} links, got {len(context.link_list)}"
)
@then('the first link should have alias "{alias}"')
def step_check_first_link_alias(context: Context, alias: str) -> None:
"""Verify the first link's alias."""
assert len(context.link_list) > 0, "No links found"
assert context.link_list[0]["alias"] == alias, (
f"Expected alias '{alias}', got '{context.link_list[0]['alias']}'"
)
@then("the first link should be read-only")
def step_check_first_link_ro(context: Context) -> None:
"""Verify the first link is read-only."""
assert len(context.link_list) > 0, "No links found"
assert context.link_list[0]["read_only"] is True, "Link is not read-only"
@then('the link detail should include resource name "{name}"')
def step_check_link_detail_name(context: Context, name: str) -> None:
"""Verify link detail includes resource name."""
assert len(context.link_details) > 0, "No link details"
assert any(d["resource_name"] == name for d in context.link_details), (
f"Resource name '{name}' not found in details: {context.link_details}"
)
@when('I try to link the same resource again with alias "{alias}"')
def step_try_duplicate_link(context: Context, alias: str) -> None:
"""Try to create a duplicate link."""
context.error = None
try:
context.link_repo.link(
project_id=context.project.id,
resource_id=context.link_resource.resource_id,
alias=alias,
)
context.db_session.commit()
except ValueError as e:
context.error = e
context.db_session.rollback()
@then("a link duplicate error should be raised")
def step_check_link_dup_error(context: Context) -> None:
"""Verify a duplicate link error was raised."""
assert context.error is not None, "Expected duplicate link error"
assert (
"already linked" in str(context.error).lower()
or "duplicate" in str(context.error).lower()
), f"Expected duplicate error, got: {context.error}"
# ---------------------------------------------------------------------------
# Validation attachment summary
# ---------------------------------------------------------------------------
@when("I get the validation attachment summary for the project")
def step_get_validation_summary(context: Context) -> None:
"""Get the validation attachment summary."""
context.validation_summary = context.link_repo.get_validation_summary(
context.project.id
)
@then("the summary should include the project validation config")
def step_check_summary_config(context: Context) -> None:
"""Verify summary includes validation config."""
assert context.validation_summary is not None, "No summary"
assert context.validation_summary["validation_config"] is not None, (
"Validation config missing from summary"
)
assert context.validation_summary["validation_config"]["test_command"] == "pytest"
@then("the summary should list {count:d} linked resource")
def step_check_summary_resources(context: Context, count: int) -> None:
"""Verify number of linked resources in summary."""
assert len(context.validation_summary["linked_resources"]) == count, (
f"Expected {count} resources, "
f"got {len(context.validation_summary['linked_resources'])}"
)
+361
View File
@@ -0,0 +1,361 @@
"""Step definitions for Project domain model extension tests (B1.1)."""
from __future__ import annotations
from pathlib import Path
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.project import Project
from cleveragents.domain.models.core.resource import (
Resource,
ResourceType,
)
VALID_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
VALID_ULID_2 = "01ARZ3NDEKTSV4RRFFQ69G5FAW"
VALID_ULID_3 = "01ARZ3NDEKTSV4RRFFQ69G5FAX"
def _make_resource(
name: str,
is_remote: bool = False,
resource_id: str = VALID_ULID_2,
) -> Resource:
"""Helper to create a test Resource."""
return Resource(
resource_id=resource_id,
name=name,
type=ResourceType.GIT_REPOSITORY,
location="/tmp/repo" if not is_remote else "https://github.com/test",
is_remote=is_remote,
)
# Project creation steps
@given('a Project with project_id "{pid}" name "{name}" namespace "{namespace}"')
def step_create_project_with_namespace(
context: Context, pid: str, name: str, namespace: str
) -> None:
"""Create a Project with explicit namespace."""
context.project = Project(
project_id=pid,
name=name,
namespace=namespace,
path=Path("/tmp/test"),
)
@given('a Project with project_id "{pid}" and name "{name}"')
def step_create_project_minimal(context: Context, pid: str, name: str) -> None:
"""Create a Project with minimal fields."""
context.project = Project(
project_id=pid,
name=name,
path=Path("/tmp/test"),
)
@given('a Project with project_id "{pid}" name "{name}" and description "{desc}"')
def step_create_project_with_description(
context: Context, pid: str, name: str, desc: str
) -> None:
"""Create a Project with description."""
context.project = Project(
project_id=pid,
name=name,
description=desc,
path=Path("/tmp/test"),
)
@given('a Project with project_id "{pid}" name "{name}" and tags "{tags_str}"')
def step_create_project_with_tags(
context: Context, pid: str, name: str, tags_str: str
) -> None:
"""Create a Project with tags."""
tags = [t.strip() for t in tags_str.split(",")]
context.project = Project(
project_id=pid,
name=name,
tags=tags,
path=Path("/tmp/test"),
)
@given('a Project with project_id "{pid}" name "{name}" and all remote resources')
def step_create_project_all_remote(context: Context, pid: str, name: str) -> None:
"""Create a Project where all resources are remote."""
resources = [
_make_resource("api-svc", is_remote=True, resource_id=VALID_ULID_2),
_make_resource("cloud-db", is_remote=True, resource_id=VALID_ULID_3),
]
context.project = Project(
project_id=pid,
name=name,
resources=resources,
path=Path("/tmp/test"),
)
@given('a Project with project_id "{pid}" name "{name}" and mixed resources')
def step_create_project_mixed_resources(context: Context, pid: str, name: str) -> None:
"""Create a Project with both local and remote resources."""
resources = [
_make_resource("local-repo", is_remote=False, resource_id=VALID_ULID_2),
_make_resource("api-svc", is_remote=True, resource_id=VALID_ULID_3),
]
context.project = Project(
project_id=pid,
name=name,
resources=resources,
path=Path("/tmp/test"),
)
@given('a Project with project_id "{pid}" name "{name}" and a resource named "{rname}"')
def step_create_project_with_resource(
context: Context, pid: str, name: str, rname: str
) -> None:
"""Create a Project with one resource."""
resource = _make_resource(rname)
context.project = Project(
project_id=pid,
name=name,
resources=[resource],
path=Path("/tmp/test"),
)
# Namespace validation steps
@when('I try to create a Project with namespace "{namespace}"')
def step_try_create_project_bad_namespace(context: Context, namespace: str) -> None:
"""Attempt to create a Project with an invalid namespace."""
context.error = None
try:
context.project = Project(
project_id=VALID_ULID,
name="test-project",
namespace=namespace,
path=Path("/tmp/test"),
)
except (ValidationError, ValueError) as e:
context.error = e
@then("a namespace validation error should be raised")
def step_check_namespace_error(context: Context) -> None:
"""Verify a namespace validation error was raised."""
assert context.error is not None, (
"Expected a namespace validation error but none was raised"
)
# Resource manipulation steps
@when('I add a resource named "{rname}" to the project')
def step_add_resource(context: Context, rname: str) -> None:
"""Add a resource to the project."""
resource = _make_resource(rname)
context.project = context.project.add_resource(resource)
@when('I remove the resource named "{rname}" from the project')
def step_remove_resource(context: Context, rname: str) -> None:
"""Remove a resource from the project."""
context.project = context.project.remove_resource(rname)
@when('I get the resource named "{rname}" from the project')
def step_get_resource(context: Context, rname: str) -> None:
"""Get a resource by name from the project."""
context.retrieved_resource = context.project.get_resource(rname)
# Project field assertions
@then('the project project_id should be "{expected}"')
def step_check_project_id(context: Context, expected: str) -> None:
"""Verify project_id."""
assert context.project.project_id == expected
# Note: "the project name should be" step is defined in service_steps.py
@then('the project namespace should be "{expected}"')
def step_check_project_namespace(context: Context, expected: str) -> None:
"""Verify project namespace."""
assert context.project.namespace == expected
@then("the project description should be none")
def step_check_project_description_none(context: Context) -> None:
"""Verify description is None."""
assert context.project.description is None
@then('the project description should be "{expected}"')
def step_check_project_description(context: Context, expected: str) -> None:
"""Verify description value."""
assert context.project.description == expected
@then("the project tags should be empty")
def step_check_project_tags_empty(context: Context) -> None:
"""Verify tags is empty list."""
assert context.project.tags == []
@then('the project tags should contain "{tag}"')
def step_check_project_tag(context: Context, tag: str) -> None:
"""Verify tags contains expected value."""
assert tag in context.project.tags
@then("the project resources should be empty")
def step_check_project_resources_empty(context: Context) -> None:
"""Verify resources is empty list."""
assert context.project.resources == []
@then("the project validation_config should be none")
def step_check_project_validation_config_none(context: Context) -> None:
"""Verify validation_config is None."""
assert context.project.validation_config is None
@then("the project context_config should not be none")
def step_check_project_context_config_not_none(context: Context) -> None:
"""Verify context_config is set."""
assert context.project.context_config is not None
@then("the project created_at should be set")
def step_check_project_created_at(context: Context) -> None:
"""Verify created_at is set."""
assert context.project.created_at is not None
@then("the project updated_at should be set")
def step_check_project_updated_at(context: Context) -> None:
"""Verify updated_at is set."""
assert context.project.updated_at is not None
@then("the project is_remote should be false")
def step_check_project_not_remote(context: Context) -> None:
"""Verify project is not remote."""
assert context.project.is_remote is False
@then("the project is_remote should be true")
def step_check_project_is_remote(context: Context) -> None:
"""Verify project is remote."""
assert context.project.is_remote is True
@then('the project namespaced_name should be "{expected}"')
def step_check_project_namespaced_name(context: Context, expected: str) -> None:
"""Verify namespaced_name property."""
assert context.project.namespaced_name == expected
@then("the project should have {count:d} resource")
def step_check_project_resource_count(context: Context, count: int) -> None:
"""Verify resource count."""
assert len(context.project.resources) == count
@then('the project should have a resource named "{rname}"')
def step_check_project_has_resource(context: Context, rname: str) -> None:
"""Verify project has a resource with the given name."""
found = context.project.get_resource(rname)
assert found is not None, f"Expected resource '{rname}' not found"
@then('the retrieved resource name should be "{expected}"')
def step_check_retrieved_resource_name(context: Context, expected: str) -> None:
"""Verify retrieved resource name."""
assert context.retrieved_resource is not None
assert context.retrieved_resource.name == expected
@then("the retrieved resource should be none")
def step_check_retrieved_resource_none(context: Context) -> None:
"""Verify retrieved resource is None."""
assert context.retrieved_resource is None
# parse_namespaced_name steps
@when('I parse the project namespaced name "{namespaced}"')
def step_parse_project_namespaced_name(context: Context, namespaced: str) -> None:
"""Parse a namespaced name string via Project.parse_namespaced_name."""
namespace, name = Project.parse_namespaced_name(namespaced)
context.parsed_project_namespace = namespace
context.parsed_project_name = name
@when('I try to parse the project namespaced name "{namespaced}"')
def step_try_parse_project_namespaced_name(context: Context, namespaced: str) -> None:
"""Attempt to parse an invalid namespaced name string."""
context.error = None
try:
Project.parse_namespaced_name(namespaced)
except ValueError as e:
context.error = e
@then('the parsed project namespace should be "{expected}"')
def step_check_parsed_project_namespace(context: Context, expected: str) -> None:
"""Verify parsed project namespace."""
assert context.parsed_project_namespace == expected
@then('the parsed project name should be "{expected}"')
def step_check_parsed_project_name(context: Context, expected: str) -> None:
"""Verify parsed project name."""
assert context.parsed_project_name == expected
@then("a project namespaced name parse error should be raised")
def step_check_project_namespaced_name_error(context: Context) -> None:
"""Verify a namespaced name parse error was raised."""
assert context.error is not None, (
"Expected a ValueError for invalid namespaced name but none was raised"
)
# Name validation steps
@when('I try to create a Project with invalid name "{name}"')
def step_try_create_project_bad_name(context: Context, name: str) -> None:
"""Attempt to create a Project with an invalid name."""
context.error = None
try:
context.project = Project(
project_id=VALID_ULID,
name=name,
path=Path("/tmp/test"),
)
except (ValidationError, ValueError) as e:
context.error = e
@then("a name validation error should be raised")
def step_check_name_error(context: Context) -> None:
"""Verify a name validation error was raised."""
assert context.error is not None, (
"Expected a name validation error but none was raised"
)
@@ -0,0 +1,322 @@
"""Step definitions for Project Service v3 with Persistence (B5 Item 5)."""
from typing import Any
from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a clean database for project service testing")
def step_clean_db(context: Context) -> None:
"""Set up a clean in-memory database and all services."""
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
from cleveragents.infrastructure.database.migration_runner import MigrationRunner
database_url = "sqlite:///:memory:"
engine = create_engine(database_url, connect_args={"check_same_thread": False})
MEMORY_ENGINES[database_url] = engine
migration_runner = MigrationRunner(database_url)
migration_runner.run_migrations(engine=engine)
context.engine = engine
context.session_factory = sessionmaker(
bind=engine, autoflush=False, autocommit=False
)
context.db_session = context.session_factory()
# Set up repositories
from cleveragents.infrastructure.database.project_link_repository import (
ProjectResourceLinkRepository,
)
from cleveragents.infrastructure.database.repositories import (
ProjectRepository,
)
from cleveragents.infrastructure.database.resource_repository import (
ResourceRepository,
)
project_repo = ProjectRepository(context.db_session)
resource_repo = ResourceRepository(context.db_session)
link_repo = ProjectResourceLinkRepository(context.db_session)
# Create the service
from cleveragents.application.services.project_service_persistent import (
ProjectServicePersistent,
)
context.project_service = ProjectServicePersistent(
project_repo=project_repo,
resource_repo=resource_repo,
link_repo=link_repo,
session=context.db_session,
)
# ---------------------------------------------------------------------------
# Create steps
# ---------------------------------------------------------------------------
@when('I create a project named "{name}" in namespace "{namespace}"')
def step_create_project(context: Context, name: str, namespace: str) -> None:
"""Create a project via the service."""
context.created_project = context.project_service.create(
name=name, namespace=namespace
)
@when(
'I create a described project named "{name}" in namespace "{namespace}"'
' described as "{description}"'
)
def step_create_project_desc(
context: Context, name: str, namespace: str, description: str
) -> None:
"""Create a project with description."""
context.created_project = context.project_service.create(
name=name, namespace=namespace, description=description
)
@given('an existing project "{name}" in namespace "{namespace}"')
def step_existing_project(context: Context, name: str, namespace: str) -> None:
"""Create a project as a precondition."""
context.created_project = context.project_service.create(
name=name, namespace=namespace
)
@given('an existing project "{name}" in namespace "{namespace}" with validation')
def step_existing_project_val(context: Context, name: str, namespace: str) -> None:
"""Create a project with validation config as precondition."""
context.created_project = context.project_service.create(
name=name,
namespace=namespace,
validation_config={
"test_command": "pytest",
"lint_command": "ruff check .",
},
)
@when('I try to create a project named "{name}" in namespace "{namespace}"')
def step_try_create_dup(context: Context, name: str, namespace: str) -> None:
"""Try to create a duplicate project."""
context.create_error = None
try:
context.project_service.create(name=name, namespace=namespace)
except ValueError as e:
context.create_error = e
@then('the created project should have name "{name}"')
def step_check_name(context: Context, name: str) -> None:
"""Verify created project name."""
assert context.created_project["name"] == name, (
f"Expected '{name}', got '{context.created_project['name']}'"
)
@then('the created project should have namespace "{namespace}"')
def step_check_namespace(context: Context, namespace: str) -> None:
"""Verify created project namespace."""
assert context.created_project["namespace"] == namespace, (
f"Expected '{namespace}', got '{context.created_project['namespace']}'"
)
@then('the created project should have description "{description}"')
def step_check_description(context: Context, description: str) -> None:
"""Verify created project description."""
assert context.created_project["description"] == description, (
f"Expected '{description}', got '{context.created_project['description']}'"
)
@then("a project already exists error should be raised")
def step_check_dup_error(context: Context) -> None:
"""Verify duplicate project error."""
assert context.create_error is not None, "Expected error"
assert "already exists" in str(context.create_error).lower(), (
f"Expected 'already exists' in: {context.create_error}"
)
# ---------------------------------------------------------------------------
# List steps
# ---------------------------------------------------------------------------
@when("I list all projects via service")
def step_list_all(context: Context) -> None:
"""List all projects."""
context.service_list = context.project_service.list_all()
@when('I list projects by namespace "{namespace}" via service')
def step_list_by_ns(context: Context, namespace: str) -> None:
"""List projects filtered by namespace."""
context.service_list = context.project_service.list_by_namespace(
namespace=namespace
)
@then("the service project list should have {count:d} entries")
def step_check_list_count(context: Context, count: int) -> None:
"""Verify list count."""
assert len(context.service_list) == count, (
f"Expected {count} projects, got {len(context.service_list)}"
)
# ---------------------------------------------------------------------------
# Show steps
# ---------------------------------------------------------------------------
@when('I show the project "{name}" via service')
def step_show_project(context: Context, name: str) -> None:
"""Show a project."""
context.shown_project = context.project_service.show(name=name)
@when('I try to show the project "{name}" via service')
def step_try_show(context: Context, name: str) -> None:
"""Try to show a non-existent project."""
context.show_error = None
try:
context.project_service.show(name=name)
except ValueError as e:
context.show_error = e
@then('the shown project should have name "{name}"')
def step_check_shown_name(context: Context, name: str) -> None:
"""Verify shown project name."""
assert context.shown_project["name"] == name, (
f"Expected '{name}', got '{context.shown_project['name']}'"
)
@then("a persistent project not found error should be raised")
def step_check_not_found(context: Context) -> None:
"""Verify not found error."""
assert context.show_error is not None, "Expected not found error"
assert "not found" in str(context.show_error).lower(), (
f"Expected 'not found' in: {context.show_error}"
)
# ---------------------------------------------------------------------------
# Delete steps
# ---------------------------------------------------------------------------
@when('I delete the project "{name}" via service')
def step_delete_project(context: Context, name: str) -> None:
"""Delete a project."""
context.project_service.delete(name=name)
@then('showing project "{name}" via service should raise not found')
def step_show_deleted(context: Context, name: str) -> None:
"""Verify deleted project raises not found."""
try:
context.project_service.show(name=name)
assert False, "Expected not found error"
except ValueError:
pass
# ---------------------------------------------------------------------------
# Link / Unlink steps
# ---------------------------------------------------------------------------
@given('a registered service resource "{name}" of type "{rtype}" at "{location}"')
def step_registered_svc_resource(
context: Context, name: str, rtype: str, location: str
) -> None:
"""Register a resource for service testing."""
context.svc_resource = context.project_service.register_resource(
name=name, resource_type=rtype, location=location
)
@when('I link resource "{res_name}" to project "{proj_name}" with alias "{alias}"')
def step_link_svc(context: Context, res_name: str, proj_name: str, alias: str) -> None:
"""Link a resource to a project via service."""
context.project_service.link_resource(
project_name=proj_name, resource_name=res_name, alias=alias
)
@given('resource "{res_name}" is linked to project "{proj_name}" with alias "{alias}"')
def step_given_linked_svc(
context: Context, res_name: str, proj_name: str, alias: str
) -> None:
"""Link a resource as a precondition."""
context.project_service.link_resource(
project_name=proj_name, resource_name=res_name, alias=alias
)
@when('I unlink resource "{res_name}" from project "{proj_name}"')
def step_unlink_svc(context: Context, res_name: str, proj_name: str) -> None:
"""Unlink a resource from a project."""
context.project_service.unlink_resource(
project_name=proj_name, resource_name=res_name
)
@then('the project "{name}" should have {count:d} linked resource')
def step_check_linked_count_s(context: Context, name: str, count: int) -> None:
"""Verify linked resource count (singular)."""
links = context.project_service.get_links(project_name=name)
assert len(links) == count, f"Expected {count} links, got {len(links)}"
@then('the project "{name}" should have {count:d} linked resources')
def step_check_linked_count_p(context: Context, name: str, count: int) -> None:
"""Verify linked resource count (plural)."""
links = context.project_service.get_links(project_name=name)
assert len(links) == count, f"Expected {count} links, got {len(links)}"
# ---------------------------------------------------------------------------
# Validation summary steps
# ---------------------------------------------------------------------------
@when('I get the validation summary for project "{name}" via service')
def step_get_val_summary_svc(context: Context, name: str) -> None:
"""Get validation summary."""
context.svc_summary = context.project_service.get_validation_summary(
project_name=name
)
@then("the service summary should include validation config")
def step_check_svc_val_config(context: Context) -> None:
"""Verify summary includes validation config."""
assert context.svc_summary is not None, "No summary"
assert context.svc_summary["validation_config"] is not None, (
"Validation config missing"
)
assert context.svc_summary["validation_config"]["test_command"] == "pytest"
@then("the service summary should list {count:d} linked resource")
def step_check_svc_summary_count(context: Context, count: int) -> None:
"""Verify summary linked resource count."""
assert len(context.svc_summary["linked_resources"]) == count, (
f"Expected {count}, got {len(context.svc_summary['linked_resources'])}"
)
+558
View File
@@ -0,0 +1,558 @@
"""Step definitions for Resource domain model tests (B1.2, B1.3, B1.4)."""
from __future__ import annotations
from pathlib import Path
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.resource import (
Resource,
ResourceType,
SandboxStrategy,
)
# ResourceType Steps
@then('the resource types should be "{expected_order}"')
def step_check_resource_type_values(context: Context, expected_order: str) -> None:
"""Verify that ResourceType has all expected values in order."""
expected = [v.strip() for v in expected_order.split(",")]
actual = [rt.value for rt in ResourceType]
assert actual == expected, f"Expected resource types {expected}, got {actual}"
@when("I access ResourceType.GIT_REPOSITORY")
def step_access_git_repository(context: Context) -> None:
"""Access the GIT_REPOSITORY enum member."""
context.resource_type = ResourceType.GIT_REPOSITORY
@when("I access ResourceType.FILESYSTEM")
def step_access_filesystem(context: Context) -> None:
"""Access the FILESYSTEM enum member."""
context.resource_type = ResourceType.FILESYSTEM
@when("I access ResourceType.DATABASE")
def step_access_database(context: Context) -> None:
"""Access the DATABASE enum member."""
context.resource_type = ResourceType.DATABASE
@when("I access ResourceType.API_ENDPOINT")
def step_access_api_endpoint(context: Context) -> None:
"""Access the API_ENDPOINT enum member."""
context.resource_type = ResourceType.API_ENDPOINT
@when("I access ResourceType.DOCUMENT_CORPUS")
def step_access_document_corpus(context: Context) -> None:
"""Access the DOCUMENT_CORPUS enum member."""
context.resource_type = ResourceType.DOCUMENT_CORPUS
@when("I access ResourceType.CLOUD_INFRASTRUCTURE")
def step_access_cloud_infrastructure(context: Context) -> None:
"""Access the CLOUD_INFRASTRUCTURE enum member."""
context.resource_type = ResourceType.CLOUD_INFRASTRUCTURE
@then('the resource type value should be "{expected}"')
def step_check_resource_type_value(context: Context, expected: str) -> None:
"""Check the resource type value matches expected."""
actual = context.resource_type.value
assert actual == expected, f"Expected value '{expected}', got '{actual}'"
@then("the resource type should be a string")
def step_check_resource_type_is_string(context: Context) -> None:
"""Verify ResourceType inherits from str."""
assert isinstance(context.resource_type, str), (
f"Expected ResourceType to be a string, got {type(context.resource_type)}"
)
@when('I create a ResourceType from string "{value}"')
def step_create_resource_type_from_string(context: Context, value: str) -> None:
"""Create a ResourceType from a string value."""
context.error = None
try:
context.resource_type = ResourceType(value)
except ValueError as e:
context.error = e
@when('I try to create a ResourceType from string "{value}"')
def step_try_create_resource_type_from_string(context: Context, value: str) -> None:
"""Attempt to create a ResourceType from an invalid string."""
context.error = None
try:
context.resource_type = ResourceType(value)
except ValueError as e:
context.error = e
@then("a ValueError should be raised")
def step_check_value_error_raised(context: Context) -> None:
"""Verify a ValueError was captured."""
assert context.error is not None, "Expected a ValueError but none was raised"
assert isinstance(context.error, ValueError), (
f"Expected ValueError, got {type(context.error).__name__}"
)
# SandboxStrategy Steps
@then('the sandbox strategies should be "{expected_order}"')
def step_check_sandbox_strategy_values(context: Context, expected_order: str) -> None:
"""Verify that SandboxStrategy has all expected values in order."""
expected = [v.strip() for v in expected_order.split(",")]
actual = [ss.value for ss in SandboxStrategy]
assert actual == expected, f"Expected sandbox strategies {expected}, got {actual}"
@when("I access SandboxStrategy.GIT_WORKTREE")
def step_access_git_worktree(context: Context) -> None:
"""Access the GIT_WORKTREE enum member."""
context.sandbox_strategy = SandboxStrategy.GIT_WORKTREE
@when("I access SandboxStrategy.COPY_ON_WRITE")
def step_access_copy_on_write(context: Context) -> None:
"""Access the COPY_ON_WRITE enum member."""
context.sandbox_strategy = SandboxStrategy.COPY_ON_WRITE
@when("I access SandboxStrategy.OVERLAY")
def step_access_overlay(context: Context) -> None:
"""Access the OVERLAY enum member."""
context.sandbox_strategy = SandboxStrategy.OVERLAY
@when("I access SandboxStrategy.TRANSACTION_ROLLBACK")
def step_access_transaction_rollback(context: Context) -> None:
"""Access the TRANSACTION_ROLLBACK enum member."""
context.sandbox_strategy = SandboxStrategy.TRANSACTION_ROLLBACK
@when("I access SandboxStrategy.VERSIONING")
def step_access_versioning(context: Context) -> None:
"""Access the VERSIONING enum member."""
context.sandbox_strategy = SandboxStrategy.VERSIONING
@when("I access SandboxStrategy.NONE")
def step_access_none_strategy(context: Context) -> None:
"""Access the NONE enum member."""
context.sandbox_strategy = SandboxStrategy.NONE
@then('the sandbox strategy value should be "{expected}"')
def step_check_sandbox_strategy_value(context: Context, expected: str) -> None:
"""Check the sandbox strategy value matches expected."""
actual = context.sandbox_strategy.value
assert actual == expected, f"Expected value '{expected}', got '{actual}'"
@then("the sandbox strategy should be a string")
def step_check_sandbox_strategy_is_string(context: Context) -> None:
"""Verify SandboxStrategy inherits from str."""
assert isinstance(context.sandbox_strategy, str), (
f"Expected SandboxStrategy to be a string, got {type(context.sandbox_strategy)}"
)
# SandboxStrategy.supports_rollback Tests
@then("SandboxStrategy.GIT_WORKTREE should support rollback")
def step_git_worktree_supports_rollback(context: Context) -> None:
"""Verify GIT_WORKTREE supports rollback."""
assert SandboxStrategy.GIT_WORKTREE.supports_rollback is True
@then("SandboxStrategy.COPY_ON_WRITE should support rollback")
def step_copy_on_write_supports_rollback(context: Context) -> None:
"""Verify COPY_ON_WRITE supports rollback."""
assert SandboxStrategy.COPY_ON_WRITE.supports_rollback is True
@then("SandboxStrategy.OVERLAY should support rollback")
def step_overlay_supports_rollback(context: Context) -> None:
"""Verify OVERLAY supports rollback."""
assert SandboxStrategy.OVERLAY.supports_rollback is True
@then("SandboxStrategy.TRANSACTION_ROLLBACK should support rollback")
def step_transaction_rollback_supports_rollback(context: Context) -> None:
"""Verify TRANSACTION_ROLLBACK supports rollback."""
assert SandboxStrategy.TRANSACTION_ROLLBACK.supports_rollback is True
@then("SandboxStrategy.VERSIONING should support rollback")
def step_versioning_supports_rollback(context: Context) -> None:
"""Verify VERSIONING supports rollback."""
assert SandboxStrategy.VERSIONING.supports_rollback is True
@then("SandboxStrategy.NONE should not support rollback")
def step_none_does_not_support_rollback(context: Context) -> None:
"""Verify NONE does not support rollback."""
assert SandboxStrategy.NONE.supports_rollback is False
# SandboxStrategy.is_copy_based Tests
@then("SandboxStrategy.COPY_ON_WRITE should be copy based")
def step_copy_on_write_is_copy_based(context: Context) -> None:
"""Verify COPY_ON_WRITE is copy based."""
assert SandboxStrategy.COPY_ON_WRITE.is_copy_based is True
@then("SandboxStrategy.OVERLAY should be copy based")
def step_overlay_is_copy_based(context: Context) -> None:
"""Verify OVERLAY is copy based."""
assert SandboxStrategy.OVERLAY.is_copy_based is True
@then("SandboxStrategy.GIT_WORKTREE should not be copy based")
def step_git_worktree_not_copy_based(context: Context) -> None:
"""Verify GIT_WORKTREE is not copy based."""
assert SandboxStrategy.GIT_WORKTREE.is_copy_based is False
@then("SandboxStrategy.TRANSACTION_ROLLBACK should not be copy based")
def step_transaction_rollback_not_copy_based(context: Context) -> None:
"""Verify TRANSACTION_ROLLBACK is not copy based."""
assert SandboxStrategy.TRANSACTION_ROLLBACK.is_copy_based is False
@then("SandboxStrategy.VERSIONING should not be copy based")
def step_versioning_not_copy_based(context: Context) -> None:
"""Verify VERSIONING is not copy based."""
assert SandboxStrategy.VERSIONING.is_copy_based is False
@then("SandboxStrategy.NONE should not be copy based")
def step_none_not_copy_based(context: Context) -> None:
"""Verify NONE is not copy based."""
assert SandboxStrategy.NONE.is_copy_based is False
# Resource Pydantic Model Steps (B1.2)
VALID_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
@given(
'a Resource with required fields id "{resource_id}" name "{name}" type "{rtype}" location "{location}"'
)
def step_create_resource(
context: Context, resource_id: str, name: str, rtype: str, location: str
) -> None:
"""Create a Resource with required fields only."""
context.error = None
try:
context.resource = Resource(
resource_id=resource_id,
name=name,
type=ResourceType(rtype),
location=location,
)
except (ValidationError, ValueError) as e:
context.error = e
@given("I have a valid Resource")
def step_have_valid_resource(context: Context) -> None:
"""Create a valid Resource for mutation testing."""
context.resource = Resource(
resource_id=VALID_ULID,
name="my-repo",
type=ResourceType.GIT_REPOSITORY,
location="/tmp/repo",
)
@when("I try to modify the resource name")
def step_try_modify_resource_name(context: Context) -> None:
"""Attempt to modify a frozen model field."""
context.error = None
try:
context.resource.name = "new-name" # type: ignore[misc]
except ValidationError as e:
context.error = e
@given(
'a remote Resource with id "{resource_id}" name "{name}" type "{rtype}" location "{location}"'
)
def step_create_resource_remote(
context: Context, resource_id: str, name: str, rtype: str, location: str
) -> None:
"""Create a remote Resource."""
context.error = None
try:
context.resource = Resource(
resource_id=resource_id,
name=name,
type=ResourceType(rtype),
location=location,
is_remote=True,
)
except (ValidationError, ValueError) as e:
context.error = e
@given(
'a read-only Resource with id "{resource_id}" name "{name}" type "{rtype}" location "{location}"'
)
def step_create_resource_read_only(
context: Context, resource_id: str, name: str, rtype: str, location: str
) -> None:
"""Create a read-only Resource."""
context.error = None
try:
context.resource = Resource(
resource_id=resource_id,
name=name,
type=ResourceType(rtype),
location=location,
read_only=True,
)
except (ValidationError, ValueError) as e:
context.error = e
@given(
'a Resource with id "{resource_id}" name "{name}" type "{rtype}" location "{location}" strategy "{strategy}"'
)
def step_create_resource_with_strategy(
context: Context,
resource_id: str,
name: str,
rtype: str,
location: str,
strategy: str,
) -> None:
"""Create a Resource with a specific sandbox strategy."""
context.error = None
try:
context.resource = Resource(
resource_id=resource_id,
name=name,
type=ResourceType(rtype),
location=location,
sandbox_strategy=SandboxStrategy(strategy),
)
except (ValidationError, ValueError) as e:
context.error = e
@given(
'a Resource with id "{resource_id}" name "{name}" type "{rtype}" location "{location}" metadata "{meta_str}"'
)
def step_create_resource_with_metadata(
context: Context,
resource_id: str,
name: str,
rtype: str,
location: str,
meta_str: str,
) -> None:
"""Create a Resource with metadata parsed from key=value pairs."""
metadata = {}
for pair in meta_str.split(","):
key, value = pair.strip().split("=", 1)
metadata[key.strip()] = value.strip()
context.error = None
try:
context.resource = Resource(
resource_id=resource_id,
name=name,
type=ResourceType(rtype),
location=location,
metadata=metadata,
)
except (ValidationError, ValueError) as e:
context.error = e
@when(
'I try to create a Resource with id "{resource_id}" name "{name}" type "{rtype}" location "{location}"'
)
def step_try_create_resource(
context: Context, resource_id: str, name: str, rtype: str, location: str
) -> None:
"""Attempt to create a Resource that may fail validation."""
context.error = None
try:
context.resource = Resource(
resource_id=resource_id,
name=name,
type=ResourceType(rtype),
location=location,
)
except (ValidationError, ValueError) as e:
context.error = e
@when("I try to create a Resource with an empty name")
def step_try_create_resource_empty_name(context: Context) -> None:
"""Attempt to create a Resource with empty name."""
context.error = None
try:
context.resource = Resource(
resource_id=VALID_ULID,
name="",
type=ResourceType.GIT_REPOSITORY,
location="/tmp/repo",
)
except (ValidationError, ValueError) as e:
context.error = e
@when("I try to create a Resource with an empty location")
def step_try_create_resource_empty_location(context: Context) -> None:
"""Attempt to create a Resource with empty location."""
context.error = None
try:
context.resource = Resource(
resource_id=VALID_ULID,
name="my-repo",
type=ResourceType.GIT_REPOSITORY,
location="",
)
except (ValidationError, ValueError) as e:
context.error = e
# Resource field assertion steps
@then('the resource name should be "{expected}"')
def step_check_resource_name(context: Context, expected: str) -> None:
"""Verify resource name."""
assert context.resource.name == expected, (
f"Expected name '{expected}', got '{context.resource.name}'"
)
@then("the resource type should be ResourceType.GIT_REPOSITORY")
def step_check_resource_type_git(context: Context) -> None:
"""Verify resource type is GIT_REPOSITORY."""
assert context.resource.type == ResourceType.GIT_REPOSITORY
@then('the resource location should be "{expected}"')
def step_check_resource_location(context: Context, expected: str) -> None:
"""Verify resource location."""
assert context.resource.location == expected, (
f"Expected location '{expected}', got '{context.resource.location}'"
)
@then("the resource is_remote should be false")
def step_check_resource_not_remote(context: Context) -> None:
"""Verify resource is not remote."""
assert context.resource.is_remote is False
@then("the resource is_remote should be true")
def step_check_resource_is_remote(context: Context) -> None:
"""Verify resource is remote."""
assert context.resource.is_remote is True
@then("the resource sandbox_strategy should be SandboxStrategy.NONE")
def step_check_resource_strategy_none(context: Context) -> None:
"""Verify default sandbox strategy is NONE."""
assert context.resource.sandbox_strategy == SandboxStrategy.NONE
@then("the resource sandbox_strategy should be SandboxStrategy.GIT_WORKTREE")
def step_check_resource_strategy_worktree(context: Context) -> None:
"""Verify sandbox strategy is GIT_WORKTREE."""
assert context.resource.sandbox_strategy == SandboxStrategy.GIT_WORKTREE
@then("the resource read_only should be false")
def step_check_resource_not_readonly(context: Context) -> None:
"""Verify resource is not read-only."""
assert context.resource.read_only is False
@then("the resource read_only should be true")
def step_check_resource_is_readonly(context: Context) -> None:
"""Verify resource is read-only."""
assert context.resource.read_only is True
@then("the resource metadata should be empty")
def step_check_resource_metadata_empty(context: Context) -> None:
"""Verify resource metadata is empty dict."""
assert context.resource.metadata == {}, (
f"Expected empty metadata, got {context.resource.metadata}"
)
@then("the resource created_at should be set")
def step_check_resource_created_at(context: Context) -> None:
"""Verify resource has a created_at timestamp."""
assert context.resource.created_at is not None
@then('the resource metadata should contain key "{key}" with value "{value}"')
def step_check_resource_metadata_key(context: Context, key: str, value: str) -> None:
"""Verify resource metadata contains expected key-value pair."""
assert key in context.resource.metadata, (
f"Expected key '{key}' in metadata, got {context.resource.metadata}"
)
assert context.resource.metadata[key] == value, (
f"Expected metadata['{key}'] = '{value}', got '{context.resource.metadata[key]}'"
)
# Note: "a validation error should be raised" step is defined in domain_models_steps.py
# Resource property assertion steps
@then("the resource supports_sandbox should be true")
def step_check_supports_sandbox_true(context: Context) -> None:
"""Verify resource supports sandbox."""
assert context.resource.supports_sandbox is True
@then("the resource supports_sandbox should be false")
def step_check_supports_sandbox_false(context: Context) -> None:
"""Verify resource does not support sandbox."""
assert context.resource.supports_sandbox is False
@then("the resource can_write should be true")
def step_check_can_write_true(context: Context) -> None:
"""Verify resource can be written to."""
assert context.resource.can_write is True
@then("the resource can_write should be false")
def step_check_can_write_false(context: Context) -> None:
"""Verify resource cannot be written to."""
assert context.resource.can_write is False
@then('the resource get_sandbox_path with base "{base}" should end with "{suffix}"')
def step_check_sandbox_path(context: Context, base: str, suffix: str) -> None:
"""Verify get_sandbox_path returns expected path."""
result = context.resource.get_sandbox_path(Path(base))
assert str(result).endswith(suffix), (
f"Expected path ending with '{suffix}', got '{result}'"
)
@@ -0,0 +1,290 @@
"""Step definitions for Resource Registry Service (B5 Item 4)."""
import re
from typing import Any
from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
ULID_RE = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$")
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a clean database for resource registry testing")
def step_clean_db(context: Context) -> None:
"""Set up a clean in-memory database and the service."""
from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
from cleveragents.infrastructure.database.migration_runner import MigrationRunner
database_url = "sqlite:///:memory:"
engine = create_engine(database_url, connect_args={"check_same_thread": False})
MEMORY_ENGINES[database_url] = engine
migration_runner = MigrationRunner(database_url)
migration_runner.run_migrations(engine=engine)
context.engine = engine
context.session_factory = sessionmaker(
bind=engine, autoflush=False, autocommit=False
)
context.db_session = context.session_factory()
# Import and create the service
from cleveragents.application.services.resource_registry_service import (
ResourceRegistryService,
)
from cleveragents.infrastructure.database.resource_repository import (
ResourceRepository,
)
resource_repo = ResourceRepository(context.db_session)
context.registry_service = ResourceRegistryService(
resource_repo=resource_repo,
session=context.db_session,
)
# ---------------------------------------------------------------------------
# Register steps
# ---------------------------------------------------------------------------
@when('I register a resource named "{name}" of type "{rtype}" at "{location}"')
def step_register_resource(
context: Context, name: str, rtype: str, location: str
) -> None:
"""Register a new resource."""
context.reg_result = context.registry_service.register(
name=name, resource_type=rtype, location=location
)
@when(
'I register a sandboxed resource named "{name}" of type "{rtype}"'
' at "{location}" with strategy "{sandbox}"'
)
def step_register_resource_sandbox(
context: Context, name: str, rtype: str, location: str, sandbox: str
) -> None:
"""Register a resource with a specific sandbox strategy."""
context.reg_result = context.registry_service.register(
name=name,
resource_type=rtype,
location=location,
sandbox_strategy=sandbox,
)
@when(
'I register a read-only resource named "{name}" of type "{rtype}" at "{location}"'
)
def step_register_readonly(
context: Context, name: str, rtype: str, location: str
) -> None:
"""Register a read-only resource."""
context.reg_result = context.registry_service.register(
name=name, resource_type=rtype, location=location, read_only=True
)
@given('a registered resource named "{name}" of type "{rtype}" at "{location}"')
def step_given_registered(
context: Context, name: str, rtype: str, location: str
) -> None:
"""Register a resource as a precondition."""
context.reg_result = context.registry_service.register(
name=name, resource_type=rtype, location=location
)
@when('I try to register a resource named "{name}" of type "{rtype}" at "{location}"')
def step_try_register_dup(
context: Context, name: str, rtype: str, location: str
) -> None:
"""Try to register a duplicate resource."""
context.reg_error = None
try:
context.registry_service.register(
name=name, resource_type=rtype, location=location
)
except ValueError as e:
context.reg_error = e
@then('the registration result should have name "{name}"')
def step_check_reg_name(context: Context, name: str) -> None:
"""Verify registration result name."""
assert context.reg_result["name"] == name, (
f"Expected name '{name}', got '{context.reg_result['name']}'"
)
@then("the registration result should have a valid resource ID")
def step_check_reg_ulid(context: Context) -> None:
"""Verify registration result has valid ULID."""
rid = context.reg_result["resource_id"]
assert ULID_RE.match(rid), f"Invalid ULID: {rid}"
@then('the registration result should have sandbox strategy "{strategy}"')
def step_check_reg_sandbox(context: Context, strategy: str) -> None:
"""Verify sandbox strategy."""
assert context.reg_result["sandbox_strategy"] == strategy, (
f"Expected '{strategy}', got '{context.reg_result['sandbox_strategy']}'"
)
@then("the registration result should be read-only")
def step_check_reg_readonly(context: Context) -> None:
"""Verify resource is read-only."""
assert context.reg_result["read_only"] is True, "Expected read-only"
@then("a duplicate resource error should be raised")
def step_check_dup_error(context: Context) -> None:
"""Verify duplicate error was raised."""
assert context.reg_error is not None, "Expected duplicate error"
assert "already exists" in str(context.reg_error).lower(), (
f"Expected 'already exists' in: {context.reg_error}"
)
@then('the registration result should have type "{rtype}"')
def step_check_reg_type(context: Context, rtype: str) -> None:
"""Verify registration result type."""
assert context.reg_result["type"] == rtype, (
f"Expected type '{rtype}', got '{context.reg_result['type']}'"
)
# ---------------------------------------------------------------------------
# Remove steps
# ---------------------------------------------------------------------------
@when('I remove the resource named "{name}"')
def step_remove_resource(context: Context, name: str) -> None:
"""Remove a resource by name."""
context.registry_service.remove(name=name)
@when('I try to remove the resource named "{name}"')
def step_try_remove(context: Context, name: str) -> None:
"""Try to remove a non-existent resource."""
context.remove_error = None
try:
context.registry_service.remove(name=name)
except ValueError as e:
context.remove_error = e
@then('showing resource "{name}" should raise not found')
def step_show_raises_not_found(context: Context, name: str) -> None:
"""Verify show raises not found."""
try:
context.registry_service.show(name=name)
assert False, "Expected not found error"
except ValueError:
pass
@then("a resource not found error should be raised")
def step_check_not_found_error(context: Context) -> None:
"""Verify not found error."""
assert context.remove_error is not None, "Expected not found error"
assert "not found" in str(context.remove_error).lower(), (
f"Expected 'not found' in: {context.remove_error}"
)
# ---------------------------------------------------------------------------
# Show steps
# ---------------------------------------------------------------------------
@when('I show the resource named "{name}"')
def step_show_resource(context: Context, name: str) -> None:
"""Show a resource by name."""
context.shown_resource = context.registry_service.show(name=name)
@when("I show the resource by its ULID")
def step_show_by_ulid(context: Context) -> None:
"""Show a resource by its ULID."""
ulid = context.reg_result["resource_id"]
context.shown_resource = context.registry_service.show(name=ulid)
@then('the shown resource should have type "{rtype}"')
def step_check_shown_type(context: Context, rtype: str) -> None:
"""Verify shown resource type."""
assert context.shown_resource["type"] == rtype, (
f"Expected type '{rtype}', got '{context.shown_resource['type']}'"
)
@then('the shown resource should have location "{location}"')
def step_check_shown_location(context: Context, location: str) -> None:
"""Verify shown resource location."""
assert context.shown_resource["location"] == location, (
f"Expected '{location}', got '{context.shown_resource['location']}'"
)
@then('the shown resource should have name "{name}"')
def step_check_shown_name(context: Context, name: str) -> None:
"""Verify shown resource name."""
assert context.shown_resource["name"] == name, (
f"Expected '{name}', got '{context.shown_resource['name']}'"
)
# ---------------------------------------------------------------------------
# Tree steps
# ---------------------------------------------------------------------------
@given('resource "{child}" is a child of "{parent}"')
def step_link_child(context: Context, child: str, parent: str) -> None:
"""Link a child resource to a parent."""
context.registry_service.link_child(parent_name=parent, child_name=child)
@when('I get the resource tree for "{name}"')
def step_get_tree(context: Context, name: str) -> None:
"""Get the resource tree."""
context.tree_result = context.registry_service.tree(name=name)
@then("the tree should contain {count:d} resources")
def step_check_tree_count(context: Context, count: int) -> None:
"""Verify tree size."""
assert len(context.tree_result) == count, (
f"Expected {count} resources, got {len(context.tree_result)}"
)
@then('the tree should include resource "{name}"')
def step_check_tree_includes(context: Context, name: str) -> None:
"""Verify tree includes a resource."""
names = [r["name"] for r in context.tree_result]
assert name in names, f"Resource '{name}' not in tree: {names}"
# ---------------------------------------------------------------------------
# Auto-discovery steps
# ---------------------------------------------------------------------------
@when('I auto-discover a resource at "{location}" with name "{name}"')
def step_auto_discover(context: Context, location: str, name: str) -> None:
"""Auto-discover a resource."""
context.reg_result = context.registry_service.auto_discover(
location=location, name=name
)
@@ -0,0 +1,382 @@
"""Persistence-backed Project Service for CleverAgents.
Provides create/list/show/delete/link/unlink operations backed by
database repositories instead of in-memory storage.
Based on implementation_plan.md B5 Item 5 and ADR-007 (Repository Pattern).
"""
import random
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from sqlalchemy.orm import Session
from cleveragents.domain.models.core.project import (
ContextConfig,
Project,
ProjectSettings,
ValidationConfig,
)
from cleveragents.domain.models.core.resource import (
Resource,
ResourceType,
SandboxStrategy,
)
from cleveragents.infrastructure.database.project_link_repository import (
ProjectResourceLinkRepository,
)
from cleveragents.infrastructure.database.repositories import (
ProjectRepository,
)
from cleveragents.infrastructure.database.resource_repository import (
ResourceRepository,
)
# Crockford's Base32 alphabet for ULID generation
_CROCKFORD_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
def _generate_ulid() -> str:
"""Generate a ULID identifier."""
ts_ms = int(time.time() * 1000)
ts_chars: list[str] = []
for _ in range(10):
ts_chars.append(_CROCKFORD_ALPHABET[ts_ms & 0x1F])
ts_ms >>= 5
ts_chars.reverse()
rand_chars: list[str] = []
for _ in range(16):
rand_chars.append(_CROCKFORD_ALPHABET[random.randint(0, 31)])
return "".join(ts_chars) + "".join(rand_chars)
class ProjectServicePersistent:
"""Persistence-backed project service.
Uses ProjectRepository, ResourceRepository, and
ProjectResourceLinkRepository for all operations.
"""
def __init__(
self,
project_repo: ProjectRepository,
resource_repo: ResourceRepository,
link_repo: ProjectResourceLinkRepository,
session: Session,
) -> None:
"""Initialize with repositories and session.
Args:
project_repo: Repository for project persistence.
resource_repo: Repository for resource persistence.
link_repo: Repository for project-resource links.
session: Database session for transaction management.
"""
self._project_repo = project_repo
self._resource_repo = resource_repo
self._link_repo = link_repo
self._session = session
def create(
self,
name: str,
namespace: str = "local",
description: str | None = None,
path: str | None = None,
tags: list[str] | None = None,
validation_config: dict[str, Any] | None = None,
context_config: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Create a new project.
Args:
name: Project name.
namespace: Project namespace (default: 'local').
description: Optional project description.
path: Optional project root path.
tags: Optional list of tags.
validation_config: Optional validation config dict.
context_config: Optional context config dict.
Returns:
Dictionary with created project details.
Raises:
ValueError: If project name already exists.
"""
existing = self._project_repo.get_by_name(name)
if existing is not None:
raise ValueError(f"Project '{name}' already exists")
vc = ValidationConfig(**validation_config) if validation_config else None
cc = ContextConfig(**context_config) if context_config else ContextConfig()
project = Project(
id=None,
name=name,
namespace=namespace,
description=description,
path=Path(path) if path else Path("/tmp/projects") / name,
settings=ProjectSettings(
auto_build=False,
auto_apply=False,
confirm_apply=True,
max_context_size=52428800,
default_model="mock-gpt",
),
project_id=_generate_ulid(),
tags=tags or [],
validation_config=vc,
context_config=cc,
created_at=datetime.now(),
updated_at=datetime.now(),
current_plan_id=None,
)
project = self._project_repo.create(project)
self._session.commit()
return self._project_to_dict(project)
def list_all(self) -> list[dict[str, Any]]:
"""List all projects.
Returns:
List of project dictionaries.
"""
projects = self._project_repo.get_all()
return [self._project_to_dict(p) for p in projects]
def list_by_namespace(self, namespace: str) -> list[dict[str, Any]]:
"""List projects filtered by namespace.
Args:
namespace: Namespace to filter by.
Returns:
List of project dictionaries.
"""
projects = self._project_repo.get_by_namespace(namespace)
return [self._project_to_dict(p) for p in projects]
def show(self, name: str) -> dict[str, Any]:
"""Show a project by name.
Args:
name: Project name.
Returns:
Project dictionary.
Raises:
ValueError: If project not found.
"""
project = self._project_repo.get_by_name(name)
if project is None:
raise ValueError(f"Project '{name}' not found")
return self._project_to_dict(project)
def delete(self, name: str) -> None:
"""Delete a project by name.
Args:
name: Project name.
Raises:
ValueError: If project not found.
"""
project = self._project_repo.get_by_name(name)
if project is None:
raise ValueError(f"Project '{name}' not found")
self._project_repo.delete(project.id) # type: ignore[arg-type]
self._session.commit()
def register_resource(
self,
name: str,
resource_type: str,
location: str,
sandbox_strategy: str = "none",
read_only: bool = False,
) -> dict[str, Any]:
"""Register a resource (convenience method).
Args:
name: Resource name.
resource_type: Resource type string.
location: Resource location.
sandbox_strategy: Sandbox strategy.
read_only: Whether resource is read-only.
Returns:
Dictionary with resource details.
Raises:
ValueError: If resource already exists or type is invalid.
"""
existing = self._resource_repo.get_by_name(name)
if existing is not None:
raise ValueError(f"Resource '{name}' already exists")
rtype = ResourceType(resource_type)
strategy = SandboxStrategy(sandbox_strategy)
resource = Resource(
resource_id=_generate_ulid(),
name=name,
type=rtype,
location=location,
is_remote=False,
sandbox_strategy=strategy,
read_only=read_only,
)
self._resource_repo.create(resource)
self._session.commit()
return {
"resource_id": resource.resource_id,
"name": resource.name,
"type": resource.type.value,
"location": resource.location,
}
def link_resource(
self,
project_name: str,
resource_name: str,
alias: str | None = None,
read_only: bool = False,
) -> dict[str, Any]:
"""Link a resource to a project.
Args:
project_name: Project name.
resource_name: Resource name.
alias: Optional alias for the link.
read_only: Whether the link is read-only.
Returns:
Dictionary with link details.
Raises:
ValueError: If project or resource not found, or already linked.
"""
project = self._project_repo.get_by_name(project_name)
if project is None:
raise ValueError(f"Project '{project_name}' not found")
resource = self._resource_repo.get_by_name(resource_name)
if resource is None:
raise ValueError(f"Resource '{resource_name}' not found")
result = self._link_repo.link(
project_id=project.id, # type: ignore[arg-type]
resource_id=resource.resource_id,
alias=alias,
read_only=read_only,
)
self._session.commit()
return result
def unlink_resource(self, project_name: str, resource_name: str) -> bool:
"""Unlink a resource from a project.
Args:
project_name: Project name.
resource_name: Resource name.
Returns:
True if unlinked, False if link not found.
Raises:
ValueError: If project or resource not found.
"""
project = self._project_repo.get_by_name(project_name)
if project is None:
raise ValueError(f"Project '{project_name}' not found")
resource = self._resource_repo.get_by_name(resource_name)
if resource is None:
raise ValueError(f"Resource '{resource_name}' not found")
result = self._link_repo.unlink(
project_id=project.id, # type: ignore[arg-type]
resource_id=resource.resource_id,
)
self._session.commit()
return result
def get_links(self, project_name: str) -> list[dict[str, Any]]:
"""Get all resource links for a project.
Args:
project_name: Project name.
Returns:
List of link dictionaries.
Raises:
ValueError: If project not found.
"""
project = self._project_repo.get_by_name(project_name)
if project is None:
raise ValueError(f"Project '{project_name}' not found")
return self._link_repo.list_links(
project.id # type: ignore[arg-type]
)
def get_validation_summary(self, project_name: str) -> dict[str, Any] | None:
"""Get validation attachment summary for a project.
Args:
project_name: Project name.
Returns:
Dictionary with validation config and linked resources.
Raises:
ValueError: If project not found.
"""
project = self._project_repo.get_by_name(project_name)
if project is None:
raise ValueError(f"Project '{project_name}' not found")
return self._link_repo.get_validation_summary(
project.id # type: ignore[arg-type]
)
@staticmethod
def _project_to_dict(project: Project) -> dict[str, Any]:
"""Convert a Project domain object to a dictionary.
Args:
project: Project to convert.
Returns:
Dictionary representation.
"""
return {
"id": project.id,
"name": project.name,
"namespace": project.namespace,
"description": project.description,
"path": str(project.path),
"project_id": project.project_id,
"tags": project.tags,
"validation_config": (
project.validation_config.model_dump()
if project.validation_config
else None
),
"context_config": project.context_config.model_dump(),
"created_at": (
project.created_at.isoformat() if project.created_at else None
),
"updated_at": (
project.updated_at.isoformat() if project.updated_at else None
),
}
@@ -0,0 +1,271 @@
"""Resource Registry Service for CleverAgents.
Provides a persistence-backed service layer for resource registration,
removal, inspection, tree traversal, and auto-discovery.
Based on implementation_plan.md B5 Item 4 and ADR-007 (Repository Pattern).
"""
import random
import time
from typing import Any
from sqlalchemy.orm import Session
from cleveragents.domain.models.core.resource import (
Resource,
ResourceType,
SandboxStrategy,
)
from cleveragents.infrastructure.database.resource_repository import (
ResourceRepository,
)
# Crockford's Base32 alphabet for ULID generation
_CROCKFORD_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
def _generate_ulid() -> str:
"""Generate a ULID (Universally Unique Lexicographically Sortable ID).
Produces a 26-character Crockford Base32 string:
10 chars from timestamp (48-bit ms) + 16 chars random (80-bit).
"""
ts_ms = int(time.time() * 1000)
ts_chars: list[str] = []
for _ in range(10):
ts_chars.append(_CROCKFORD_ALPHABET[ts_ms & 0x1F])
ts_ms >>= 5
ts_chars.reverse()
rand_chars: list[str] = []
for _ in range(16):
rand_chars.append(_CROCKFORD_ALPHABET[random.randint(0, 31)])
return "".join(ts_chars) + "".join(rand_chars)
class ResourceRegistryService:
"""Service for managing resource registration and discovery.
Wraps ResourceRepository with higher-level operations including
name/ULID resolution, auto-discovery, and tree traversal.
"""
def __init__(
self,
resource_repo: ResourceRepository,
session: Session,
) -> None:
"""Initialize the service.
Args:
resource_repo: Repository for resource persistence.
session: Database session for transaction management.
"""
self._repo = resource_repo
self._session = session
def register(
self,
name: str,
resource_type: str,
location: str,
sandbox_strategy: str = "none",
read_only: bool = False,
is_remote: bool = False,
metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Register a new resource.
Args:
name: Human-readable resource name.
resource_type: Resource type string (must match ResourceType enum).
location: Path or URI to the resource.
sandbox_strategy: Sandbox strategy string (default: "none").
read_only: Whether the resource is read-only.
is_remote: Whether the resource is remote.
metadata: Optional key-value metadata.
Returns:
Dictionary with the registered resource details.
Raises:
ValueError: If resource name already exists or type is invalid.
"""
# Check for duplicate name
existing = self._repo.get_by_name(name)
if existing is not None:
raise ValueError(f"Resource '{name}' already exists")
# Validate resource type
try:
rtype = ResourceType(resource_type)
except ValueError as exc:
valid = [t.value for t in ResourceType]
raise ValueError(
f"Invalid resource type '{resource_type}'. "
f"Valid types: {', '.join(valid)}"
) from exc
# Validate sandbox strategy
try:
strategy = SandboxStrategy(sandbox_strategy)
except ValueError as exc:
valid = [s.value for s in SandboxStrategy]
raise ValueError(
f"Invalid sandbox strategy '{sandbox_strategy}'. "
f"Valid strategies: {', '.join(valid)}"
) from exc
resource_id = _generate_ulid()
resource = Resource(
resource_id=resource_id,
name=name,
type=rtype,
location=location,
sandbox_strategy=strategy,
read_only=read_only,
is_remote=is_remote,
metadata=metadata or {},
)
self._repo.create(resource)
self._session.commit()
return self._resource_to_dict(resource)
def remove(self, name: str) -> None:
"""Remove a resource by name.
Args:
name: Resource name.
Raises:
ValueError: If resource not found.
"""
resource = self._resolve(name)
self._repo.delete(resource.resource_id)
self._session.commit()
def show(self, name: str) -> dict[str, Any]:
"""Show a resource by name or ULID.
Args:
name: Resource name or ULID.
Returns:
Dictionary with resource details.
Raises:
ValueError: If resource not found.
"""
resource = self._resolve(name)
return self._resource_to_dict(resource)
def tree(self, name: str) -> list[dict[str, Any]]:
"""Get the full subtree rooted at a resource.
Args:
name: Resource name or ULID of the root.
Returns:
List of resource dictionaries in the subtree.
Raises:
ValueError: If root resource not found.
"""
root = self._resolve(name)
subtree = self._repo.get_subtree(root.resource_id)
return [self._resource_to_dict(r) for r in subtree]
def link_child(self, parent_name: str, child_name: str) -> None:
"""Link a child resource to a parent resource.
Args:
parent_name: Name or ULID of the parent resource.
child_name: Name or ULID of the child resource.
Raises:
ValueError: If either resource not found, or edge invalid.
"""
parent = self._resolve(parent_name)
child = self._resolve(child_name)
self._repo.add_edge(parent.resource_id, child.resource_id)
self._session.commit()
def auto_discover(
self,
location: str,
name: str | None = None,
) -> dict[str, Any]:
"""Auto-discover and register a resource from a location.
For MVP, assumes git-checkout type for any directory.
Args:
location: Path to discover.
name: Optional name override (defaults to location basename).
Returns:
Dictionary with the registered resource details.
"""
if name is None:
# Derive name from location path
from pathlib import Path
name = Path(location).name
# MVP: always register as git_repository
return self.register(
name=name,
resource_type=ResourceType.GIT_REPOSITORY.value,
location=location,
sandbox_strategy=SandboxStrategy.GIT_WORKTREE.value,
)
def _resolve(self, name_or_id: str) -> Resource:
"""Resolve a resource by name or ULID.
Args:
name_or_id: Resource name or ULID identifier.
Returns:
The resolved Resource domain object.
Raises:
ValueError: If resource not found.
"""
# Try by name first
resource = self._repo.get_by_name(name_or_id)
if resource is not None:
return resource
# Try by ULID
resource = self._repo.get_by_id(name_or_id)
if resource is not None:
return resource
raise ValueError(f"Resource '{name_or_id}' not found")
@staticmethod
def _resource_to_dict(resource: Resource) -> dict[str, Any]:
"""Convert a Resource domain object to a dictionary.
Args:
resource: Resource to convert.
Returns:
Dictionary representation.
"""
return {
"resource_id": resource.resource_id,
"name": resource.name,
"type": resource.type.value,
"location": resource.location,
"is_remote": resource.is_remote,
"sandbox_strategy": resource.sandbox_strategy.value,
"read_only": resource.read_only,
"metadata": dict(resource.metadata),
"created_at": resource.created_at.isoformat(),
}
@@ -50,9 +50,16 @@ from cleveragents.domain.models.core.plan_legacy import (
PlanStatus,
)
from cleveragents.domain.models.core.project import (
ContextConfig,
Project,
ProjectSettings,
ProjectStats,
ValidationConfig,
)
from cleveragents.domain.models.core.resource import (
Resource,
ResourceType,
SandboxStrategy,
)
__all__ = [
@@ -62,6 +69,7 @@ __all__ = [
"ChangeSet",
"CloudBillingFields",
"Context",
"ContextConfig",
"ContextFile",
"ContextType",
"ContextUpdateResult",
@@ -92,7 +100,11 @@ __all__ = [
"ProjectLink",
"ProjectSettings",
"ProjectStats",
"Resource",
"ResourceType",
"SandboxStrategy",
"SummaryForUpdateContextParams",
"User",
"ValidationConfig",
"can_transition",
]
+280 -16
View File
@@ -1,26 +1,158 @@
"""Project domain model for CleverAgents.
Based on Phase 0 discovery and ADR-004 (Pydantic Validation).
Includes ValidationConfig (B1.5), ContextConfig (B1.6), and the Project model (B1.1).
Based on Phase 0 discovery, implementation_plan.md, and ADR-004 (Pydantic Validation).
"""
import re
from datetime import datetime
from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field, field_validator
from cleveragents.domain.models.core.resource import Resource
ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
NAMESPACE_PATTERN = r"^(local|[a-z][a-z0-9_]{0,49})$"
RESERVED_NAMESPACES = frozenset({"system", "internal", "admin", "root"})
DEFAULT_IGNORE_PATTERNS: list[str] = [
".git/",
"node_modules/",
"__pycache__/",
".venv/",
"*.pyc",
".DS_Store",
]
class ValidationConfig(BaseModel):
"""Configuration for project validation commands.
Defines test, lint, type-check, and build commands along with
custom commands and execution settings.
Implements B1.5 from the implementation plan.
"""
test_command: str | None = Field(default=None, description="Command to run tests")
lint_command: str | None = Field(default=None, description="Command to run linting")
type_check_command: str | None = Field(
default=None, description="Command to run type checking"
)
build_command: str | None = Field(
default=None, description="Command to build the project"
)
custom_commands: dict[str, str] = Field(
default_factory=dict,
description="Custom named validation commands",
)
timeout_seconds: int = Field(
default=300, description="Timeout for each validation command in seconds"
)
fail_on_lint_error: bool = Field(
default=True, description="Whether lint errors should fail validation"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
def get_all_commands(self) -> dict[str, str]:
"""Return all configured commands as a name->command dict.
Returns:
Dictionary mapping command names to command strings.
Includes standard commands (test, lint, type_check, build)
and any custom commands.
"""
commands: dict[str, str] = {}
if self.test_command is not None:
commands["test"] = self.test_command
if self.lint_command is not None:
commands["lint"] = self.lint_command
if self.type_check_command is not None:
commands["type_check"] = self.type_check_command
if self.build_command is not None:
commands["build"] = self.build_command
commands.update(self.custom_commands)
return commands
def has_any_validation(self) -> bool:
"""Check if any validation commands are configured.
Returns:
True if at least one command (standard or custom) is set.
"""
return len(self.get_all_commands()) > 0
class ContextConfig(BaseModel):
"""Configuration for project context indexing and filtering.
Controls how project files are indexed, chunked, and filtered
for inclusion in AI context windows.
Implements B1.6 from the implementation plan.
"""
ignore_patterns: list[str] = Field(
default_factory=lambda: list(DEFAULT_IGNORE_PATTERNS),
description="File patterns to ignore during indexing",
)
include_patterns: list[str] | None = Field(
default=None, description="File patterns to include (None means include all)"
)
max_file_size: int = Field(
default=1_000_000, description="Maximum file size in bytes (1MB default)"
)
max_files: int = Field(
default=100_000, description="Maximum number of files to index"
)
indexing_strategy: str = Field(
default="full_text",
description="Indexing strategy: full_text, semantic, etc.",
)
chunking_policy: str = Field(
default="smart", description="Chunking policy: smart, fixed, etc."
)
chunk_size: int = Field(default=1000, description="Chunk size in tokens")
@field_validator("ignore_patterns")
@classmethod
def merge_default_ignore_patterns(
cls: "type[ContextConfig]", v: list[str]
) -> list[str]:
"""Merge user-provided ignore patterns with defaults.
Ensures default patterns like .git/, node_modules/, etc.
are always included.
"""
merged = list(DEFAULT_IGNORE_PATTERNS)
for pattern in v:
if pattern not in merged:
merged.append(pattern)
return merged
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class ProjectSettings(BaseModel):
"""Project-specific settings and configuration."""
auto_build: bool = Field(False, description="Automatically build plans")
auto_apply: bool = Field(False, description="Automatically apply changes")
auto_build: bool = Field(default=False, description="Automatically build plans")
auto_apply: bool = Field(default=False, description="Automatically apply changes")
confirm_apply: bool = Field(
True, description="Require confirmation before applying"
default=True, description="Require confirmation before applying"
)
max_context_size: int = Field(
52428800, description="Maximum context size in bytes (50MB)"
default=52428800, description="Maximum context size in bytes (50MB)"
)
default_model: str = Field("mock-gpt", description="Default AI model")
default_model: str = Field(default="mock-gpt", description="Default AI model")
include_paths: list[str] = Field(
default_factory=list, description="Relative include globs"
)
@@ -37,10 +169,10 @@ class ProjectSettings(BaseModel):
class ProjectStats(BaseModel):
"""Statistics for a project."""
plans: int = Field(0, ge=0)
context_files: int = Field(0, ge=0)
changes: int = Field(0, ge=0)
current_plan: str | None = Field(None)
plans: int = Field(default=0, ge=0)
context_files: int = Field(default=0, ge=0)
changes: int = Field(default=0, ge=0)
current_plan: str | None = Field(default=None)
model_config = ConfigDict(
str_strip_whitespace=True,
@@ -52,19 +184,52 @@ class Project(BaseModel):
"""Domain model for a CleverAgents project.
A project represents a workspace with plans, contexts, and changes.
Projects can have resources, validation configs, and context configs.
Extends original model with B1.1 fields: project_id, namespace,
description, tags, resources, validation_config, context_config.
"""
id: int | None = Field(None, description="Project ID")
name: str = Field(..., min_length=1, max_length=255)
# Legacy fields (preserved for backward compatibility)
id: int | None = Field(default=None, description="Legacy integer project ID")
path: Path = Field(..., description="Project root path")
settings: ProjectSettings = Field(default_factory=lambda: ProjectSettings())
current_plan_id: int | None = Field(default=None)
# B1.1b - Identity fields
project_id: str | None = Field(
default=None, description="Unique ULID identifier", pattern=ULID_PATTERN
)
name: str = Field(..., min_length=1, max_length=255)
namespace: str = Field(
default="local", description="Project namespace for grouping"
)
description: str | None = Field(
default=None, description="Human-readable project description"
)
# B1.1c - Categorization fields
tags: list[str] = Field(
default_factory=list, description="Project tags for filtering"
)
resources: list[Resource] = Field(
default_factory=list, description="Project resource references"
)
validation_config: ValidationConfig | None = Field(
default=None, description="Validation command configuration"
)
context_config: ContextConfig = Field(
default_factory=lambda: ContextConfig(),
description="Context indexing and filtering configuration",
)
# B1.1d - Timestamps
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
settings: ProjectSettings = Field(default_factory=lambda: ProjectSettings()) # type: ignore
current_plan_id: int | None = Field(None)
@field_validator("name")
@classmethod
def validate_name(cls: type["Project"], v: str) -> str:
def validate_name(cls: "type[Project]", v: str) -> str:
"""Validate project name."""
if not v.replace("-", "").replace("_", "").replace(" ", "").isalnum():
raise ValueError(
@@ -72,9 +237,31 @@ class Project(BaseModel):
)
return v
@field_validator("namespace")
@classmethod
def validate_namespace(cls: "type[Project]", v: str) -> str:
"""Validate namespace format and reject reserved names.
Namespace must match pattern: 'local' or start with lowercase letter
followed by up to 49 lowercase alphanumeric/underscore chars.
Reserved names (system, internal, admin, root) are rejected.
"""
if not re.match(NAMESPACE_PATTERN, v):
raise ValueError(
f"Namespace '{v}' must match pattern: 'local' or "
"start with lowercase letter followed by lowercase "
"alphanumeric/underscore characters (max 50 chars)"
)
if v in RESERVED_NAMESPACES:
raise ValueError(
f"Namespace '{v}' is reserved. "
f"Reserved namespaces: {', '.join(sorted(RESERVED_NAMESPACES))}"
)
return v
@field_validator("path")
@classmethod
def validate_path(cls: type["Project"], v: Path) -> Path:
def validate_path(cls: "type[Project]", v: Path) -> Path:
"""Ensure path is absolute."""
return v.resolve()
@@ -83,3 +270,80 @@ class Project(BaseModel):
validate_assignment=True,
arbitrary_types_allowed=True, # Allow Path
)
# B1.1e - Computed property
@property
def is_remote(self) -> bool:
"""Check if all project resources are remote.
A project is remote only if it has resources and ALL of them
are remotely accessible. Empty resource lists return False.
"""
if not self.resources:
return False
return all(r.is_remote for r in self.resources)
# B1.1g - Helper methods
@property
def namespaced_name(self) -> str:
"""Return the fully qualified namespace/name string."""
return f"{self.namespace}/{self.name}"
@staticmethod
def parse_namespaced_name(namespaced: str) -> tuple[str, str]:
"""Parse a 'namespace/name' string into (namespace, name).
Args:
namespaced: String in 'namespace/name' format.
Returns:
Tuple of (namespace, name).
Raises:
ValueError: If the string doesn't contain exactly one '/'.
"""
parts = namespaced.split("/", 1)
if len(parts) != 2:
raise ValueError(f"Expected 'namespace/name' format, got '{namespaced}'")
return parts[0], parts[1]
def add_resource(self, resource: Resource) -> "Project":
"""Return a new Project with the resource added.
Since the model may be frozen or validated, this returns
a new instance with the resource appended.
Args:
resource: The Resource to add.
Returns:
New Project instance with the resource added.
"""
new_resources = [*list(self.resources), resource]
return self.model_copy(update={"resources": new_resources})
def remove_resource(self, name: str) -> "Project":
"""Return a new Project with the named resource removed.
Args:
name: Name of the resource to remove.
Returns:
New Project instance without the named resource.
"""
new_resources = [r for r in self.resources if r.name != name]
return self.model_copy(update={"resources": new_resources})
def get_resource(self, name: str) -> Resource | None:
"""Get a resource by name.
Args:
name: Name of the resource to find.
Returns:
The Resource if found, None otherwise.
"""
for r in self.resources:
if r.name == name:
return r
return None
@@ -0,0 +1,139 @@
"""Resource domain model for CleverAgents.
Defines ResourceType enum, SandboxStrategy enum, and the Resource
Pydantic model for classifying project resources and determining
sandboxing behavior.
Based on implementation_plan.md (B1.2, B1.3, B1.4) and ADR-004 (Pydantic Validation).
"""
from __future__ import annotations
from datetime import datetime
from enum import StrEnum
from pathlib import Path
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
ULID_PATTERN = r"^[0-9A-HJKMNP-TV-Z]{26}$"
class ResourceType(StrEnum):
"""Types of resources a project can reference.
Each resource type maps to a category of external data source
or infrastructure that a project operates on.
"""
GIT_REPOSITORY = "git_repository"
FILESYSTEM = "filesystem"
DATABASE = "database"
API_ENDPOINT = "api_endpoint"
DOCUMENT_CORPUS = "document_corpus"
CLOUD_INFRASTRUCTURE = "cloud_infrastructure"
class SandboxStrategy(StrEnum):
"""Strategies for isolating resource modifications during plan execution.
Sandboxing ensures that changes made during plan execution can be
reviewed, committed, or rolled back without affecting the original resource.
"""
GIT_WORKTREE = "git_worktree"
COPY_ON_WRITE = "copy_on_write"
OVERLAY = "overlay"
TRANSACTION_ROLLBACK = "transaction_rollback"
VERSIONING = "versioning"
NONE = "none"
@property
def supports_rollback(self) -> bool:
"""Check if this strategy supports rolling back changes.
All strategies except NONE support rollback, since NONE means
modifications are immediate and irreversible.
"""
return self != SandboxStrategy.NONE
@property
def is_copy_based(self) -> bool:
"""Check if this strategy creates a copy of the resource.
COPY_ON_WRITE and OVERLAY both duplicate the resource data
to a separate location for isolation.
"""
return self in (SandboxStrategy.COPY_ON_WRITE, SandboxStrategy.OVERLAY)
class Resource(BaseModel):
"""A project resource reference.
Resources represent external data sources or repositories
that a project operates on. Each resource has a type, location,
and sandboxing strategy that determines how modifications are isolated.
Implements B1.2 from the implementation plan.
"""
resource_id: str = Field(
..., description="Unique ULID identifier", pattern=ULID_PATTERN
)
name: str = Field(
..., min_length=1, max_length=255, description="Human-readable name"
)
type: ResourceType = Field(..., description="Resource classification")
location: str = Field(..., min_length=1, description="Path or URI to the resource")
is_remote: bool = Field(
False, description="Whether the resource is remotely accessible"
)
sandbox_strategy: SandboxStrategy = Field(
SandboxStrategy.NONE,
description="Strategy for isolating modifications",
)
read_only: bool = Field(False, description="Whether the resource is read-only")
metadata: dict[str, Any] = Field(
default_factory=dict,
description="Arbitrary key-value metadata for the resource",
)
created_at: datetime = Field(
default_factory=datetime.now,
description="Timestamp when the resource was created",
)
@field_validator("name")
@classmethod
def validate_name(cls: type[Resource], v: str) -> str:
"""Enforce naming rules: alphanumeric with hyphens/underscores, lowercased."""
if not v.replace("-", "").replace("_", "").isalnum():
raise ValueError("Name must be alphanumeric with hyphens/underscores only")
return v.lower()
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
use_enum_values=False,
frozen=True,
)
@property
def supports_sandbox(self) -> bool:
"""Check if this resource has a sandbox strategy other than NONE."""
return self.sandbox_strategy != SandboxStrategy.NONE
@property
def can_write(self) -> bool:
"""Check if this resource allows write operations."""
return not self.read_only
def get_sandbox_path(self, base_path: Path) -> Path:
"""Get the sandbox directory path for this resource.
Args:
base_path: The base directory for sandboxes.
Returns:
Path combining base_path with the resource name.
"""
return base_path / self.name
@@ -0,0 +1,220 @@
"""Project-resource link repository for CleverAgents.
Manages the relationship between projects and resources,
including link/unlink operations and validation attachment summaries.
Based on ADR-007 (Repository Pattern) and implementation_plan.md B5 Item 3.
"""
from datetime import datetime
from typing import Any
from sqlalchemy.orm import Session
from cleveragents.infrastructure.database.models import (
ProjectModel,
ProjectResourceLinkModel,
ResourceModel,
)
def _generate_ulid() -> str:
"""Generate a simple ULID-like identifier for link IDs.
Uses timestamp + random bytes encoded as Crockford Base32.
"""
import random
import time
# Crockford's Base32 alphabet
alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
# 10 chars from timestamp (48-bit ms)
ts_ms = int(time.time() * 1000)
ts_chars: list[str] = []
for _ in range(10):
ts_chars.append(alphabet[ts_ms & 0x1F])
ts_ms >>= 5
ts_chars.reverse()
# 16 chars from random (80-bit)
rand_chars: list[str] = []
for _ in range(16):
rand_chars.append(alphabet[random.randint(0, 31)])
return "".join(ts_chars) + "".join(rand_chars)
class ProjectResourceLinkRepository:
"""Repository for project-resource link management."""
def __init__(self, session: Session) -> None:
"""Initialize repository with database session."""
self.session = session
def link(
self,
project_id: int,
resource_id: str,
alias: str | None = None,
read_only: bool = False,
) -> dict[str, Any]:
"""Link a resource to a project.
Args:
project_id: The project's database ID.
resource_id: The resource's ULID.
alias: Optional alias for the link.
read_only: Whether the link is read-only.
Returns:
Dictionary with link details.
Raises:
ValueError: If the resource is already linked to the project.
"""
# Check for duplicate link
existing = (
self.session.query(ProjectResourceLinkModel)
.filter_by(project_id=project_id, resource_id=resource_id)
.first()
)
if existing:
raise ValueError(
f"Resource '{resource_id}' is already linked to project '{project_id}'"
)
link_id = _generate_ulid()
link_model = ProjectResourceLinkModel(
link_id=link_id,
project_id=project_id,
resource_id=resource_id,
alias=alias,
read_only=read_only,
created_at=datetime.now(),
)
self.session.add(link_model)
self.session.flush()
return {
"link_id": link_id,
"project_id": project_id,
"resource_id": resource_id,
"alias": alias,
"read_only": read_only,
}
def unlink(self, project_id: int, resource_id: str) -> bool:
"""Remove a link between a project and a resource.
Args:
project_id: The project's database ID.
resource_id: The resource's ULID.
Returns:
True if a link was removed, False if none found.
"""
count = (
self.session.query(ProjectResourceLinkModel)
.filter_by(project_id=project_id, resource_id=resource_id)
.delete()
)
self.session.flush()
return count > 0
def list_links(self, project_id: int) -> list[dict[str, Any]]:
"""List all resource links for a project.
Args:
project_id: The project's database ID.
Returns:
List of dictionaries with link details.
"""
links = (
self.session.query(ProjectResourceLinkModel)
.filter_by(project_id=project_id)
.all()
)
return [
{
"link_id": link.link_id, # type: ignore[union-attr]
"project_id": link.project_id, # type: ignore[union-attr]
"resource_id": link.resource_id, # type: ignore[union-attr]
"alias": link.alias, # type: ignore[union-attr]
"read_only": link.read_only, # type: ignore[union-attr]
}
for link in links
]
def list_links_with_details(self, project_id: int) -> list[dict[str, Any]]:
"""List links with full resource details.
Joins the link table with the resources table to include
resource name, type, and location in the results.
Args:
project_id: The project's database ID.
Returns:
List of dictionaries with link and resource details.
"""
results = (
self.session.query(ProjectResourceLinkModel, ResourceModel)
.join(
ResourceModel,
ProjectResourceLinkModel.resource_id == ResourceModel.resource_id,
)
.filter(ProjectResourceLinkModel.project_id == project_id)
.all()
)
return [
{
"link_id": link.link_id, # type: ignore[union-attr]
"project_id": link.project_id, # type: ignore[union-attr]
"resource_id": link.resource_id, # type: ignore[union-attr]
"alias": link.alias, # type: ignore[union-attr]
"read_only": link.read_only, # type: ignore[union-attr]
"resource_name": resource.name, # type: ignore[union-attr]
"resource_type": resource.type, # type: ignore[union-attr]
"resource_location": resource.location, # type: ignore[union-attr]
}
for link, resource in results
]
def get_validation_summary(self, project_id: int) -> dict[str, Any] | None:
"""Get validation attachment summary for a project.
Returns the project's validation config alongside a list of
linked resources, useful for understanding what validations
apply to which resources.
Args:
project_id: The project's database ID.
Returns:
Dictionary with validation_config and linked_resources,
or None if project not found.
"""
db_project = self.session.query(ProjectModel).filter_by(id=project_id).first()
if not db_project:
return None
vc_json: dict[str, Any] | None = db_project.validation_config_json # type: ignore[assignment]
linked = self.list_links_with_details(project_id)
return {
"project_id": project_id,
"project_name": db_project.name, # type: ignore[union-attr]
"validation_config": vc_json,
"linked_resources": [
{
"resource_id": link["resource_id"],
"resource_name": link["resource_name"],
"alias": link["alias"],
"read_only": link["read_only"],
}
for link in linked
],
}
@@ -29,6 +29,10 @@ from cleveragents.domain.models.core import (
Project,
ProjectSettings,
)
from cleveragents.domain.models.core.project import (
ContextConfig,
ValidationConfig,
)
from cleveragents.infrastructure.database.models import (
ActorModel,
ChangeModel,
@@ -55,6 +59,16 @@ class ProjectRepository:
name=project.name,
path=str(project.path),
settings=project.settings.model_dump(),
project_id=project.project_id,
namespace=project.namespace,
description=project.description,
tags_json=project.tags,
validation_config_json=(
project.validation_config.model_dump()
if project.validation_config
else None
),
context_config_json=project.context_config.model_dump(),
)
self.session.add(db_project)
@@ -67,6 +81,34 @@ class ProjectRepository:
self.session.rollback()
raise DatabaseError(f"Failed to create project: {e}") from e
def _to_domain(self, db_project: ProjectModel) -> Project:
"""Convert a database ProjectModel to a domain Project."""
# Reconstruct validation config from JSON
validation_config = None
vc_json: dict[str, Any] | None = db_project.validation_config_json # type: ignore[assignment]
if vc_json:
validation_config = ValidationConfig(**vc_json)
# Reconstruct context config from JSON
cc_json: dict[str, Any] | None = db_project.context_config_json # type: ignore[assignment]
context_config = ContextConfig(**cc_json) if cc_json else ContextConfig()
return Project(
id=db_project.id, # type: ignore
name=db_project.name, # type: ignore
path=Path(db_project.path), # type: ignore
created_at=db_project.created_at, # type: ignore
updated_at=db_project.updated_at, # type: ignore
settings=ProjectSettings(**db_project.settings), # type: ignore
current_plan_id=db_project.current_plan_id, # type: ignore
project_id=db_project.project_id, # type: ignore
namespace=db_project.namespace or "local", # type: ignore
description=db_project.description, # type: ignore
tags=db_project.tags_json or [], # type: ignore
validation_config=validation_config,
context_config=context_config,
)
@database_retry
def get_by_id(self, project_id: int) -> Project | None:
"""Get project by ID with retry logic."""
@@ -78,15 +120,7 @@ class ProjectRepository:
if not db_project:
return None
return Project(
id=db_project.id, # type: ignore
name=db_project.name, # type: ignore
path=Path(db_project.path), # type: ignore
created_at=db_project.created_at, # type: ignore
updated_at=db_project.updated_at, # type: ignore
settings=ProjectSettings(**db_project.settings), # type: ignore
current_plan_id=db_project.current_plan_id, # type: ignore
)
return self._to_domain(db_project)
except (OperationalError, SQLAlchemyDatabaseError) as e:
raise DatabaseError(f"Failed to get project {project_id}: {e}") from e
@@ -97,32 +131,12 @@ class ProjectRepository:
if not db_project:
return None
return Project(
id=db_project.id, # type: ignore
name=db_project.name, # type: ignore
path=Path(db_project.path), # type: ignore
created_at=db_project.created_at, # type: ignore
updated_at=db_project.updated_at, # type: ignore
settings=ProjectSettings(**db_project.settings), # type: ignore
current_plan_id=db_project.current_plan_id, # type: ignore
)
return self._to_domain(db_project)
def get_all(self) -> list[Project]:
"""Get all projects."""
db_projects = self.session.query(ProjectModel).all()
return [
Project(
id=p.id, # type: ignore
name=p.name, # type: ignore
path=Path(p.path), # type: ignore
created_at=p.created_at, # type: ignore
updated_at=p.updated_at, # type: ignore
settings=ProjectSettings(**p.settings), # type: ignore
current_plan_id=p.current_plan_id, # type: ignore
)
for p in db_projects
]
return [self._to_domain(p) for p in db_projects]
def update(self, project: Project) -> Project:
"""Update an existing project."""
@@ -134,11 +148,42 @@ class ProjectRepository:
db_project.settings = project.settings.model_dump() # type: ignore
db_project.current_plan_id = project.current_plan_id # type: ignore
db_project.updated_at = datetime.now() # type: ignore
# B1.1 fields
db_project.project_id = project.project_id # type: ignore
db_project.namespace = project.namespace # type: ignore
db_project.description = project.description # type: ignore
db_project.tags_json = project.tags # type: ignore
db_project.validation_config_json = ( # type: ignore
project.validation_config.model_dump()
if project.validation_config
else None
)
db_project.context_config_json = ( # type: ignore
project.context_config.model_dump()
)
self.session.flush()
return project
def get_by_namespace(self, namespace: str) -> list[Project]:
"""Get all projects in a given namespace."""
db_projects = (
self.session.query(ProjectModel).filter_by(namespace=namespace).all()
)
return [self._to_domain(p) for p in db_projects]
def get_by_namespace_and_name(self, namespace: str, name: str) -> Project | None:
"""Get a project by its namespace and name combination."""
db_project = (
self.session.query(ProjectModel)
.filter_by(namespace=namespace, name=name)
.first()
)
if not db_project:
return None
return self._to_domain(db_project)
def delete(self, project_id: int) -> None:
"""Delete a project by ID."""
db_project = self.session.query(ProjectModel).filter_by(id=project_id).first()