Files
cleveragents-core/features/steps/namespaced_project_model_steps.py
T
khyari hamza f89c60595f
CI / lint (pull_request) Successful in 15s
CI / typecheck (pull_request) Successful in 26s
CI / security (pull_request) Successful in 21s
CI / quality (pull_request) Successful in 15s
CI / integration_tests (pull_request) Successful in 4m21s
CI / build (pull_request) Successful in 16s
CI / unit_tests (pull_request) Successful in 8m19s
CI / docker (pull_request) Successful in 40s
CI / coverage (pull_request) Successful in 6m29s
feat(domain): add spec-aligned Resource and Project models
Implement the B1 (Project Data Models) domain models from scratch,
aligned with docs/specification.md:

- Resource model: ULID PK, PhysVirt enum (physical|virtual),
  SandboxStrategy enum (5 spec values), ResourceCapabilities,
  extensible resource_type_name, DAG relationships, frozen model

- NamespacedProject model: identified solely by [[server:]namespace/]name,
  LinkedResource for project-resource links, ContextConfig with memory
  tiers/retention/temporal scope, domain methods (link/unlink/get)

- parse_namespaced_name(): full namespace parsing with reserved/provider
  namespace validation, bare name defaulting to local/

- Extract legacy Project/ProjectSettings/ProjectStats to project_legacy.py

- 81 Behave scenarios (233 steps), all passing
- Lint (ruff), typecheck (pyright) clean
- Full existing test suite (2336 scenarios) unaffected
2026-02-13 21:54:31 +00:00

614 lines
21 KiB
Python

"""Step definitions for Namespaced Project domain model tests.
All assertion steps use 'nsproject' prefix to avoid collisions
with the 128 other step files loaded by Behave globally.
"""
from __future__ import annotations
from behave import given, then, use_step_matcher, when # type: ignore[attr-defined]
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.project import (
ContextConfig,
LinkedResource,
NamespacedProject,
TemporalScope,
parse_namespaced_name,
)
VALID_ULID = "01HZQX7K8M3N4P5R6S7T8V9W0X"
# -- parse_namespaced_name ---------------------------------------------------
@when('I parse the project namespaced name "{value}"')
def step_parse_project_name(context: Context, value: str) -> None:
"""Parse a namespaced name string."""
context.parsed_name = parse_namespaced_name(value)
@then('the nsproject parsed namespace should be "{expected}"')
def step_check_parsed_namespace(context: Context, expected: str) -> None:
"""Check parsed namespace."""
assert context.parsed_name.namespace == expected, (
f"Expected '{expected}', got '{context.parsed_name.namespace}'"
)
@then('the nsproject parsed name should be "{expected}"')
def step_check_parsed_name(context: Context, expected: str) -> None:
"""Check parsed name."""
assert context.parsed_name.name == expected, (
f"Expected '{expected}', got '{context.parsed_name.name}'"
)
@then("the nsproject parsed server should be empty")
def step_check_parsed_server_empty(context: Context) -> None:
"""Check parsed server is None."""
assert context.parsed_name.server is None, (
f"Expected None, got '{context.parsed_name.server}'"
)
@then('the nsproject parsed server should be "{expected}"')
def step_check_parsed_server(context: Context, expected: str) -> None:
"""Check parsed server."""
assert context.parsed_name.server == expected, (
f"Expected '{expected}', got '{context.parsed_name.server}'"
)
@then('the nsproject qualified name should be "{expected}"')
def step_check_qualified_name(context: Context, expected: str) -> None:
"""Check qualified name from parsed or project context."""
obj = getattr(context, "parsed_name", None) or context.project
assert obj.qualified_name == expected, (
f"Expected '{expected}', got '{obj.qualified_name}'"
)
@then('the nsproject namespaced name should be "{expected}"')
def step_check_namespaced_name(context: Context, expected: str) -> None:
"""Check namespaced_name from parsed or project context."""
obj = getattr(context, "parsed_name", None) or context.project
assert obj.namespaced_name == expected, (
f"Expected '{expected}', got '{obj.namespaced_name}'"
)
@then("the nsproject parsed name should be local")
def step_check_parsed_is_local(context: Context) -> None:
"""Check parsed name is local."""
assert context.parsed_name.is_local, "Expected is_local=True"
@then("the nsproject parsed name should be remote")
def step_check_parsed_is_remote(context: Context) -> None:
"""Check parsed name is remote."""
assert context.parsed_name.is_remote, "Expected is_remote=True"
@when('I try to parse the project namespaced name "{value}"')
def step_try_parse_project_name(context: Context, value: str) -> None:
"""Try to parse a namespaced name, expecting failure."""
try:
parse_namespaced_name(value)
context.project_error = None
except ValueError as e:
context.project_error = e
@when("I try to parse an empty project namespaced name")
def step_try_parse_empty_name(context: Context) -> None:
"""Try to parse an empty string."""
try:
parse_namespaced_name("")
context.project_error = None
except ValueError as e:
context.project_error = e
@then("a nsproject validation error should be raised")
def step_check_project_error(context: Context) -> None:
"""Check that a validation error was raised."""
assert context.project_error is not None, (
"Expected a validation error but none was raised"
)
# -- TemporalScope enum -----------------------------------------------------
@then('the nsproject TemporalScope enum should have values "{values}"')
def step_temporal_scope_values(context: Context, values: str) -> None:
"""Check TemporalScope enum values."""
expected = {v.strip() for v in values.split(",")}
actual = {member.value for member in TemporalScope}
assert actual == expected, f"Expected {expected}, got {actual}"
# -- ContextConfig model -----------------------------------------------------
@given("a default ContextConfig")
def step_default_context_config(context: Context) -> None:
"""Create a default ContextConfig."""
context.context_config = ContextConfig()
@given("a ContextConfig with custom ignore patterns")
def step_context_config_custom_ignores(context: Context) -> None:
"""Create a ContextConfig with extra ignore patterns."""
context.context_config = ContextConfig(ignore_patterns=["*.log"])
@given("a ContextConfig with hot_max_tokens {hot:d} and warm_max_decisions {warm:d}")
def step_context_config_memory_tiers(context: Context, hot: int, warm: int) -> None:
"""Create a ContextConfig with memory tier settings."""
context.context_config = ContextConfig(hot_max_tokens=hot, warm_max_decisions=warm)
@then("the nsproject ctx max_file_size should be {expected:d}")
def step_check_max_file_size(context: Context, expected: int) -> None:
"""Check max_file_size."""
assert context.context_config.max_file_size == expected
@then("the nsproject ctx max_total_size should be {expected:d}")
def step_check_max_total_size(context: Context, expected: int) -> None:
"""Check max_total_size."""
assert context.context_config.max_total_size == expected
@then('the nsproject ctx indexing_strategy should be "{expected}"')
def step_check_indexing_strategy(context: Context, expected: str) -> None:
"""Check indexing_strategy."""
assert context.context_config.indexing_strategy == expected
@then('the nsproject ctx chunking_policy should be "{expected}"')
def step_check_chunking_policy(context: Context, expected: str) -> None:
"""Check chunking_policy."""
assert context.context_config.chunking_policy == expected
@then("the nsproject ctx chunk_size should be {expected:d}")
def step_check_chunk_size(context: Context, expected: int) -> None:
"""Check chunk_size."""
assert context.context_config.chunk_size == expected
@then("the nsproject ctx summarize should be {expected}")
def step_check_summarize(context: Context, expected: str) -> None:
"""Check summarize."""
expected_bool = expected.lower() == "true"
assert context.context_config.summarize == expected_bool
@then("the nsproject ctx auto_refresh should be {expected}")
def step_check_auto_refresh(context: Context, expected: str) -> None:
"""Check auto_refresh."""
expected_bool = expected.lower() == "true"
assert context.context_config.auto_refresh == expected_bool
@then('the nsproject ctx temporal_scope should be "{expected}"')
def step_check_temporal_scope(context: Context, expected: str) -> None:
"""Check temporal_scope."""
assert context.context_config.temporal_scope == expected
@then("the nsproject ctx retention_policy should be empty")
def step_check_retention_policy_empty(context: Context) -> None:
"""Check retention_policy is None."""
assert context.context_config.retention_policy is None
@then("the nsproject ctx ignore patterns should include defaults")
def step_check_ignore_patterns_defaults(context: Context) -> None:
"""Check that default ignore patterns are present."""
from cleveragents.domain.models.core.project import (
DEFAULT_IGNORE_PATTERNS,
)
for pattern in DEFAULT_IGNORE_PATTERNS:
assert pattern in context.context_config.ignore_patterns, (
f"Default pattern '{pattern}' missing"
)
@then("the nsproject ctx ignore patterns should include the custom pattern")
def step_check_ignore_patterns_custom(context: Context) -> None:
"""Check that custom ignore pattern is present."""
assert "*.log" in context.context_config.ignore_patterns
@then("the nsproject ctx hot_max_tokens should be {expected:d}")
def step_check_hot_max_tokens(context: Context, expected: int) -> None:
"""Check hot_max_tokens."""
assert context.context_config.hot_max_tokens == expected
@then("the nsproject ctx warm_max_decisions should be {expected:d}")
def step_check_warm_max_decisions(context: Context, expected: int) -> None:
"""Check warm_max_decisions."""
assert context.context_config.warm_max_decisions == expected
@when("I try to mutate a ContextConfig field")
def step_try_mutate_context_config(context: Context) -> None:
"""Try to mutate a frozen ContextConfig."""
try:
context.context_config.chunk_size = 500 # type: ignore[misc]
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a ContextConfig with max_file_size {size:d}")
def step_try_create_context_config_bad_size(context: Context, size: int) -> None:
"""Try to create ContextConfig with invalid max_file_size."""
try:
ContextConfig(max_file_size=size)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a ContextConfig with chunk_size {size:d}")
def step_try_create_context_config_bad_chunk(context: Context, size: int) -> None:
"""Try to create ContextConfig with invalid chunk_size."""
try:
ContextConfig(chunk_size=size)
context.project_error = None
except ValidationError as e:
context.project_error = e
# -- LinkedResource model ----------------------------------------------------
@given('a LinkedResource with resource_id "{rid}"')
def step_create_linked_resource(context: Context, rid: str) -> None:
"""Create a LinkedResource with defaults."""
context.linked_resource = LinkedResource(resource_id=rid)
@given('a LinkedResource with id "{rid}" and alias "{alias}"')
def step_create_linked_resource_with_alias(
context: Context, rid: str, alias: str
) -> None:
"""Create a LinkedResource with alias."""
context.linked_resource = LinkedResource(resource_id=rid, alias=alias)
@then('the nsproject linked rid should be "{expected}"')
def step_check_linked_rid(context: Context, expected: str) -> None:
"""Check linked resource_id."""
assert context.linked_resource.resource_id == expected
@then("the nsproject linked read_only should be {expected}")
def step_check_linked_readonly(context: Context, expected: str) -> None:
"""Check project_read_only."""
expected_bool = expected.lower() == "true"
assert context.linked_resource.project_read_only == expected_bool
@then("the nsproject linked alias should be empty")
def step_check_linked_alias_empty(context: Context) -> None:
"""Check alias is None."""
assert context.linked_resource.alias is None
@then('the nsproject linked alias should be "{expected}"')
def step_check_linked_alias(context: Context, expected: str) -> None:
"""Check alias."""
assert context.linked_resource.alias == expected
@then("the nsproject linked_at should be set")
def step_check_linked_at(context: Context) -> None:
"""Check linked_at is set."""
assert context.linked_resource.linked_at is not None
@when('I try to create a LinkedResource with resource_id "{rid}"')
def step_try_create_linked_resource_bad(context: Context, rid: str) -> None:
"""Try to create a LinkedResource with invalid resource_id."""
try:
LinkedResource(resource_id=rid)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a LinkedResource with empty alias")
def step_try_create_linked_resource_empty_alias(
context: Context,
) -> None:
"""Try to create a LinkedResource with empty alias."""
try:
LinkedResource(resource_id=VALID_ULID, alias="")
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to mutate a LinkedResource field")
def step_try_mutate_linked_resource(context: Context) -> None:
"""Try to mutate a frozen LinkedResource."""
try:
context.linked_resource.alias = "x" # type: ignore[misc]
context.project_error = None
except ValidationError as e:
context.project_error = e
# -- NamespacedProject model -------------------------------------------------
@given('a NamespacedProject with name "{name}"')
def step_create_namespaced_project(context: Context, name: str) -> None:
"""Create a minimal NamespacedProject."""
context.project = NamespacedProject(name=name)
use_step_matcher("re") # type: ignore[attr-defined]
@given( # type: ignore[no-redef]
r'a project "(?P<name>[^"]+)" under namespace'
r' "(?P<namespace>[^"]+)"'
)
def step_create_namespaced_project_with_ns(
context: Context, name: str, namespace: str
) -> None:
"""Create a NamespacedProject with namespace."""
context.project = NamespacedProject(name=name, namespace=namespace)
@given( # type: ignore[no-redef]
r'a project "(?P<name>[^"]+)" under namespace'
r' "(?P<namespace>[^"]+)" on server "(?P<server>[^"]+)"'
)
def step_create_namespaced_project_full(
context: Context, name: str, namespace: str, server: str
) -> None:
"""Create a NamespacedProject with all identity fields."""
context.project = NamespacedProject(name=name, namespace=namespace, server=server)
use_step_matcher("parse") # type: ignore[attr-defined]
@then('the nsproject name should be "{expected}"')
def step_check_project_name(context: Context, expected: str) -> None:
"""Check project name."""
assert context.project.name == expected
@then('the nsproject namespace should be "{expected}"')
def step_check_project_namespace(context: Context, expected: str) -> None:
"""Check project namespace."""
assert context.project.namespace == expected
@then("the nsproject server should be empty")
def step_check_project_server_empty(context: Context) -> None:
"""Check project server is None."""
assert context.project.server is None
@then('the nsproject server should be "{expected}"')
def step_check_project_server(context: Context, expected: str) -> None:
"""Check project server."""
assert context.project.server == expected
@then("the nsproject linked resources should be empty")
def step_check_project_no_linked_resources(
context: Context,
) -> None:
"""Check linked_resources is empty."""
assert context.project.linked_resources == []
@then("the nsproject created_at should be set")
def step_check_project_created_at(context: Context) -> None:
"""Check created_at is set."""
assert context.project.created_at is not None
@then("the nsproject updated_at should be set")
def step_check_project_updated_at(context: Context) -> None:
"""Check updated_at is set."""
assert context.project.updated_at is not None
@then("the nsproject should be local")
def step_check_project_is_local(context: Context) -> None:
"""Check project is local."""
assert context.project.is_local, "Expected is_local=True"
@then("the nsproject should be remote")
def step_check_project_is_remote(context: Context) -> None:
"""Check project is remote."""
assert context.project.is_remote, "Expected is_remote=True"
# -- NamespacedProject validation -------------------------------------------
@when('I try to create a NamespacedProject with name "{name}"')
def step_try_create_project_bad_name(context: Context, name: str) -> None:
"""Try to create a NamespacedProject with invalid name."""
try:
NamespacedProject(name=name)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when('I try to create a NamespacedProject with namespace "{namespace}"')
def step_try_create_project_bad_namespace(context: Context, namespace: str) -> None:
"""Try to create a NamespacedProject with reserved namespace."""
try:
NamespacedProject(name="test", namespace=namespace)
context.project_error = None
except ValidationError as e:
context.project_error = e
@when("I try to create a NamespacedProject with duplicate linked resources")
def step_try_create_project_dup_resources(
context: Context,
) -> None:
"""Try to create a NamespacedProject with duplicate links."""
rid = VALID_ULID
lr1 = LinkedResource(resource_id=rid)
lr2 = LinkedResource(resource_id=rid)
try:
NamespacedProject(name="test", linked_resources=[lr1, lr2])
context.project_error = None
except ValidationError as e:
context.project_error = e
# -- NamespacedProject domain methods ----------------------------------------
@when('I link resource "{rid}" to the project')
@given('I link resource "{rid}" to the project')
def step_link_resource(context: Context, rid: str) -> None:
"""Link a resource to the project."""
context.project = context.project.link_resource(rid)
@when('I link resource "{rid}" with read_only {ro} and alias "{alias}"')
@given('I link resource "{rid}" with read_only {ro} and alias "{alias}"')
def step_link_resource_with_options(
context: Context, rid: str, ro: str, alias: str
) -> None:
"""Link a resource with options."""
read_only = ro.lower() == "true"
context.project = context.project.link_resource(
rid, read_only=read_only, alias=alias
)
@when('I unlink resource "{rid}" from the project')
def step_unlink_resource(context: Context, rid: str) -> None:
"""Unlink a resource from the project."""
context.project = context.project.unlink_resource(rid)
@when('I try to link resource "{rid}" again')
def step_try_link_duplicate(context: Context, rid: str) -> None:
"""Try to link a resource that's already linked."""
try:
context.project.link_resource(rid)
context.project_error = None
except ValueError as e:
context.project_error = e
@when('I try to unlink resource "{rid}" from the project')
def step_try_unlink_nonexistent(context: Context, rid: str) -> None:
"""Try to unlink a resource that's not linked."""
try:
context.project.unlink_resource(rid)
context.project_error = None
except ValueError as e:
context.project_error = e
@when('I try to link resource "{rid}" to the project')
def step_try_link_resource(context: Context, rid: str) -> None:
"""Try to link a resource, might fail."""
try:
context.project = context.project.link_resource(rid)
context.project_error = None
except ValueError as e:
context.project_error = e
@when("I try to link an empty resource_id to the project")
def step_try_link_empty_resource(context: Context) -> None:
"""Try to link with empty resource_id."""
try:
context.project = context.project.link_resource("")
context.project_error = None
except ValueError as e:
context.project_error = e
@when("I try to unlink an empty resource_id from the project")
def step_try_unlink_empty_resource(context: Context) -> None:
"""Try to unlink with empty resource_id."""
try:
context.project = context.project.unlink_resource("")
context.project_error = None
except ValueError as e:
context.project_error = e
@when('I get linked resource "{rid}" from the project')
def step_get_linked_resource(context: Context, rid: str) -> None:
"""Get a linked resource by resource_id."""
context.retrieved_link = context.project.get_linked_resource(rid)
@when('I get linked resource by alias "{alias}" from the project')
def step_get_linked_resource_by_alias(context: Context, alias: str) -> None:
"""Get a linked resource by alias."""
context.retrieved_link = context.project.get_linked_resource_by_alias(alias)
@then("the nsproject should have {count:d} linked resource")
def step_check_linked_count(context: Context, count: int) -> None:
"""Check linked resource count."""
actual = len(context.project.linked_resources)
assert actual == count, f"Expected {count}, got {actual}"
@then('the nsproject first linked resource_id should be "{expected}"')
def step_check_linked_resource_id(context: Context, expected: str) -> None:
"""Check the first linked resource's resource_id."""
assert context.project.linked_resources[0].resource_id == expected
@then("the nsproject first linked resource should be read_only")
def step_check_first_linked_readonly(context: Context) -> None:
"""Check first linked resource is read_only."""
assert context.project.linked_resources[0].project_read_only
@then('the nsproject first linked resource alias should be "{expected}"')
def step_check_first_linked_alias(context: Context, expected: str) -> None:
"""Check first linked resource alias."""
assert context.project.linked_resources[0].alias == expected
@then("the nsproject retrieved link should not be empty")
def step_check_retrieved_link_not_none(context: Context) -> None:
"""Check retrieved link is not None."""
assert context.retrieved_link is not None
@then("the nsproject retrieved link should be empty")
def step_check_retrieved_link_none(context: Context) -> None:
"""Check retrieved link is None."""
assert context.retrieved_link is None
@then('the nsproject retrieved link alias should be "{expected}"')
def step_check_retrieved_link_alias(context: Context, expected: str) -> None:
"""Check retrieved link alias."""
assert context.retrieved_link.alias == expected