forked from HAL9000/cleveragents-core
freemo
051ee7c290
Added 52 new .feature files and corresponding _steps.py files targeting previously uncovered code paths in the following areas: - TUI layer: app, commands, persona (state/schema/registry), widgets, input (shell_exec, reference_parser) - Application services: plan lifecycle/service/executor, session, project, repo indexing, correction, checkpoint, actor, llm_actors, strategy coordinator, resource file watcher, service retry wiring - CLI commands: session, resource, repl, plan, db, automation_profile - Domain models: retry_policy, resource_type, cost_budget, docker_compose_analyzer, detail_level, _sql_string_aware, _postgresql_helpers - Core: circuit_breaker, retry_service_patterns - Infrastructure: repositories, transaction_sandbox, strategy_registry, plugins/loader, container - Config: settings - Agents: plan_generation, context_analysis, auto_debug - A2A: facade All new tests follow the Behave/Gherkin BDD standard. Resolved step definition collisions with unique prefixes. Fixed Alembic fileConfig logger disabling issue (disable_existing_loggers=False). ISSUES CLOSED: #1068
601 lines
22 KiB
Python
601 lines
22 KiB
Python
"""Step definitions for db_repositories_coverage_boost.feature.
|
|
|
|
Targets specific uncovered lines in repositories.py:
|
|
- Lines 1080, 1083: ActionRepository.update - arg_type / requirement as plain
|
|
strings (not enums with .value), hitting the ``else`` branches.
|
|
- Line 1340: LifecyclePlanRepository.update - automation_profile truthy path.
|
|
- Line 1357: LifecyclePlanRepository.update - validation_summary not None.
|
|
- Line 1367: LifecyclePlanRepository.update - execution_env_priority.value.
|
|
- Lines 1472-1473: LifecyclePlanRepository.list_all - DatabaseError branch.
|
|
- Lines 1730, 1734, 1737: ResourceTypeRepository.create - namespaced name
|
|
with "/" and string (non-enum) resource_kind / sandbox_strategy.
|
|
- Lines 1857, 1862, 1867-1872: ResourceTypeRepository.update - string enums
|
|
and cli_args serialization.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.exc import OperationalError
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from cleveragents.core.exceptions import DatabaseError
|
|
from cleveragents.domain.models.core.action import (
|
|
Action,
|
|
ActionArgument,
|
|
ActionState,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
AutomationProfileProvenance,
|
|
AutomationProfileRef,
|
|
ExecutionEnvPriority,
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
from cleveragents.domain.models.core.resource_type import (
|
|
ResourceKind,
|
|
ResourceTypeArgument,
|
|
SandboxStrategy,
|
|
)
|
|
from cleveragents.infrastructure.database.models import Base
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
ActionRepository,
|
|
LifecyclePlanRepository,
|
|
ResourceTypeRepository,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _next_ulid() -> str:
|
|
from ulid import ULID
|
|
|
|
return str(ULID())
|
|
|
|
|
|
def _make_action(
|
|
name: str = "local/test-action",
|
|
state: str = "available",
|
|
arguments: list[ActionArgument] | None = None,
|
|
invariants: list[str] | None = None,
|
|
) -> Action:
|
|
parts = name.split("/", 1)
|
|
namespace = parts[0] if len(parts) == 2 else "local"
|
|
short_name = parts[1] if len(parts) == 2 else parts[0]
|
|
return Action(
|
|
namespaced_name=NamespacedName(namespace=namespace, name=short_name),
|
|
description=f"Test action {short_name}",
|
|
long_description=None,
|
|
definition_of_done=f"Verify {short_name} completes",
|
|
strategy_actor="local/strategist",
|
|
execution_actor="local/executor",
|
|
estimation_actor=None,
|
|
review_actor=None,
|
|
arguments=arguments or [],
|
|
invariants=invariants or [],
|
|
reusable=True,
|
|
read_only=False,
|
|
state=ActionState(state),
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
created_by=None,
|
|
tags=[],
|
|
)
|
|
|
|
|
|
def _make_plan(
|
|
action_name: str,
|
|
plan_id: str | None = None,
|
|
phase: PlanPhase = PlanPhase.STRATEGIZE,
|
|
ns_name: str = "local/test-plan",
|
|
automation_profile: AutomationProfileRef | None = None,
|
|
validation_summary: dict | None = None,
|
|
execution_environment: str | None = None,
|
|
execution_env_priority: ExecutionEnvPriority | None = None,
|
|
) -> Plan:
|
|
pid = plan_id or _next_ulid()
|
|
parts = ns_name.split("/", 1)
|
|
namespace = parts[0] if len(parts) == 2 else "local"
|
|
short_name = parts[1] if len(parts) == 2 else parts[0]
|
|
return Plan(
|
|
identity=PlanIdentity(
|
|
plan_id=pid,
|
|
parent_plan_id=None,
|
|
root_plan_id=None,
|
|
attempt=1,
|
|
),
|
|
namespaced_name=NamespacedName(namespace=namespace, name=short_name),
|
|
action_name=action_name,
|
|
description=f"Test plan for {action_name}",
|
|
definition_of_done="Tests pass",
|
|
phase=phase,
|
|
processing_state=ProcessingState.QUEUED,
|
|
strategy_actor="local/strategist",
|
|
execution_actor="local/executor",
|
|
timestamps=PlanTimestamps(
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
),
|
|
created_by=None,
|
|
tags=[],
|
|
reusable=True,
|
|
read_only=False,
|
|
automation_profile=automation_profile,
|
|
validation_summary=validation_summary,
|
|
execution_environment=execution_environment,
|
|
execution_env_priority=execution_env_priority,
|
|
)
|
|
|
|
|
|
# ========================================================================
|
|
# Background
|
|
# ========================================================================
|
|
|
|
|
|
@given("a fresh in-memory database for repo coverage boost")
|
|
def step_fresh_db(context: Context) -> None:
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
context.rcb_engine = engine
|
|
session = sessionmaker(bind=engine)()
|
|
context.rcb_session = session
|
|
context.rcb_session_factory = lambda: session
|
|
context.rcb_error = None
|
|
|
|
|
|
@given("an action repository for coverage boost")
|
|
def step_action_repo(context: Context) -> None:
|
|
context.rcb_action_repo = ActionRepository(
|
|
session_factory=context.rcb_session_factory,
|
|
)
|
|
|
|
|
|
@given("a lifecycle plan repository for coverage boost")
|
|
def step_plan_repo(context: Context) -> None:
|
|
context.rcb_plan_repo = LifecyclePlanRepository(
|
|
session_factory=context.rcb_session_factory,
|
|
)
|
|
|
|
|
|
@given("a resource type repository for coverage boost")
|
|
def step_resource_type_repo(context: Context) -> None:
|
|
context.rcb_rt_repo = ResourceTypeRepository(
|
|
session_factory=context.rcb_session_factory,
|
|
)
|
|
|
|
|
|
# ========================================================================
|
|
# ActionRepository.update - plain-string arg_type/requirement (L1080, L1083)
|
|
# ========================================================================
|
|
|
|
|
|
@given('a valid action named "{name}" for coverage boost')
|
|
def step_make_action(context: Context, name: str) -> None:
|
|
context.rcb_action = _make_action(name=name)
|
|
|
|
|
|
@given("the action is persisted for coverage boost")
|
|
def step_persist_action(context: Context) -> None:
|
|
context.rcb_action_repo.create(context.rcb_action)
|
|
context.rcb_session.commit()
|
|
|
|
|
|
@when("the action arguments are replaced with plain-string arg_type and requirement")
|
|
def step_replace_args_strings(context: Context) -> None:
|
|
"""Build ActionArguments whose arg_type and requirement are plain strings
|
|
(not enum members) so that ``hasattr(arg.arg_type, 'value')`` returns
|
|
False in the repository update code, exercising lines 1080 and 1083.
|
|
"""
|
|
# We bypass pydantic validation with model_construct so that arg_type
|
|
# and requirement remain plain strings (no .value attribute), hitting
|
|
# the ``else`` branch in ActionRepository.update().
|
|
patched_arg = ActionArgument.model_construct(
|
|
name="target_file",
|
|
arg_type="string", # plain str, no .value
|
|
requirement="required", # plain str, no .value
|
|
description="Target file path",
|
|
default_value=None,
|
|
min_value=None,
|
|
max_value=None,
|
|
validation_pattern=None,
|
|
)
|
|
context.rcb_action = context.rcb_action.model_copy(
|
|
update={"arguments": [patched_arg]}
|
|
)
|
|
|
|
|
|
@when("the action is updated for coverage boost")
|
|
def step_update_action(context: Context) -> None:
|
|
try:
|
|
context.rcb_action = context.rcb_action.model_copy(
|
|
update={"updated_at": datetime.now()}
|
|
)
|
|
context.rcb_result = context.rcb_action_repo.update(context.rcb_action)
|
|
context.rcb_session.commit()
|
|
except Exception as exc:
|
|
context.rcb_error = exc
|
|
|
|
|
|
@then("the update should succeed for coverage boost")
|
|
def step_verify_no_error(context: Context) -> None:
|
|
assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}"
|
|
|
|
|
|
@then("the retrieved action should have the string-typed arguments")
|
|
def step_verify_string_args(context: Context) -> None:
|
|
fetched = context.rcb_action_repo.get_by_name(
|
|
str(context.rcb_action.namespaced_name)
|
|
)
|
|
assert fetched is not None, "Action not found after update"
|
|
assert len(fetched.arguments) == 1
|
|
arg = fetched.arguments[0]
|
|
assert arg.name == "target_file"
|
|
|
|
|
|
# ========================================================================
|
|
# LifecyclePlanRepository.update - automation_profile (L1340)
|
|
# ========================================================================
|
|
|
|
|
|
@given('a lifecycle plan linked to "{action_name}" with automation profile')
|
|
def step_make_plan_with_auto_profile(context: Context, action_name: str) -> None:
|
|
context.rcb_plan = _make_plan(
|
|
action_name=action_name,
|
|
ns_name=f"local/auto-prof-plan-{_next_ulid()[:8]}",
|
|
automation_profile=AutomationProfileRef(
|
|
profile_name="trusted",
|
|
provenance=AutomationProfileProvenance.PLAN,
|
|
),
|
|
)
|
|
|
|
|
|
@given("the lifecycle plan is persisted for coverage boost")
|
|
def step_persist_plan(context: Context) -> None:
|
|
context.rcb_plan_repo.create(context.rcb_plan)
|
|
context.rcb_session.commit()
|
|
|
|
|
|
@when("the plan automation profile is updated to a new profile ref")
|
|
def step_update_auto_profile(context: Context) -> None:
|
|
context.rcb_plan = context.rcb_plan.model_copy(
|
|
update={
|
|
"automation_profile": AutomationProfileRef(
|
|
profile_name="local/careful-auto",
|
|
provenance=AutomationProfileProvenance.ACTION,
|
|
),
|
|
"timestamps": context.rcb_plan.timestamps.model_copy(
|
|
update={"updated_at": datetime.now()}
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
@when("the lifecycle plan is updated for coverage boost")
|
|
def step_update_plan(context: Context) -> None:
|
|
try:
|
|
context.rcb_plan_repo.update(context.rcb_plan)
|
|
context.rcb_session.commit()
|
|
except Exception as exc:
|
|
context.rcb_error = exc
|
|
|
|
|
|
@then("the plan update should succeed for coverage boost")
|
|
def step_verify_plan_update_ok(context: Context) -> None:
|
|
assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}"
|
|
|
|
|
|
@then("the retrieved plan should have the automation profile set")
|
|
def step_verify_auto_profile(context: Context) -> None:
|
|
fetched = context.rcb_plan_repo.get(str(context.rcb_plan.identity.plan_id))
|
|
assert fetched is not None, "Plan not found"
|
|
assert fetched.automation_profile is not None, (
|
|
"automation_profile should not be None"
|
|
)
|
|
assert fetched.automation_profile.profile_name == "local/careful-auto"
|
|
|
|
|
|
# ========================================================================
|
|
# LifecyclePlanRepository.update - validation_summary (L1357)
|
|
# ========================================================================
|
|
|
|
|
|
@given('a lifecycle plan linked to "{action_name}" with validation summary')
|
|
def step_make_plan_with_val_summary(context: Context, action_name: str) -> None:
|
|
context.rcb_plan = _make_plan(
|
|
action_name=action_name,
|
|
ns_name=f"local/val-sum-plan-{_next_ulid()[:8]}",
|
|
validation_summary={"total": 5, "passed": 4, "failed": 1},
|
|
)
|
|
|
|
|
|
@when("the plan validation summary is set to a non-None value")
|
|
def step_set_val_summary(context: Context) -> None:
|
|
context.rcb_plan = context.rcb_plan.model_copy(
|
|
update={
|
|
"validation_summary": {"total": 10, "passed": 10, "failed": 0},
|
|
"timestamps": context.rcb_plan.timestamps.model_copy(
|
|
update={"updated_at": datetime.now()}
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
@then("the retrieved plan should have the validation summary")
|
|
def step_verify_val_summary(context: Context) -> None:
|
|
fetched = context.rcb_plan_repo.get(str(context.rcb_plan.identity.plan_id))
|
|
assert fetched is not None, "Plan not found"
|
|
assert fetched.validation_summary is not None, (
|
|
"validation_summary should not be None"
|
|
)
|
|
assert fetched.validation_summary["total"] == 10
|
|
assert fetched.validation_summary["passed"] == 10
|
|
|
|
|
|
# ========================================================================
|
|
# LifecyclePlanRepository.update - execution_env_priority (L1367)
|
|
# ========================================================================
|
|
|
|
|
|
@given('a lifecycle plan linked to "{action_name}" with env priority')
|
|
def step_make_plan_with_env_priority(context: Context, action_name: str) -> None:
|
|
context.rcb_plan = _make_plan(
|
|
action_name=action_name,
|
|
ns_name=f"local/env-pri-plan-{_next_ulid()[:8]}",
|
|
execution_environment="host",
|
|
execution_env_priority=ExecutionEnvPriority.FALLBACK,
|
|
)
|
|
|
|
|
|
@when("the plan execution env priority is set")
|
|
def step_set_env_priority(context: Context) -> None:
|
|
context.rcb_plan = context.rcb_plan.model_copy(
|
|
update={
|
|
"execution_environment": "container",
|
|
"execution_env_priority": ExecutionEnvPriority.OVERRIDE,
|
|
"timestamps": context.rcb_plan.timestamps.model_copy(
|
|
update={"updated_at": datetime.now()}
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
@then("the retrieved plan should have execution env priority set")
|
|
def step_verify_env_priority(context: Context) -> None:
|
|
fetched = context.rcb_plan_repo.get(str(context.rcb_plan.identity.plan_id))
|
|
assert fetched is not None, "Plan not found"
|
|
assert fetched.execution_env_priority is not None
|
|
assert (
|
|
str(fetched.execution_env_priority) == "override"
|
|
or fetched.execution_env_priority == ExecutionEnvPriority.OVERRIDE
|
|
)
|
|
|
|
|
|
# ========================================================================
|
|
# LifecyclePlanRepository.list_all - DatabaseError (L1472-1473)
|
|
# ========================================================================
|
|
|
|
|
|
@given("a lifecycle plan repository with a broken session factory")
|
|
def step_broken_plan_repo(context: Context) -> None:
|
|
def broken_factory() -> Session:
|
|
mock_session = MagicMock(spec=Session)
|
|
mock_session.query.side_effect = OperationalError(
|
|
"database is locked", None, None
|
|
)
|
|
return mock_session
|
|
|
|
context.rcb_broken_plan_repo = LifecyclePlanRepository(
|
|
session_factory=broken_factory,
|
|
)
|
|
|
|
|
|
@when("list_all is called on the broken plan repository")
|
|
def step_call_list_all_broken(context: Context) -> None:
|
|
try:
|
|
context.rcb_broken_plan_repo.list_all()
|
|
except Exception as exc:
|
|
context.rcb_error = exc
|
|
|
|
|
|
@then('a DatabaseError mentioning "{text}" should be raised for coverage boost')
|
|
def step_verify_db_error(context: Context, text: str) -> None:
|
|
assert context.rcb_error is not None, "Expected DatabaseError, got no error"
|
|
assert isinstance(context.rcb_error, DatabaseError), (
|
|
f"Expected DatabaseError, got {type(context.rcb_error).__name__}: {context.rcb_error}"
|
|
)
|
|
assert text in str(context.rcb_error), (
|
|
f"Expected '{text}' in error message: {context.rcb_error}"
|
|
)
|
|
|
|
|
|
# ========================================================================
|
|
# ResourceTypeRepository.create - namespaced name with "/" (L1730)
|
|
# and string resource_kind / sandbox_strategy (L1734, L1737)
|
|
# ========================================================================
|
|
|
|
|
|
def _make_resource_type_spec(
|
|
name: str,
|
|
description: str = "Test resource type",
|
|
resource_kind: str | ResourceKind = "physical",
|
|
sandbox_strategy: str | SandboxStrategy = "none",
|
|
user_addable: bool = True,
|
|
handler: str | None = None,
|
|
cli_args: list[ResourceTypeArgument] | None = None,
|
|
parent_types: list[str] | None = None,
|
|
child_types: list[str] | None = None,
|
|
auto_discovery: dict | None = None,
|
|
capabilities: dict | None = None,
|
|
equivalence: dict | None = None,
|
|
) -> SimpleNamespace:
|
|
"""Create a SimpleNamespace that looks like a ResourceTypeSpec for the
|
|
repository, but allows plain strings for resource_kind / sandbox_strategy
|
|
to hit the ``else`` branches.
|
|
"""
|
|
return SimpleNamespace(
|
|
name=name,
|
|
description=description,
|
|
resource_kind=resource_kind,
|
|
sandbox_strategy=sandbox_strategy,
|
|
user_addable=user_addable,
|
|
handler=handler,
|
|
cli_args=cli_args or [],
|
|
parent_types=parent_types or [],
|
|
child_types=child_types or [],
|
|
auto_discovery=auto_discovery,
|
|
capabilities=capabilities,
|
|
equivalence=equivalence,
|
|
source=None,
|
|
)
|
|
|
|
|
|
@given('a resource type spec with namespaced name "{name}" and string enums')
|
|
def step_make_rt_namespaced(context: Context, name: str) -> None:
|
|
context.rcb_rt_spec = _make_resource_type_spec(
|
|
name=name,
|
|
resource_kind="physical", # plain string, no .value
|
|
sandbox_strategy="none", # plain string, no .value
|
|
)
|
|
|
|
|
|
@when("the resource type is created for coverage boost")
|
|
def step_create_rt(context: Context) -> None:
|
|
try:
|
|
context.rcb_rt_repo.create(context.rcb_rt_spec)
|
|
context.rcb_session.commit()
|
|
except Exception as exc:
|
|
context.rcb_error = exc
|
|
|
|
|
|
@then("the resource type creation should succeed for coverage boost")
|
|
def step_verify_rt_created(context: Context) -> None:
|
|
assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}"
|
|
|
|
|
|
@then('the retrieved resource type should have namespace "{expected_ns}"')
|
|
def step_verify_rt_namespace(context: Context, expected_ns: str) -> None:
|
|
fetched = context.rcb_rt_repo.get(context.rcb_rt_spec.name)
|
|
assert fetched is not None, f"Resource type '{context.rcb_rt_spec.name}' not found"
|
|
# The name includes namespace
|
|
if "/" in context.rcb_rt_spec.name:
|
|
actual_ns = context.rcb_rt_spec.name.split("/")[0]
|
|
assert actual_ns == expected_ns, (
|
|
f"Expected namespace '{expected_ns}', got '{actual_ns}'"
|
|
)
|
|
|
|
|
|
# ========================================================================
|
|
# ResourceTypeRepository.update - string enums + cli_args (L1857, L1862, L1867-1872)
|
|
# ========================================================================
|
|
|
|
|
|
@given('a resource type spec with name "{name}" and string enums')
|
|
def step_make_rt_for_update(context: Context, name: str) -> None:
|
|
context.rcb_rt_spec = _make_resource_type_spec(
|
|
name=name,
|
|
resource_kind="physical",
|
|
sandbox_strategy="none",
|
|
)
|
|
|
|
|
|
@given("the resource type is persisted for coverage boost")
|
|
def step_persist_rt(context: Context) -> None:
|
|
context.rcb_rt_repo.create(context.rcb_rt_spec)
|
|
context.rcb_session.commit()
|
|
|
|
|
|
@when("the resource type is updated with new description and cli_args")
|
|
def step_update_rt(context: Context) -> None:
|
|
try:
|
|
# Build an update spec with cli_args and plain-string enums
|
|
context.rcb_rt_spec = _make_resource_type_spec(
|
|
name=context.rcb_rt_spec.name,
|
|
description="Updated description with CLI args",
|
|
resource_kind="physical", # plain string, hits else on L1857
|
|
sandbox_strategy="copy_on_write", # plain string, hits else on L1862
|
|
cli_args=[
|
|
ResourceTypeArgument(
|
|
name="target-path",
|
|
type="string",
|
|
required=True,
|
|
description="Path to the target",
|
|
default=None,
|
|
validation_pattern=None,
|
|
),
|
|
ResourceTypeArgument(
|
|
name="verbose",
|
|
type="boolean",
|
|
required=False,
|
|
description="Enable verbose output",
|
|
default="false",
|
|
validation_pattern=None,
|
|
),
|
|
],
|
|
)
|
|
context.rcb_rt_repo.update(context.rcb_rt_spec)
|
|
context.rcb_session.commit()
|
|
except Exception as exc:
|
|
context.rcb_error = exc
|
|
|
|
|
|
@then("the resource type update should succeed for coverage boost")
|
|
def step_verify_rt_updated(context: Context) -> None:
|
|
assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}"
|
|
|
|
|
|
@then("the retrieved resource type should have updated cli_args")
|
|
def step_verify_rt_cli_args(context: Context) -> None:
|
|
fetched = context.rcb_rt_repo.get(context.rcb_rt_spec.name)
|
|
assert fetched is not None, f"Resource type '{context.rcb_rt_spec.name}' not found"
|
|
assert fetched.description == "Updated description with CLI args"
|
|
assert len(fetched.cli_args) == 2
|
|
arg_names = [a.name for a in fetched.cli_args]
|
|
assert "target-path" in arg_names
|
|
assert "verbose" in arg_names
|
|
|
|
|
|
# ========================================================================
|
|
# ResourceTypeRepository.create - builtin name without "/" (L1730 else branch)
|
|
# ========================================================================
|
|
|
|
|
|
@given('a resource type spec with builtin name "{name}" and string enums')
|
|
def step_make_rt_builtin(context: Context, name: str) -> None:
|
|
context.rcb_rt_spec = _make_resource_type_spec(
|
|
name=name,
|
|
resource_kind="physical",
|
|
sandbox_strategy="none",
|
|
)
|
|
|
|
|
|
@then('the retrieved builtin resource type should have namespace "builtin"')
|
|
def step_verify_rt_builtin_ns(context: Context) -> None:
|
|
fetched = context.rcb_rt_repo.get(context.rcb_rt_spec.name)
|
|
assert fetched is not None, f"Resource type '{context.rcb_rt_spec.name}' not found"
|
|
# For builtin names (no "/"), the repo stores namespace="builtin"
|
|
# The domain object doesn't expose the namespace column directly,
|
|
# but we can verify via the database row
|
|
from cleveragents.infrastructure.database.models import ResourceTypeModel
|
|
|
|
row = (
|
|
context.rcb_session.query(ResourceTypeModel)
|
|
.filter_by(name=context.rcb_rt_spec.name)
|
|
.first()
|
|
)
|
|
assert row is not None
|
|
assert row.namespace == "builtin", (
|
|
f"Expected namespace 'builtin', got '{row.namespace}'"
|
|
)
|