brent.edwards
5d2e70cff0
CI / lint (pull_request) Successful in 14s
CI / typecheck (pull_request) Successful in 28s
CI / security (pull_request) Successful in 25s
CI / quality (pull_request) Successful in 47s
CI / unit_tests (pull_request) Successful in 16m4s
CI / build (pull_request) Successful in 15s
CI / integration_tests (pull_request) Successful in 9m36s
CI / coverage (pull_request) Successful in 8m43s
CI / docker (pull_request) Successful in 40s
975 lines
34 KiB
Python
975 lines
34 KiB
Python
"""Step definitions for resource registry database table tests.
|
|
|
|
Tests the ResourceTypeModel, ResourceModel, and ResourceEdgeModel ORM models
|
|
and their corresponding database tables (resource_types, resources,
|
|
resource_edges). Uses an in-memory SQLite database with schema created via
|
|
``Base.metadata.create_all``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from sqlalchemy import create_engine, event, inspect, text
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
ResourceEdgeModel,
|
|
ResourceModel,
|
|
ResourceTypeModel,
|
|
)
|
|
|
|
|
|
def _now_iso() -> str:
|
|
"""Return current UTC time as ISO-8601 string."""
|
|
return datetime.now(tz=UTC).isoformat()
|
|
|
|
|
|
def _make_ulid(suffix: str = "0") -> str:
|
|
"""Create a deterministic 26-char fake ULID for testing."""
|
|
return ("0" * (26 - len(suffix))) + suffix
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _set_sqlite_pragma(dbapi_conn: Any, _connection_record: Any) -> None:
|
|
"""Enable FK enforcement on every new SQLite connection."""
|
|
cursor = dbapi_conn.cursor()
|
|
cursor.execute("PRAGMA foreign_keys = ON")
|
|
cursor.close()
|
|
|
|
|
|
@given("a resource registry in-memory database is initialized")
|
|
def step_resource_registry_init_db(context: Any) -> None:
|
|
"""Create an in-memory SQLite database with all tables."""
|
|
engine = create_engine("sqlite:///:memory:")
|
|
# Enable FK enforcement on every connection via event listener
|
|
event.listen(engine, "connect", _set_sqlite_pragma)
|
|
# Re-establish FK pragma on the initial connection used for metadata
|
|
with engine.connect() as conn:
|
|
conn.execute(text("PRAGMA foreign_keys = ON"))
|
|
conn.commit()
|
|
Base.metadata.create_all(engine)
|
|
context.rr_engine = engine
|
|
context.rr_session_factory = sessionmaker(bind=engine)
|
|
context.rr_session = context.rr_session_factory()
|
|
# Storage for cross-step references
|
|
context.rr_resource_type = None
|
|
context.rr_resource = None
|
|
context.rr_parent_resource = None
|
|
context.rr_parent_resource_2 = None
|
|
context.rr_child_resource = None
|
|
context.rr_child_resource_2 = None
|
|
context.rr_edge = None
|
|
context.rr_error = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Table existence
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('the resource registry database should contain table "{table_name}"')
|
|
def step_resource_registry_table_exists(context: Any, table_name: str) -> None:
|
|
inspector = inspect(context.rr_engine)
|
|
tables = inspector.get_table_names()
|
|
assert table_name in tables, f"Table {table_name} not found. Tables: {tables}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ResourceTypeModel CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a resource type "{name}" with kind "{kind}" and user_addable {addable}')
|
|
def step_resource_registry_create_type(
|
|
context: Any, name: str, kind: str, addable: str
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
namespace = name.split("/")[0]
|
|
rt = ResourceTypeModel()
|
|
rt.name = name
|
|
rt.namespace = namespace
|
|
rt.resource_kind = kind
|
|
rt.user_addable = addable.lower() == "true"
|
|
rt.created_at = _now_iso()
|
|
rt.updated_at = _now_iso()
|
|
session.add(rt)
|
|
session.commit()
|
|
context.rr_resource_type = rt
|
|
|
|
|
|
@then('I can retrieve resource type "{name}"')
|
|
def step_resource_registry_get_type(context: Any, name: str) -> None:
|
|
session: Session = context.rr_session
|
|
rt = session.get(ResourceTypeModel, name)
|
|
assert rt is not None, f"ResourceType {name} not found"
|
|
context.rr_resource_type = rt
|
|
|
|
|
|
@then('the retrieved resource type namespace is "{expected}"')
|
|
def step_resource_registry_type_namespace(context: Any, expected: str) -> None:
|
|
assert context.rr_resource_type.namespace == expected
|
|
|
|
|
|
@then('the retrieved resource type kind is "{expected}"')
|
|
def step_resource_registry_type_kind(context: Any, expected: str) -> None:
|
|
assert context.rr_resource_type.resource_kind == expected
|
|
|
|
|
|
@then("the retrieved resource type user_addable is {expected}")
|
|
def step_resource_registry_type_user_addable(context: Any, expected: str) -> None:
|
|
expected_bool = expected.lower() == "true"
|
|
assert context.rr_resource_type.user_addable == expected_bool
|
|
|
|
|
|
@when('I create a resource type "{name}" with JSON fields')
|
|
def step_resource_registry_create_type_json(context: Any, name: str) -> None:
|
|
session: Session = context.rr_session
|
|
namespace = name.split("/")[0]
|
|
rt = ResourceTypeModel()
|
|
rt.name = name
|
|
rt.namespace = namespace
|
|
rt.resource_kind = "physical"
|
|
rt.user_addable = True
|
|
rt.args_schema_json = json.dumps(
|
|
[{"flag": "--path", "help": "Path to repo", "required": True}]
|
|
)
|
|
rt.allowed_parent_types_json = json.dumps([])
|
|
rt.allowed_child_types_json = json.dumps(
|
|
[{"type": "builtin/fs-directory"}, {"type": "builtin/git"}]
|
|
)
|
|
rt.auto_discover_json = json.dumps({"enabled": True, "max_depth": 10})
|
|
rt.capabilities_json = json.dumps(
|
|
{
|
|
"readable": True,
|
|
"writable": True,
|
|
"sandboxable": True,
|
|
"checkpointable": True,
|
|
}
|
|
)
|
|
rt.created_at = _now_iso()
|
|
rt.updated_at = _now_iso()
|
|
session.add(rt)
|
|
session.commit()
|
|
context.rr_resource_type = rt
|
|
|
|
|
|
@then("the resource type args_schema_json round-trips correctly")
|
|
def step_resource_registry_type_args_json(context: Any) -> None:
|
|
data = json.loads(context.rr_resource_type.args_schema_json)
|
|
assert isinstance(data, list)
|
|
assert data[0]["flag"] == "--path"
|
|
|
|
|
|
@then("the resource type allowed_parent_types_json round-trips correctly")
|
|
def step_resource_registry_type_parents_json(context: Any) -> None:
|
|
data = json.loads(context.rr_resource_type.allowed_parent_types_json)
|
|
assert isinstance(data, list)
|
|
|
|
|
|
@then("the resource type allowed_child_types_json round-trips correctly")
|
|
def step_resource_registry_type_children_json(context: Any) -> None:
|
|
data = json.loads(context.rr_resource_type.allowed_child_types_json)
|
|
assert isinstance(data, list)
|
|
assert len(data) == 2
|
|
|
|
|
|
@then("the resource type auto_discover_json round-trips correctly")
|
|
def step_resource_registry_type_autodiscover_json(context: Any) -> None:
|
|
data = json.loads(context.rr_resource_type.auto_discover_json)
|
|
assert data["enabled"] is True
|
|
assert data["max_depth"] == 10
|
|
|
|
|
|
@then("the resource type capabilities_json round-trips correctly")
|
|
def step_resource_registry_type_capabilities_json(context: Any) -> None:
|
|
data = json.loads(context.rr_resource_type.capabilities_json)
|
|
assert data["readable"] is True
|
|
assert data["sandboxable"] is True
|
|
|
|
|
|
@then('creating a duplicate resource type "{name}" raises an integrity error')
|
|
def step_resource_registry_duplicate_type(context: Any, name: str) -> None:
|
|
session: Session = context.rr_session
|
|
namespace = name.split("/")[0]
|
|
rt = ResourceTypeModel()
|
|
rt.name = name
|
|
rt.namespace = namespace
|
|
rt.resource_kind = "physical"
|
|
rt.user_addable = False
|
|
rt.created_at = _now_iso()
|
|
rt.updated_at = _now_iso()
|
|
try:
|
|
session.add(rt)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for duplicate resource type"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then('creating a resource type with kind "{kind}" raises an integrity error')
|
|
def step_resource_registry_invalid_kind(context: Any, kind: str) -> None:
|
|
session: Session = context.rr_session
|
|
try:
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO resource_types (name, namespace, resource_kind, "
|
|
"user_addable, created_at, updated_at) "
|
|
"VALUES (:name, :ns, :kind, 0, :ts, :ts)"
|
|
),
|
|
{
|
|
"name": "test/invalid-kind",
|
|
"ns": "test",
|
|
"kind": kind,
|
|
"ts": _now_iso(),
|
|
},
|
|
)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for invalid resource_kind"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then('listing resource types by namespace "{ns}" returns {count:d} results')
|
|
def step_resource_registry_list_types_ns_plural(
|
|
context: Any, ns: str, count: int
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
results = (
|
|
session.query(ResourceTypeModel).filter(ResourceTypeModel.namespace == ns).all()
|
|
)
|
|
assert len(results) == count, f"Expected {count}, got {len(results)}"
|
|
|
|
|
|
@then('listing resource types by namespace "{ns}" returns {count:d} result')
|
|
def step_resource_registry_list_types_ns_singular(
|
|
context: Any, ns: str, count: int
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
results = (
|
|
session.query(ResourceTypeModel).filter(ResourceTypeModel.namespace == ns).all()
|
|
)
|
|
assert len(results) == count, f"Expected {count}, got {len(results)}"
|
|
|
|
|
|
@then('listing resource types by kind "{kind}" returns {count:d} results')
|
|
def step_resource_registry_list_types_kind_plural(
|
|
context: Any, kind: str, count: int
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
results = (
|
|
session.query(ResourceTypeModel)
|
|
.filter(ResourceTypeModel.resource_kind == kind)
|
|
.all()
|
|
)
|
|
assert len(results) == count, f"Expected {count}, got {len(results)}"
|
|
|
|
|
|
@then('listing resource types by kind "{kind}" returns {count:d} result')
|
|
def step_resource_registry_list_types_kind_singular(
|
|
context: Any, kind: str, count: int
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
results = (
|
|
session.query(ResourceTypeModel)
|
|
.filter(ResourceTypeModel.resource_kind == kind)
|
|
.all()
|
|
)
|
|
assert len(results) == count, f"Expected {count}, got {len(results)}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ResourceModel CRUD - prerequisites
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a resource type "{name}" exists')
|
|
def step_resource_registry_ensure_type_physical(context: Any, name: str) -> None:
|
|
session: Session = context.rr_session
|
|
existing = session.get(ResourceTypeModel, name)
|
|
if existing is None:
|
|
namespace = name.split("/")[0]
|
|
rt = ResourceTypeModel()
|
|
rt.name = name
|
|
rt.namespace = namespace
|
|
rt.resource_kind = "physical"
|
|
rt.user_addable = True
|
|
rt.created_at = _now_iso()
|
|
rt.updated_at = _now_iso()
|
|
session.add(rt)
|
|
session.commit()
|
|
|
|
|
|
@given('a resource type "{name}" exists as virtual')
|
|
def step_resource_registry_ensure_type_virtual(context: Any, name: str) -> None:
|
|
session: Session = context.rr_session
|
|
existing = session.get(ResourceTypeModel, name)
|
|
if existing is None:
|
|
namespace = name.split("/")[0]
|
|
rt = ResourceTypeModel()
|
|
rt.name = name
|
|
rt.namespace = namespace
|
|
rt.resource_kind = "virtual"
|
|
rt.user_addable = False
|
|
rt.created_at = _now_iso()
|
|
rt.updated_at = _now_iso()
|
|
session.add(rt)
|
|
session.commit()
|
|
|
|
|
|
@given('a resource type "{name}" exists as physical')
|
|
def step_resource_registry_ensure_type_physical_explicit(
|
|
context: Any, name: str
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
existing = session.get(ResourceTypeModel, name)
|
|
if existing is None:
|
|
namespace = name.split("/")[0]
|
|
rt = ResourceTypeModel()
|
|
rt.name = name
|
|
rt.namespace = namespace
|
|
rt.resource_kind = "physical"
|
|
rt.user_addable = True
|
|
rt.created_at = _now_iso()
|
|
rt.updated_at = _now_iso()
|
|
session.add(rt)
|
|
session.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ResourceModel CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create a resource "{rname}" of type "{tname}" with location "{loc}"')
|
|
def step_resource_registry_create_resource(
|
|
context: Any, rname: str, tname: str, loc: str
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
namespace = rname.split("/")[0]
|
|
rid = _make_ulid(rname.replace("/", "_"))
|
|
# Look up the type to get the kind
|
|
rt = session.get(ResourceTypeModel, tname)
|
|
kind = rt.resource_kind if rt else "physical"
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.namespaced_name = rname
|
|
r.namespace = namespace
|
|
r.type_name = tname
|
|
r.resource_kind = kind
|
|
r.location = loc
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_resource = r
|
|
|
|
|
|
@then("I can retrieve the resource by id")
|
|
def step_resource_registry_get_resource(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
rid = context.rr_resource.resource_id
|
|
r = session.get(ResourceModel, rid)
|
|
assert r is not None, f"Resource {rid} not found"
|
|
context.rr_resource = r
|
|
|
|
|
|
@then('the retrieved resource namespaced_name is "{expected}"')
|
|
def step_resource_registry_resource_name(context: Any, expected: str) -> None:
|
|
assert context.rr_resource.namespaced_name == expected
|
|
|
|
|
|
@then('the retrieved resource namespace is "{expected}"')
|
|
def step_resource_registry_resource_namespace(context: Any, expected: str) -> None:
|
|
assert context.rr_resource.namespace == expected
|
|
|
|
|
|
@then('the retrieved resource type_name is "{expected}"')
|
|
def step_resource_registry_resource_type(context: Any, expected: str) -> None:
|
|
assert context.rr_resource.type_name == expected
|
|
|
|
|
|
@then('the retrieved resource kind is "{expected}"')
|
|
def step_resource_registry_resource_kind(context: Any, expected: str) -> None:
|
|
assert context.rr_resource.resource_kind == expected
|
|
|
|
|
|
@then('the retrieved resource location is "{expected}"')
|
|
def step_resource_registry_resource_location(context: Any, expected: str) -> None:
|
|
assert context.rr_resource.location == expected
|
|
|
|
|
|
@when('I create an unnamed resource of type "{tname}"')
|
|
def step_resource_registry_create_unnamed(context: Any, tname: str) -> None:
|
|
session: Session = context.rr_session
|
|
rid = _make_ulid("unnamed1")
|
|
rt = session.get(ResourceTypeModel, tname)
|
|
kind = rt.resource_kind if rt else "virtual"
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.namespaced_name = None
|
|
r.namespace = None
|
|
r.type_name = tname
|
|
r.resource_kind = kind
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_resource = r
|
|
|
|
|
|
@then("I can retrieve the unnamed resource by id")
|
|
def step_resource_registry_get_unnamed(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
rid = context.rr_resource.resource_id
|
|
r = session.get(ResourceModel, rid)
|
|
assert r is not None, f"Resource {rid} not found"
|
|
context.rr_resource = r
|
|
|
|
|
|
@then("the unnamed resource namespaced_name is null")
|
|
def step_resource_registry_unnamed_null(context: Any) -> None:
|
|
assert context.rr_resource.namespaced_name is None
|
|
|
|
|
|
@when("I create a resource with JSON properties")
|
|
def step_resource_registry_create_resource_json(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
rid = _make_ulid("jsonprops")
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.namespaced_name = "local/json-test"
|
|
r.namespace = "local"
|
|
r.type_name = "builtin/git-checkout"
|
|
r.resource_kind = "physical"
|
|
r.location = "/path/to/repo"
|
|
r.properties_json = json.dumps({"branch": "main", "remote": "origin"})
|
|
r.metadata_json = json.dumps({"tags": ["important"], "priority": 1})
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_resource = r
|
|
|
|
|
|
@then("the resource properties_json round-trips correctly")
|
|
def step_resource_registry_resource_props(context: Any) -> None:
|
|
data = json.loads(context.rr_resource.properties_json)
|
|
assert data["branch"] == "main"
|
|
assert data["remote"] == "origin"
|
|
|
|
|
|
@then("the resource metadata_json round-trips correctly")
|
|
def step_resource_registry_resource_meta(context: Any) -> None:
|
|
data = json.loads(context.rr_resource.metadata_json)
|
|
assert data["tags"] == ["important"]
|
|
assert data["priority"] == 1
|
|
|
|
|
|
@then('creating a resource with kind "{kind}" raises an integrity error')
|
|
def step_resource_registry_invalid_resource_kind(context: Any, kind: str) -> None:
|
|
session: Session = context.rr_session
|
|
try:
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO resources (resource_id, type_name, resource_kind, "
|
|
"read_only, auto_discovered, created_at, updated_at) "
|
|
"VALUES (:rid, :tname, :kind, 0, 0, :ts, :ts)"
|
|
),
|
|
{
|
|
"rid": _make_ulid("badkind"),
|
|
"tname": "builtin/git-checkout",
|
|
"kind": kind,
|
|
"ts": _now_iso(),
|
|
},
|
|
)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for invalid resource_kind"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then("creating a resource with non-existent type raises an integrity error")
|
|
def step_resource_registry_missing_type(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
r = ResourceModel()
|
|
r.resource_id = _make_ulid("orphan")
|
|
r.type_name = "nonexistent/type"
|
|
r.resource_kind = "physical"
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
try:
|
|
session.add(r)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for missing FK"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then('creating a duplicate named resource "{name}" raises an integrity error')
|
|
def step_resource_registry_duplicate_resource(context: Any, name: str) -> None:
|
|
session: Session = context.rr_session
|
|
r = ResourceModel()
|
|
r.resource_id = _make_ulid("dup")
|
|
r.namespaced_name = name
|
|
r.namespace = name.split("/")[0]
|
|
r.type_name = "builtin/git-checkout"
|
|
r.resource_kind = "physical"
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
try:
|
|
session.add(r)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for duplicate named resource"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then('deleting resource type "{name}" raises an integrity error')
|
|
def step_resource_registry_restrict_type_delete(context: Any, name: str) -> None:
|
|
session: Session = context.rr_session
|
|
rt = session.get(ResourceTypeModel, name)
|
|
assert rt is not None
|
|
try:
|
|
session.delete(rt)
|
|
session.flush()
|
|
msg = "Expected IntegrityError for RESTRICT delete"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then('querying resources by type "{tname}" returns {count:d} result')
|
|
def step_resource_registry_query_by_type(context: Any, tname: str, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
results = (
|
|
session.query(ResourceModel).filter(ResourceModel.type_name == tname).all()
|
|
)
|
|
assert len(results) == count, f"Expected {count}, got {len(results)}"
|
|
|
|
|
|
@then('querying resources by namespace "{ns}" returns {count:d} result')
|
|
def step_resource_registry_query_by_ns(context: Any, ns: str, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
results = session.query(ResourceModel).filter(ResourceModel.namespace == ns).all()
|
|
assert len(results) == count, f"Expected {count}, got {len(results)}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ResourceEdgeModel - prerequisites
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('a parent resource "{rname}" of type "{tname}"')
|
|
def step_resource_registry_create_parent(context: Any, rname: str, tname: str) -> None:
|
|
session: Session = context.rr_session
|
|
rid = _make_ulid("parent_" + rname.replace("/", "_"))
|
|
rt = session.get(ResourceTypeModel, tname)
|
|
kind = rt.resource_kind if rt else "physical"
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.namespaced_name = rname
|
|
r.namespace = rname.split("/")[0]
|
|
r.type_name = tname
|
|
r.resource_kind = kind
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_parent_resource = r
|
|
|
|
|
|
@given('a second parent resource "{rname}" of type "{tname}"')
|
|
def step_resource_registry_create_second_parent(
|
|
context: Any, rname: str, tname: str
|
|
) -> None:
|
|
session: Session = context.rr_session
|
|
rid = _make_ulid("parent2_" + rname.replace("/", "_"))
|
|
rt = session.get(ResourceTypeModel, tname)
|
|
kind = rt.resource_kind if rt else "physical"
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.namespaced_name = rname
|
|
r.namespace = rname.split("/")[0]
|
|
r.type_name = tname
|
|
r.resource_kind = kind
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_parent_resource_2 = r
|
|
|
|
|
|
@given('a child resource of type "{tname}"')
|
|
def step_resource_registry_create_child(context: Any, tname: str) -> None:
|
|
session: Session = context.rr_session
|
|
rid = _make_ulid("child1")
|
|
rt = session.get(ResourceTypeModel, tname)
|
|
kind = rt.resource_kind if rt else "physical"
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.type_name = tname
|
|
r.resource_kind = kind
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_child_resource = r
|
|
|
|
|
|
@given('a first child resource of type "{tname}"')
|
|
def step_resource_registry_create_first_child(context: Any, tname: str) -> None:
|
|
session: Session = context.rr_session
|
|
rid = _make_ulid("child_a")
|
|
rt = session.get(ResourceTypeModel, tname)
|
|
kind = rt.resource_kind if rt else "physical"
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.type_name = tname
|
|
r.resource_kind = kind
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_child_resource = r
|
|
|
|
|
|
@given('a second child resource of type "{tname}"')
|
|
def step_resource_registry_create_second_child(context: Any, tname: str) -> None:
|
|
session: Session = context.rr_session
|
|
rid = _make_ulid("child_b")
|
|
rt = session.get(ResourceTypeModel, tname)
|
|
kind = rt.resource_kind if rt else "physical"
|
|
r = ResourceModel()
|
|
r.resource_id = rid
|
|
r.type_name = tname
|
|
r.resource_kind = kind
|
|
r.created_at = _now_iso()
|
|
r.updated_at = _now_iso()
|
|
session.add(r)
|
|
session.commit()
|
|
context.rr_child_resource_2 = r
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ResourceEdgeModel CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('I create an edge from parent to child with link_type "{lt}"')
|
|
@when('I create an edge from parent to child with link_type "{lt}"')
|
|
def step_resource_registry_create_edge(context: Any, lt: str) -> None:
|
|
session: Session = context.rr_session
|
|
edge = ResourceEdgeModel()
|
|
edge.parent_id = context.rr_parent_resource.resource_id
|
|
edge.child_id = context.rr_child_resource.resource_id
|
|
edge.link_type = lt
|
|
edge.created_at = _now_iso()
|
|
session.add(edge)
|
|
session.commit()
|
|
context.rr_edge = edge
|
|
|
|
|
|
@when('I create an edge from second parent to child with link_type "{lt}"')
|
|
def step_resource_registry_create_edge_second_parent(context: Any, lt: str) -> None:
|
|
session: Session = context.rr_session
|
|
edge = ResourceEdgeModel()
|
|
edge.parent_id = context.rr_parent_resource_2.resource_id
|
|
edge.child_id = context.rr_child_resource.resource_id
|
|
edge.link_type = lt
|
|
edge.created_at = _now_iso()
|
|
session.add(edge)
|
|
session.commit()
|
|
|
|
|
|
@when("I create edges from parent to both children")
|
|
def step_resource_registry_create_edges_multi(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
e1 = ResourceEdgeModel()
|
|
e1.parent_id = context.rr_parent_resource.resource_id
|
|
e1.child_id = context.rr_child_resource.resource_id
|
|
e1.link_type = "contains"
|
|
e1.created_at = _now_iso()
|
|
session.add(e1)
|
|
|
|
e2 = ResourceEdgeModel()
|
|
e2.parent_id = context.rr_parent_resource.resource_id
|
|
e2.child_id = context.rr_child_resource_2.resource_id
|
|
e2.link_type = "contains"
|
|
e2.created_at = _now_iso()
|
|
session.add(e2)
|
|
session.commit()
|
|
|
|
|
|
@then("the edge exists in the database")
|
|
def step_resource_registry_edge_exists(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
edge = (
|
|
session.query(ResourceEdgeModel)
|
|
.filter(
|
|
ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id,
|
|
ResourceEdgeModel.child_id == context.rr_child_resource.resource_id,
|
|
)
|
|
.first()
|
|
)
|
|
assert edge is not None
|
|
|
|
|
|
@then('the edge link_type is "{expected}"')
|
|
def step_resource_registry_edge_link_type(context: Any, expected: str) -> None:
|
|
assert context.rr_edge.link_type == expected
|
|
|
|
|
|
@then("querying children of the parent returns {count:d} result")
|
|
def step_resource_registry_children_count_single(context: Any, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
edges = (
|
|
session.query(ResourceEdgeModel)
|
|
.filter(ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id)
|
|
.all()
|
|
)
|
|
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
|
|
|
|
|
|
@then("querying children of the parent returns {count:d} results")
|
|
def step_resource_registry_children_count(context: Any, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
edges = (
|
|
session.query(ResourceEdgeModel)
|
|
.filter(ResourceEdgeModel.parent_id == context.rr_parent_resource.resource_id)
|
|
.all()
|
|
)
|
|
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
|
|
|
|
|
|
@then("querying parents of the child returns {count:d} result")
|
|
def step_resource_registry_parents_count_single(context: Any, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
edges = (
|
|
session.query(ResourceEdgeModel)
|
|
.filter(ResourceEdgeModel.child_id == context.rr_child_resource.resource_id)
|
|
.all()
|
|
)
|
|
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
|
|
|
|
|
|
@then("querying parents of the child returns {count:d} results")
|
|
def step_resource_registry_parents_count(context: Any, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
edges = (
|
|
session.query(ResourceEdgeModel)
|
|
.filter(ResourceEdgeModel.child_id == context.rr_child_resource.resource_id)
|
|
.all()
|
|
)
|
|
assert len(edges) == count, f"Expected {count}, got {len(edges)}"
|
|
|
|
|
|
@then("creating a duplicate edge raises an integrity error")
|
|
def step_resource_registry_duplicate_edge(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
edge = ResourceEdgeModel()
|
|
edge.parent_id = context.rr_parent_resource.resource_id
|
|
edge.child_id = context.rr_child_resource.resource_id
|
|
edge.link_type = "contains"
|
|
edge.created_at = _now_iso()
|
|
try:
|
|
session.add(edge)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for duplicate edge"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then("creating a self-loop edge raises an integrity error")
|
|
def step_resource_registry_self_loop(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
parent_id = context.rr_parent_resource.resource_id
|
|
try:
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO resource_edges (parent_id, child_id, link_type, "
|
|
"auto_discovered, created_at) "
|
|
"VALUES (:pid, :cid, 'contains', 0, :ts)"
|
|
),
|
|
{"pid": parent_id, "cid": parent_id, "ts": _now_iso()},
|
|
)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for self-loop"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@then("creating an edge with invalid link_type raises an integrity error")
|
|
def step_resource_registry_invalid_link_type(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
try:
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO resource_edges (parent_id, child_id, link_type, "
|
|
"auto_discovered, created_at) "
|
|
"VALUES (:pid, :cid, 'invalid_type', 0, :ts)"
|
|
),
|
|
{
|
|
"pid": context.rr_parent_resource.resource_id,
|
|
"cid": context.rr_child_resource.resource_id,
|
|
"ts": _now_iso(),
|
|
},
|
|
)
|
|
session.commit()
|
|
msg = "Expected IntegrityError for invalid link_type"
|
|
raise AssertionError(msg)
|
|
except IntegrityError:
|
|
session.rollback()
|
|
|
|
|
|
@when("I delete the parent resource")
|
|
def step_resource_registry_delete_parent(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
r = session.get(ResourceModel, context.rr_parent_resource.resource_id)
|
|
assert r is not None
|
|
session.delete(r)
|
|
session.commit()
|
|
|
|
|
|
@when("I delete the child resource")
|
|
def step_resource_registry_delete_child(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
r = session.get(ResourceModel, context.rr_child_resource.resource_id)
|
|
assert r is not None
|
|
session.delete(r)
|
|
session.commit()
|
|
|
|
|
|
@then("the edge no longer exists in the database")
|
|
def step_resource_registry_edge_gone(context: Any) -> None:
|
|
session: Session = context.rr_session
|
|
count = session.query(ResourceEdgeModel).count()
|
|
assert count == 0, f"Expected 0 edges, got {count}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ORM relationship navigation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then('navigating from resource to resource_type returns "{expected}"')
|
|
def step_resource_registry_nav_to_type(context: Any, expected: str) -> None:
|
|
session: Session = context.rr_session
|
|
r = session.get(ResourceModel, context.rr_resource.resource_id)
|
|
assert r is not None
|
|
assert r.resource_type is not None
|
|
assert r.resource_type.name == expected
|
|
|
|
|
|
@then("navigating from resource_type to resources returns {count:d} items")
|
|
def step_resource_registry_nav_to_resources(context: Any, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
rt = session.get(ResourceTypeModel, "builtin/git-checkout")
|
|
assert rt is not None
|
|
assert len(rt.resources) == count, f"Expected {count}, got {len(rt.resources)}"
|
|
|
|
|
|
@then("navigating from parent to child_edges returns {count:d} edge")
|
|
def step_resource_registry_nav_child_edges(context: Any, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
r = session.get(ResourceModel, context.rr_parent_resource.resource_id)
|
|
assert r is not None
|
|
assert len(r.child_edges) == count, f"Expected {count}, got {len(r.child_edges)}"
|
|
|
|
|
|
@then("navigating from child to parent_edges returns {count:d} edge")
|
|
def step_resource_registry_nav_parent_edges(context: Any, count: int) -> None:
|
|
session: Session = context.rr_session
|
|
r = session.get(ResourceModel, context.rr_child_resource.resource_id)
|
|
assert r is not None
|
|
assert len(r.parent_edges) == count, f"Expected {count}, got {len(r.parent_edges)}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Timestamps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the resource type has valid created_at and updated_at timestamps")
|
|
def step_resource_registry_type_timestamps(context: Any) -> None:
|
|
rt = context.rr_resource_type
|
|
assert rt.created_at is not None
|
|
assert rt.updated_at is not None
|
|
# Parse to verify ISO-8601 format
|
|
datetime.fromisoformat(rt.created_at)
|
|
datetime.fromisoformat(rt.updated_at)
|
|
|
|
|
|
@then("the resource has valid created_at and updated_at timestamps")
|
|
def step_resource_registry_resource_timestamps(context: Any) -> None:
|
|
r = context.rr_resource
|
|
assert r.created_at is not None
|
|
assert r.updated_at is not None
|
|
datetime.fromisoformat(r.created_at)
|
|
datetime.fromisoformat(r.updated_at)
|
|
|
|
|
|
@then("the edge has a valid created_at timestamp")
|
|
def step_resource_registry_edge_timestamp(context: Any) -> None:
|
|
edge = context.rr_edge
|
|
assert edge.created_at is not None
|
|
datetime.fromisoformat(edge.created_at)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Migration smoke tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I run init_database to simulate migration")
|
|
def step_run_init_database(context: Any) -> None:
|
|
"""Run init_database on a fresh temp file to simulate migration."""
|
|
import tempfile
|
|
|
|
from cleveragents.infrastructure.database.models import init_database
|
|
|
|
tmp = tempfile.mktemp(suffix=".db")
|
|
db_url = f"sqlite:///{tmp}"
|
|
engine = init_database(db_url)
|
|
context.rr_migrated_engine = engine
|
|
context.rr_migrated_tmp = tmp
|
|
|
|
|
|
@when("I run init_database twice on the same database file")
|
|
def step_run_init_database_twice(context: Any) -> None:
|
|
"""Run init_database twice on the same file to verify idempotency."""
|
|
import tempfile
|
|
|
|
from cleveragents.infrastructure.database.models import init_database
|
|
|
|
tmp = tempfile.mktemp(suffix=".db")
|
|
db_url = f"sqlite:///{tmp}"
|
|
engine1 = init_database(db_url)
|
|
engine1.dispose()
|
|
engine2 = init_database(db_url)
|
|
context.rr_migrated_engine = engine2
|
|
context.rr_migrated_tmp = tmp
|
|
|
|
|
|
@then('the migrated database should contain table "{table_name}"')
|
|
def step_migrated_table_exists(context: Any, table_name: str) -> None:
|
|
inspector = inspect(context.rr_migrated_engine)
|
|
tables = inspector.get_table_names()
|
|
assert table_name in tables, f"Table {table_name} not found. Tables: {tables}"
|