forked from HAL9000/cleveragents-core
brent.edwards
9db88a9541
The "Deleting a resource with edges is rejected" scenario was failing because the ResourceEdgeModel row was flushed but never committed. Since ResourceRepository.delete() obtains its own session from the factory, the uncommitted edge was invisible and ResourceHasEdgesError was never raised.
794 lines
27 KiB
Python
794 lines
27 KiB
Python
"""Step definitions for resource_repository.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
from datetime import UTC, datetime
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from cleveragents.domain.models.core.resource import (
|
|
PhysVirt,
|
|
Resource,
|
|
ResourceCapabilities,
|
|
)
|
|
from cleveragents.domain.models.core.resource_type import (
|
|
ResourceKind,
|
|
ResourceTypeArgument,
|
|
ResourceTypeSpec,
|
|
SandboxStrategy,
|
|
)
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
ResourceEdgeModel,
|
|
)
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
DuplicateResourceError,
|
|
DuplicateResourceTypeError,
|
|
ResourceHasEdgesError,
|
|
ResourceNotFoundRepoError,
|
|
ResourceRepository,
|
|
ResourceTypeHasResourcesError,
|
|
ResourceTypeNotFoundError,
|
|
ResourceTypeRepository,
|
|
)
|
|
|
|
# Crockford base32 alphabet for generating ULIDs
|
|
_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
|
|
_ULID_CTR = 0
|
|
|
|
|
|
def _next_ulid() -> str:
|
|
"""Return a unique, valid ULID string for each call."""
|
|
global _ULID_CTR
|
|
_ULID_CTR += 1
|
|
n = _ULID_CTR
|
|
suffix = ""
|
|
for _ in range(8):
|
|
suffix = _CB32[n % 32] + suffix
|
|
n //= 32
|
|
return f"01HGZ6FE0AQDYTR4BX{suffix}"
|
|
|
|
|
|
def _make_resource_type_spec(
|
|
name: str = "myorg/test-type",
|
|
user_addable: bool = True,
|
|
) -> ResourceTypeSpec:
|
|
"""Create a minimal valid ResourceTypeSpec."""
|
|
return ResourceTypeSpec(
|
|
name=name,
|
|
description="Test resource type",
|
|
resource_kind=ResourceKind.PHYSICAL,
|
|
sandbox_strategy=SandboxStrategy.GIT_WORKTREE,
|
|
user_addable=user_addable,
|
|
cli_args=[],
|
|
parent_types=[],
|
|
child_types=[],
|
|
auto_discovery=None,
|
|
equivalence=None,
|
|
handler=None,
|
|
capabilities={
|
|
"read": True,
|
|
"write": True,
|
|
"sandbox": True,
|
|
"checkpoint": False,
|
|
},
|
|
built_in=False,
|
|
)
|
|
|
|
|
|
def _make_resource(
|
|
name: str | None = "myorg/test-resource",
|
|
type_name: str = "myorg/test-type",
|
|
resource_id: str | None = None,
|
|
) -> Resource:
|
|
"""Create a minimal valid Resource domain object."""
|
|
return Resource(
|
|
resource_id=resource_id or _next_ulid(),
|
|
name=name,
|
|
resource_type_name=type_name,
|
|
classification=PhysVirt.PHYSICAL,
|
|
description="Test resource",
|
|
properties={},
|
|
location="/tmp/test",
|
|
content_hash=None,
|
|
sandbox_strategy=None,
|
|
capabilities=ResourceCapabilities(
|
|
readable=True,
|
|
writable=True,
|
|
sandboxable=True,
|
|
checkpointable=False,
|
|
),
|
|
created_at=datetime.now(tz=UTC),
|
|
updated_at=datetime.now(tz=UTC),
|
|
)
|
|
|
|
|
|
# ── Background ────────────────────────────────────────────────
|
|
|
|
|
|
@given("a clean resource repository database")
|
|
def step_clean_resource_repo_db(context: Context) -> None:
|
|
engine = create_engine("sqlite:///:memory:")
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def _set_fk_pragma(dbapi_conn: object, _rec: object) -> None:
|
|
cursor = dbapi_conn.cursor() # type: ignore[union-attr]
|
|
cursor.execute("PRAGMA foreign_keys=ON")
|
|
cursor.close()
|
|
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine)
|
|
context.repo_session_factory = factory
|
|
context.repo_engine = engine
|
|
context.repo_error = None
|
|
|
|
|
|
@given("a resource type repository backed by the database")
|
|
def step_resource_type_repo(context: Context) -> None:
|
|
context.rt_repo = ResourceTypeRepository(context.repo_session_factory)
|
|
|
|
|
|
@given("a resource repository backed by the database")
|
|
def step_resource_repo(context: Context) -> None:
|
|
context.res_repo = ResourceRepository(context.repo_session_factory)
|
|
|
|
|
|
# ── ResourceTypeRepository: Create ────────────────────────────
|
|
|
|
|
|
@given('a valid resource type spec named "{name}"')
|
|
def step_valid_rt_spec(context: Context, name: str) -> None:
|
|
context.rt_spec = _make_resource_type_spec(name=name)
|
|
|
|
|
|
@given("the resource type has already been saved once")
|
|
def step_rt_saved_once(context: Context) -> None:
|
|
context.rt_repo.create(context.rt_spec)
|
|
|
|
|
|
@when("the resource type is saved through the repository")
|
|
def step_save_rt(context: Context) -> None:
|
|
try:
|
|
context.rt_repo.create(context.rt_spec)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
|
|
|
|
@when('a second resource type with the same name "{name}" is saved')
|
|
def step_save_dup_rt(context: Context, name: str) -> None:
|
|
try:
|
|
dup = _make_resource_type_spec(name=name)
|
|
context.rt_repo.create(dup)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
|
|
|
|
@then("the resource type repository should not raise an error")
|
|
def step_rt_no_error(context: Context) -> None:
|
|
assert context.repo_error is None, f"Unexpected error: {context.repo_error}"
|
|
|
|
|
|
@then('the persisted resource type should have name "{name}"')
|
|
def step_rt_persisted_name(context: Context, name: str) -> None:
|
|
result = context.rt_repo.get(name)
|
|
assert result is not None, f"Resource type '{name}' not found"
|
|
assert result.name == name
|
|
|
|
|
|
@then('a DuplicateResourceTypeError should be raised mentioning "{name}"')
|
|
def step_dup_rt_error(context: Context, name: str) -> None:
|
|
assert isinstance(context.repo_error, DuplicateResourceTypeError), (
|
|
f"Expected DuplicateResourceTypeError, got {type(context.repo_error)}"
|
|
)
|
|
assert name in str(context.repo_error)
|
|
|
|
|
|
# ── ResourceTypeRepository: Read ──────────────────────────────
|
|
|
|
|
|
@given("the resource type has been saved through the repository")
|
|
def step_rt_saved(context: Context) -> None:
|
|
context.rt_repo.create(context.rt_spec)
|
|
|
|
|
|
@when('the resource type is looked up by name "{name}"')
|
|
def step_lookup_rt_by_name(context: Context, name: str) -> None:
|
|
context.rt_result = context.rt_repo.get(name)
|
|
|
|
|
|
@then('the returned resource type should have name "{name}"')
|
|
def step_rt_has_name(context: Context, name: str) -> None:
|
|
assert context.rt_result is not None, "Expected a resource type, got None"
|
|
assert context.rt_result.name == name
|
|
|
|
|
|
@then("no resource type should be returned")
|
|
def step_no_rt_returned(context: Context) -> None:
|
|
assert context.rt_result is None, f"Expected None, got {context.rt_result}"
|
|
|
|
|
|
# ── ResourceTypeRepository: List ──────────────────────────────
|
|
|
|
|
|
@given('resource types "{name1}" and "{name2}" have been saved')
|
|
def step_two_rts_saved(context: Context, name1: str, name2: str) -> None:
|
|
context.rt_repo.create(_make_resource_type_spec(name=name1))
|
|
context.rt_repo.create(_make_resource_type_spec(name=name2))
|
|
|
|
|
|
@when("all resource types are listed")
|
|
def step_list_all_rts(context: Context) -> None:
|
|
context.rt_list = context.rt_repo.list_types()
|
|
|
|
|
|
@when('resource types are listed with namespace "{namespace}"')
|
|
def step_list_rts_by_namespace(context: Context, namespace: str) -> None:
|
|
context.rt_list = context.rt_repo.list_types(namespace=namespace)
|
|
|
|
|
|
@given('a resource type "{name}" with user_addable {flag} is saved')
|
|
def step_rt_with_addable_flag(context: Context, name: str, flag: str) -> None:
|
|
spec = _make_resource_type_spec(name=name, user_addable=flag.lower() == "true")
|
|
context.rt_repo.create(spec)
|
|
|
|
|
|
@when("resource types are listed with user_addable {flag}")
|
|
def step_list_rts_by_addable(context: Context, flag: str) -> None:
|
|
context.rt_list = context.rt_repo.list_types(user_addable=flag.lower() == "true")
|
|
|
|
|
|
@then("the resource type list should have {count:d} entries")
|
|
def step_rt_list_count(context: Context, count: int) -> None:
|
|
assert len(context.rt_list) == count, (
|
|
f"Expected {count} entries, got {len(context.rt_list)}"
|
|
)
|
|
|
|
|
|
@then('the first resource type in the list should be "{name}"')
|
|
def step_rt_list_first(context: Context, name: str) -> None:
|
|
assert context.rt_list[0].name == name, (
|
|
f"Expected first entry '{name}', got '{context.rt_list[0].name}'"
|
|
)
|
|
|
|
|
|
# ── ResourceTypeRepository: Update ────────────────────────────
|
|
|
|
|
|
@when('the resource type description is updated to "{desc}"')
|
|
def step_update_rt_desc(context: Context, desc: str) -> None:
|
|
updated = ResourceTypeSpec(
|
|
name=context.rt_spec.name,
|
|
description=desc,
|
|
resource_kind=context.rt_spec.resource_kind,
|
|
sandbox_strategy=context.rt_spec.sandbox_strategy,
|
|
user_addable=context.rt_spec.user_addable,
|
|
cli_args=context.rt_spec.cli_args,
|
|
parent_types=context.rt_spec.parent_types,
|
|
child_types=context.rt_spec.child_types,
|
|
auto_discovery=context.rt_spec.auto_discovery,
|
|
equivalence=context.rt_spec.equivalence,
|
|
handler=context.rt_spec.handler,
|
|
capabilities=context.rt_spec.capabilities,
|
|
built_in=context.rt_spec.built_in,
|
|
)
|
|
context.rt_repo.update(updated)
|
|
|
|
|
|
@then('the resource type should be retrievable with description "{desc}"')
|
|
def step_rt_has_desc(context: Context, desc: str) -> None:
|
|
result = context.rt_repo.get(context.rt_spec.name)
|
|
assert result is not None
|
|
assert result.description == desc, (
|
|
f"Expected description '{desc}', got '{result.description}'"
|
|
)
|
|
|
|
|
|
@when('a non-existent resource type "{name}" is updated')
|
|
def step_update_nonexistent_rt(context: Context, name: str) -> None:
|
|
try:
|
|
fake = _make_resource_type_spec(name=name)
|
|
context.rt_repo.update(fake)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
|
|
|
|
@then("a ResourceTypeNotFoundError should be raised")
|
|
def step_rt_not_found_error(context: Context) -> None:
|
|
assert isinstance(context.repo_error, ResourceTypeNotFoundError), (
|
|
f"Expected ResourceTypeNotFoundError, got {type(context.repo_error)}"
|
|
)
|
|
|
|
|
|
# ── ResourceTypeRepository: Delete ────────────────────────────
|
|
|
|
|
|
@when('the resource type "{name}" is deleted')
|
|
def step_delete_rt(context: Context, name: str) -> None:
|
|
try:
|
|
context.delete_result = context.rt_repo.delete(name)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
context.delete_result = None
|
|
|
|
|
|
@then("the resource type deletion should return true")
|
|
def step_rt_delete_true(context: Context) -> None:
|
|
assert context.delete_result is True, f"Expected True, got {context.delete_result}"
|
|
|
|
|
|
@then("the resource type deletion should return false")
|
|
def step_rt_delete_false(context: Context) -> None:
|
|
assert context.delete_result is False, (
|
|
f"Expected False, got {context.delete_result}"
|
|
)
|
|
|
|
|
|
@then('the resource type "{name}" should no longer exist')
|
|
def step_rt_no_longer_exists(context: Context, name: str) -> None:
|
|
result = context.rt_repo.get(name)
|
|
assert result is None, f"Resource type '{name}' still exists"
|
|
|
|
|
|
@given('a resource of type "{type_name}" exists in the database')
|
|
def step_resource_of_type_exists(context: Context, type_name: str) -> None:
|
|
res = _make_resource(name=None, type_name=type_name)
|
|
context.res_repo.create(res)
|
|
|
|
|
|
@then("a ResourceTypeHasResourcesError should be raised")
|
|
def step_rt_has_resources_error(context: Context) -> None:
|
|
assert isinstance(context.repo_error, ResourceTypeHasResourcesError), (
|
|
f"Expected ResourceTypeHasResourcesError, got {type(context.repo_error)}"
|
|
)
|
|
|
|
|
|
# ── ResourceRepository: Create ────────────────────────────────
|
|
|
|
|
|
@given('a resource type "{name}" exists for resource tests')
|
|
def step_rt_exists_for_res(context: Context, name: str) -> None:
|
|
with contextlib.suppress(DuplicateResourceTypeError):
|
|
context.rt_repo.create(_make_resource_type_spec(name=name))
|
|
|
|
|
|
@given('a valid resource domain object with name "{name}"')
|
|
def step_valid_res_obj(context: Context, name: str) -> None:
|
|
type_name = "myorg/res-type"
|
|
if hasattr(context, "res_type_name"):
|
|
type_name = context.res_type_name
|
|
context.res_obj = _make_resource(name=name, type_name=type_name)
|
|
context.res_type_name = type_name
|
|
|
|
|
|
@given('a valid resource domain object with type "{type_name}"')
|
|
def step_valid_res_obj_with_type(context: Context, type_name: str) -> None:
|
|
context.res_obj = _make_resource(name="myorg/orphan-res", type_name=type_name)
|
|
|
|
|
|
@when("the resource is saved through the repository")
|
|
def step_save_res(context: Context) -> None:
|
|
try:
|
|
context.res_repo.create(context.res_obj)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
|
|
|
|
@then("the resource repository should not raise an error")
|
|
def step_res_no_error(context: Context) -> None:
|
|
assert context.repo_error is None, f"Unexpected error: {context.repo_error}"
|
|
|
|
|
|
@then('the persisted resource should have name "{name}"')
|
|
def step_res_persisted_name(context: Context, name: str) -> None:
|
|
result = context.res_repo.get_by_name(name)
|
|
assert result is not None, f"Resource '{name}' not found"
|
|
assert result.name == name
|
|
|
|
|
|
@then('a ResourceTypeNotFoundError should be raised for type "{type_name}"')
|
|
def step_rt_not_found_for_type(context: Context, type_name: str) -> None:
|
|
assert isinstance(context.repo_error, ResourceTypeNotFoundError), (
|
|
f"Expected ResourceTypeNotFoundError, got {type(context.repo_error)}"
|
|
)
|
|
assert type_name in str(context.repo_error)
|
|
|
|
|
|
@given('a resource with name "{name}" has been saved')
|
|
def step_res_named_saved(context: Context, name: str) -> None:
|
|
# Determine the type name from the most recent resource type created
|
|
type_name = "myorg/res-type"
|
|
if hasattr(context, "res_type_name"):
|
|
type_name = context.res_type_name
|
|
else:
|
|
# Try to guess from existing resource types
|
|
for attr_name in ("rt_spec",):
|
|
if hasattr(context, attr_name):
|
|
type_name = getattr(context, attr_name).name
|
|
break
|
|
|
|
# Use type name from context if set up by previous step
|
|
rt_list = context.rt_repo.list_types()
|
|
if rt_list:
|
|
type_name = rt_list[-1].name
|
|
|
|
res = _make_resource(name=name, type_name=type_name)
|
|
context.res_repo.create(res)
|
|
context.last_saved_resource = res
|
|
context.res_type_name = type_name
|
|
|
|
|
|
@when('a second resource with name "{name}" is saved')
|
|
def step_save_dup_res(context: Context, name: str) -> None:
|
|
try:
|
|
dup = _make_resource(name=name, type_name=context.res_type_name)
|
|
context.res_repo.create(dup)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
|
|
|
|
@then('a DuplicateResourceError should be raised mentioning "{name}"')
|
|
def step_dup_res_error(context: Context, name: str) -> None:
|
|
assert isinstance(context.repo_error, DuplicateResourceError), (
|
|
f"Expected DuplicateResourceError, got {type(context.repo_error)}"
|
|
)
|
|
assert name in str(context.repo_error)
|
|
|
|
|
|
# ── ResourceRepository: Read ──────────────────────────────────
|
|
|
|
|
|
@given("a resource with known ULID has been saved")
|
|
def step_res_known_ulid_saved(context: Context) -> None:
|
|
rt_list = context.rt_repo.list_types()
|
|
type_name = rt_list[-1].name if rt_list else "myorg/get-type"
|
|
ulid = _next_ulid()
|
|
res = _make_resource(
|
|
name=f"myorg/ulid-res-{ulid[-6:]}",
|
|
type_name=type_name,
|
|
resource_id=ulid,
|
|
)
|
|
context.res_repo.create(res)
|
|
context.known_ulid = ulid
|
|
context.last_saved_resource = res
|
|
|
|
|
|
@when("the resource is looked up by its ULID")
|
|
def step_lookup_res_by_ulid(context: Context) -> None:
|
|
context.res_result = context.res_repo.get(context.known_ulid)
|
|
|
|
|
|
@then("the returned resource should have the correct ULID")
|
|
def step_res_has_ulid(context: Context) -> None:
|
|
assert context.res_result is not None, "Expected a resource, got None"
|
|
assert context.res_result.resource_id == context.known_ulid
|
|
|
|
|
|
@when('the resource is looked up by name "{name}"')
|
|
def step_lookup_res_by_name(context: Context, name: str) -> None:
|
|
context.res_result = context.res_repo.get_by_name(name)
|
|
|
|
|
|
@then('the returned resource should have name "{name}"')
|
|
def step_res_has_name(context: Context, name: str) -> None:
|
|
assert context.res_result is not None, "Expected a resource, got None"
|
|
assert context.res_result.name == name
|
|
|
|
|
|
@when('the resource is looked up by ULID "{ulid}"')
|
|
def step_lookup_res_by_specific_ulid(context: Context, ulid: str) -> None:
|
|
context.res_result = context.res_repo.get(ulid)
|
|
|
|
|
|
@then("no resource should be returned")
|
|
def step_no_res_returned(context: Context) -> None:
|
|
assert context.res_result is None, f"Expected None, got {context.res_result}"
|
|
|
|
|
|
# ── ResourceRepository: List ──────────────────────────────────
|
|
|
|
|
|
@given('{count:d} resources of type "{type_name}" have been saved')
|
|
def step_n_resources_saved(context: Context, count: int, type_name: str) -> None:
|
|
for i in range(count):
|
|
res = _make_resource(
|
|
name=f"myorg/batch-{type_name.split('/')[-1]}-{i}",
|
|
type_name=type_name,
|
|
)
|
|
context.res_repo.create(res)
|
|
|
|
|
|
@when("resources are listed with limit {limit:d} and offset {offset:d}")
|
|
def step_list_res_paginated(context: Context, limit: int, offset: int) -> None:
|
|
context.res_list = context.res_repo.list_resources(limit=limit, offset=offset)
|
|
|
|
|
|
@when('resources are listed with type_name "{type_name}"')
|
|
def step_list_res_by_type(context: Context, type_name: str) -> None:
|
|
context.res_list = context.res_repo.list_resources(type_name=type_name)
|
|
|
|
|
|
@given('resource types "{name1}" and "{name2}" exist for resource tests')
|
|
def step_two_rts_for_res(context: Context, name1: str, name2: str) -> None:
|
|
for n in (name1, name2):
|
|
with contextlib.suppress(DuplicateResourceTypeError):
|
|
context.rt_repo.create(_make_resource_type_spec(name=n))
|
|
|
|
|
|
@then("the resource list should have {count:d} entries")
|
|
def step_res_list_count(context: Context, count: int) -> None:
|
|
assert len(context.res_list) == count, (
|
|
f"Expected {count} entries, got {len(context.res_list)}"
|
|
)
|
|
|
|
|
|
# ── ResourceRepository: Update ────────────────────────────────
|
|
|
|
|
|
@when('the resource description is updated to "{desc}"')
|
|
def step_update_res_desc(context: Context, desc: str) -> None:
|
|
original = context.last_saved_resource
|
|
updated = Resource(
|
|
resource_id=original.resource_id,
|
|
name=original.name,
|
|
resource_type_name=original.resource_type_name,
|
|
classification=PhysVirt.PHYSICAL,
|
|
description=desc,
|
|
properties=original.properties,
|
|
location=original.location,
|
|
content_hash=original.content_hash,
|
|
sandbox_strategy=original.sandbox_strategy,
|
|
capabilities=original.capabilities,
|
|
created_at=original.created_at,
|
|
updated_at=datetime.now(tz=UTC),
|
|
)
|
|
context.res_repo.update(updated)
|
|
|
|
|
|
@then('the resource should be retrievable with description "{desc}"')
|
|
def step_res_has_desc(context: Context, desc: str) -> None:
|
|
result = context.res_repo.get(context.last_saved_resource.resource_id)
|
|
assert result is not None
|
|
assert result.description == desc, (
|
|
f"Expected description '{desc}', got '{result.description}'"
|
|
)
|
|
|
|
|
|
@when('a non-existent resource with ULID "{ulid}" is updated')
|
|
def step_update_nonexistent_res(context: Context, ulid: str) -> None:
|
|
try:
|
|
fake = _make_resource(
|
|
name="myorg/phantom-res",
|
|
type_name="myorg/phantom-type",
|
|
resource_id=ulid,
|
|
)
|
|
context.res_repo.update(fake)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
|
|
|
|
@then("a ResourceNotFoundRepoError should be raised")
|
|
def step_res_not_found_error(context: Context) -> None:
|
|
assert isinstance(context.repo_error, ResourceNotFoundRepoError), (
|
|
f"Expected ResourceNotFoundRepoError, got {type(context.repo_error)}"
|
|
)
|
|
|
|
|
|
# ── ResourceRepository: Delete ────────────────────────────────
|
|
|
|
|
|
@when("the resource is deleted by its ULID")
|
|
def step_delete_res_by_ulid(context: Context) -> None:
|
|
try:
|
|
ulid = context.last_saved_resource.resource_id
|
|
context.delete_result = context.res_repo.delete(ulid)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
context.delete_result = None
|
|
|
|
|
|
@then("the resource deletion should return true")
|
|
def step_res_delete_true(context: Context) -> None:
|
|
assert context.delete_result is True, f"Expected True, got {context.delete_result}"
|
|
|
|
|
|
@then("the resource deletion should return false")
|
|
def step_res_delete_false(context: Context) -> None:
|
|
assert context.delete_result is False, (
|
|
f"Expected False, got {context.delete_result}"
|
|
)
|
|
|
|
|
|
@then("the resource should no longer exist")
|
|
def step_res_no_longer_exists(context: Context) -> None:
|
|
result = context.res_repo.get(context.last_saved_resource.resource_id)
|
|
assert result is None, "Resource still exists"
|
|
|
|
|
|
@when('the resource with ULID "{ulid}" is deleted')
|
|
def step_delete_res_specific_ulid(context: Context, ulid: str) -> None:
|
|
try:
|
|
context.delete_result = context.res_repo.delete(ulid)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
context.delete_result = None
|
|
|
|
|
|
@given('a parent and child resource of type "{type_name}" are linked')
|
|
def step_parent_child_linked(context: Context, type_name: str) -> None:
|
|
parent = _make_resource(
|
|
name=f"myorg/parent-{_next_ulid()[-6:]}",
|
|
type_name=type_name,
|
|
)
|
|
child = _make_resource(
|
|
name=f"myorg/child-{_next_ulid()[-6:]}",
|
|
type_name=type_name,
|
|
)
|
|
context.res_repo.create(parent)
|
|
context.res_repo.create(child)
|
|
context.parent_resource = parent
|
|
context.child_resource = child
|
|
|
|
# Insert edge directly via session
|
|
session: Session = context.repo_session_factory()
|
|
edge = ResourceEdgeModel(
|
|
parent_id=parent.resource_id,
|
|
child_id=child.resource_id,
|
|
link_type="contains",
|
|
auto_discovered=False,
|
|
created_at=datetime.now(tz=UTC).isoformat(),
|
|
)
|
|
session.add(edge)
|
|
session.flush()
|
|
session.commit()
|
|
|
|
|
|
@when("the parent resource is deleted")
|
|
def step_delete_parent_res(context: Context) -> None:
|
|
try:
|
|
context.delete_result = context.res_repo.delete(
|
|
context.parent_resource.resource_id
|
|
)
|
|
context.repo_error = None
|
|
except Exception as exc:
|
|
context.repo_error = exc
|
|
context.delete_result = None
|
|
|
|
|
|
@then("a ResourceHasEdgesError should be raised")
|
|
def step_res_has_edges_error(context: Context) -> None:
|
|
assert isinstance(context.repo_error, ResourceHasEdgesError), (
|
|
f"Expected ResourceHasEdgesError, got {type(context.repo_error)}"
|
|
)
|
|
|
|
|
|
# ── ResourceRepository: resolve_namespaced_name ───────────────
|
|
|
|
|
|
@when('the resource is resolved by name_or_id "{name_or_id}"')
|
|
def step_resolve_res(context: Context, name_or_id: str) -> None:
|
|
context.resolved_result = context.res_repo.resolve_namespaced_name(name_or_id)
|
|
|
|
|
|
@then('the resolved resource should have name "{name}"')
|
|
def step_resolved_has_name(context: Context, name: str) -> None:
|
|
assert context.resolved_result is not None, "Expected a resource, got None"
|
|
assert context.resolved_result.name == name
|
|
|
|
|
|
@then("no resolved resource should be returned")
|
|
def step_no_resolved_res(context: Context) -> None:
|
|
assert context.resolved_result is None, (
|
|
f"Expected None, got {context.resolved_result}"
|
|
)
|
|
|
|
|
|
@given("a resource with known ULID has been saved for resolve test")
|
|
def step_res_ulid_for_resolve(context: Context) -> None:
|
|
rt_list = context.rt_repo.list_types()
|
|
type_name = rt_list[-1].name if rt_list else "myorg/resolve-ulid-type"
|
|
ulid = _next_ulid()
|
|
res = _make_resource(
|
|
name=None,
|
|
type_name=type_name,
|
|
resource_id=ulid,
|
|
)
|
|
context.res_repo.create(res)
|
|
context.known_ulid = ulid
|
|
context.last_saved_resource = res
|
|
|
|
|
|
@when("the resource is resolved by its ULID")
|
|
def step_resolve_by_ulid(context: Context) -> None:
|
|
context.resolved_result = context.res_repo.resolve_namespaced_name(
|
|
context.known_ulid
|
|
)
|
|
|
|
|
|
@then("the resolved resource should have the correct ULID")
|
|
def step_resolved_has_ulid(context: Context) -> None:
|
|
assert context.resolved_result is not None, "Expected a resource, got None"
|
|
assert context.resolved_result.resource_id == context.known_ulid
|
|
|
|
|
|
# ── ResourceTypeRepository: domain round-trip ─────────────────
|
|
|
|
|
|
@given("a full resource type spec with cli_args and capabilities")
|
|
def step_full_rt_spec(context: Context) -> None:
|
|
context.full_rt_spec = ResourceTypeSpec(
|
|
name="myorg/full-type",
|
|
description="Full test type with all fields",
|
|
resource_kind=ResourceKind.PHYSICAL,
|
|
sandbox_strategy=SandboxStrategy.COPY_ON_WRITE,
|
|
user_addable=True,
|
|
cli_args=[
|
|
ResourceTypeArgument(
|
|
name="path",
|
|
type="path",
|
|
required=True,
|
|
description="Path to resource",
|
|
),
|
|
ResourceTypeArgument(
|
|
name="branch",
|
|
type="string",
|
|
required=False,
|
|
description="Branch name",
|
|
default="main",
|
|
),
|
|
],
|
|
parent_types=["myorg/parent-type"],
|
|
child_types=["myorg/child-type"],
|
|
auto_discovery={"strategy": "glob", "pattern": "**/*.py"},
|
|
equivalence=None,
|
|
handler="myorg.handlers:FullHandler",
|
|
capabilities={
|
|
"read": True,
|
|
"write": True,
|
|
"sandbox": True,
|
|
"checkpoint": True,
|
|
},
|
|
built_in=False,
|
|
)
|
|
|
|
|
|
@when("the full resource type is saved and retrieved")
|
|
def step_save_and_retrieve_full_rt(context: Context) -> None:
|
|
context.rt_repo.create(context.full_rt_spec)
|
|
context.restored_rt = context.rt_repo.get(context.full_rt_spec.name)
|
|
|
|
|
|
@then("the restored resource type should match the original fields")
|
|
def step_rt_fields_match(context: Context) -> None:
|
|
original = context.full_rt_spec
|
|
restored = context.restored_rt
|
|
assert restored is not None, "Restored resource type is None"
|
|
assert restored.name == original.name
|
|
assert restored.description == original.description
|
|
assert restored.resource_kind == original.resource_kind
|
|
assert restored.sandbox_strategy == original.sandbox_strategy
|
|
assert restored.user_addable == original.user_addable
|
|
assert len(restored.cli_args) == len(original.cli_args)
|
|
assert restored.cli_args[0].name == "path"
|
|
assert restored.cli_args[0].type == "path"
|
|
assert restored.cli_args[0].required is True
|
|
assert restored.cli_args[1].name == "branch"
|
|
assert restored.cli_args[1].default == "main"
|
|
assert restored.parent_types == original.parent_types
|
|
assert restored.child_types == original.child_types
|
|
assert restored.auto_discovery == original.auto_discovery
|
|
assert restored.handler == original.handler
|
|
assert restored.capabilities["checkpoint"] is True
|