forked from HAL9000/cleveragents-core
CoreRasurae
007af498b8
Renamed all 11 task-type confidence threshold fields in AutomationProfile from phase-transition semantics to spec-defined task-type semantics. Updated all 8 built-in profiles, CLI formatting, YAML schema, services, and all Behave/Robot tests referencing the old field names. Post-review fixes: - Fixed 24 stale old field names in M6 fixture files (automation_profiles.json, autonomy_guardrails.json) - Added model_validator(mode='before') to detect legacy field names and raise actionable ValueError with rename mapping - Added semantic bridge comments in PlanLifecycleService mapping task-type thresholds to phase-transition gates - Added threshold_field to structured log messages for observability - Restored categorised CLI automation-profile show output to match spec (Phase Transitions / Decision Automation / Self-Repair / Execution Controls) instead of flat list - Added missing access_network field to spec show output examples (Rich, Plain, JSON, YAML variants) - Aligned ADR-017 profile fields table to all 11 fields with descriptions matching spec Automatable Tasks table - Aligned automation_profiles.md threshold descriptions with spec - Added spec section references in phase_reversion.md, error_recovery.md, and plan_execute.md for field naming context - Extended repository roundtrip test to assert all 11 threshold fields - Fixed benchmark _make_profile() passing safety fields as top-level kwargs instead of via SafetyProfile sub-model (incompatible with extra="forbid") - Aligned CLI JSON/YAML output structure for automation-profile show with the specification grouped format (phase_transitions, decision_automation, self_repair, execution_controls) - Moved safety boolean fields into the Execution Controls section of Rich output per spec examples - Reverted auto profile description to "Fully automatic except apply" per specification (line 16703, line 28406) - Improved bridge comments in test steps with semantic context for threshold-to-gate mappings ISSUES CLOSED: #902
1517 lines
53 KiB
Python
1517 lines
53 KiB
Python
"""Step definitions for repositories_uncovered_lines.feature.
|
|
|
|
Targets the remaining ~84 uncovered lines in repositories.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.exc import IntegrityError, OperationalError
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
ResourceLinkModel,
|
|
ResourceModel,
|
|
ResourceTypeModel,
|
|
ToolBindingModel,
|
|
ToolModel,
|
|
)
|
|
from cleveragents.infrastructure.database.repositories import (
|
|
AutomationProfileNotFoundError,
|
|
AutomationProfileRepository,
|
|
AutomationProfileSchemaVersionError,
|
|
DatabaseError,
|
|
DuplicateAutomationProfileError,
|
|
DuplicateResourceTypeError,
|
|
DuplicateValidationAttachmentError,
|
|
InvalidToolTypeError,
|
|
LinkNotFoundError,
|
|
ResourceNotFoundRepoError,
|
|
ResourceRepository,
|
|
ResourceTypeRepository,
|
|
ToolRepository,
|
|
ValidationAttachmentRepository,
|
|
)
|
|
|
|
# ── helpers ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_engine_and_session():
|
|
"""Create an in-memory SQLite engine + session factory."""
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine)
|
|
return engine, factory
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(tz=UTC).isoformat()
|
|
|
|
|
|
def _make_rt_model_row(
|
|
name: str,
|
|
description: str = "test type",
|
|
resource_kind: str = "physical",
|
|
sandbox_strategy: str = "none",
|
|
user_addable: bool = True,
|
|
allowed_child_types_json: str = "[]",
|
|
auto_discover_json: str | None = None,
|
|
equivalence_json: str | None = None,
|
|
capabilities_json: str = '{"read": true, "write": true, "sandbox": true, "checkpoint": false}',
|
|
) -> ResourceTypeModel:
|
|
"""Create a ResourceTypeModel with correct column names."""
|
|
now = _now_iso()
|
|
ns = name.split("/")[0] if "/" in name else "builtin"
|
|
return ResourceTypeModel(
|
|
name=name,
|
|
namespace=ns,
|
|
description=description,
|
|
resource_kind=resource_kind,
|
|
sandbox_strategy=sandbox_strategy,
|
|
user_addable=user_addable,
|
|
handler_ref=None,
|
|
args_schema_json=None,
|
|
allowed_parent_types_json=None,
|
|
allowed_child_types_json=allowed_child_types_json,
|
|
auto_discover_json=auto_discover_json,
|
|
equivalence_json=equivalence_json,
|
|
capabilities_json=capabilities_json,
|
|
source=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
|
|
|
|
def _make_automation_profile(
|
|
name: str,
|
|
description: str = "test",
|
|
schema_version: str = "1.0",
|
|
**overrides: Any,
|
|
) -> Any:
|
|
"""Create an AutomationProfile domain object."""
|
|
from cleveragents.domain.models.core.automation_profile import AutomationProfile
|
|
from cleveragents.domain.models.core.safety_profile import SafetyProfile
|
|
|
|
# Extract safety-related overrides into a SafetyProfile.
|
|
safety_keys = {
|
|
"require_sandbox",
|
|
"require_checkpoints",
|
|
"allow_unsafe_tools",
|
|
}
|
|
safety_defaults: dict[str, Any] = {
|
|
"require_sandbox": True,
|
|
"require_checkpoints": True,
|
|
"allow_unsafe_tools": False,
|
|
"require_human_approval": False,
|
|
"max_retries_per_step": 3,
|
|
}
|
|
for key in safety_keys:
|
|
if key in overrides:
|
|
safety_defaults[key] = overrides.pop(key)
|
|
|
|
defaults = dict(
|
|
name=name,
|
|
description=description,
|
|
schema_version=schema_version,
|
|
decompose_task=0.0,
|
|
create_tool=0.0,
|
|
select_tool=0.0,
|
|
edit_code=0.0,
|
|
execute_command=0.0,
|
|
create_file=0.0,
|
|
delete_content=0.0,
|
|
access_network=0.0,
|
|
install_dependency=0.0,
|
|
modify_config=0.0,
|
|
approve_plan=0.0,
|
|
safety=SafetyProfile(**safety_defaults),
|
|
)
|
|
defaults.update(overrides)
|
|
return AutomationProfile(**defaults)
|
|
|
|
|
|
# ── LinkNotFoundError ──────────────────────────────────────────────────────
|
|
|
|
|
|
@given('a LinkNotFoundError for parent "{pid}" and child "{cid}"')
|
|
def step_given_link_not_found_error(context, pid, cid):
|
|
context.link_error = LinkNotFoundError(pid, cid)
|
|
|
|
|
|
@then('the LinkNotFoundError message should contain "{pid}" and "{cid}"')
|
|
def step_then_link_error_message(context, pid, cid):
|
|
msg = str(context.link_error)
|
|
assert pid in msg, f"Expected '{pid}' in '{msg}'"
|
|
assert cid in msg, f"Expected '{cid}' in '{msg}'"
|
|
|
|
|
|
@then('the LinkNotFoundError parent_id should be "{pid}"')
|
|
def step_then_link_error_parent(context, pid):
|
|
assert context.link_error.parent_id == pid
|
|
|
|
|
|
@then('the LinkNotFoundError child_id should be "{cid}"')
|
|
def step_then_link_error_child(context, cid):
|
|
assert context.link_error.child_id == cid
|
|
|
|
|
|
# ── ResourceTypeRepository helpers ─────────────────────────────────────────
|
|
|
|
|
|
@given("a resource type repository backed by an in-memory database for uncovered lines")
|
|
def step_given_rt_repo_inmem(context):
|
|
engine, factory = _make_engine_and_session()
|
|
context._ucl_engine = engine
|
|
context._ucl_session_factory = factory
|
|
context._ucl_rt_repo = ResourceTypeRepository(factory)
|
|
context._ucl_res_repo = ResourceRepository(factory)
|
|
|
|
|
|
def _make_resource_type_spec(name: str, **kw: Any) -> Any:
|
|
"""Create a ResourceTypeSpec domain object."""
|
|
from cleveragents.domain.models.core.resource_type import (
|
|
ResourceKind,
|
|
ResourceTypeSpec,
|
|
SandboxStrategy,
|
|
)
|
|
|
|
defaults = dict(
|
|
name=name,
|
|
description="test type",
|
|
resource_kind=ResourceKind.PHYSICAL,
|
|
sandbox_strategy=SandboxStrategy.NONE,
|
|
user_addable=True,
|
|
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,
|
|
)
|
|
defaults.update(kw)
|
|
return ResourceTypeSpec(**defaults)
|
|
|
|
|
|
@given('a resource type "{name}" exists in the database for uncovered lines')
|
|
def step_given_rt_exists(context, name):
|
|
spec = _make_resource_type_spec(name)
|
|
context._ucl_rt_repo.create(spec)
|
|
|
|
|
|
@when('the same resource type "{name}" is created again for uncovered lines')
|
|
def step_when_rt_dup_create(context, name):
|
|
spec = _make_resource_type_spec(name)
|
|
try:
|
|
context._ucl_rt_repo.create(spec)
|
|
context._ucl_error = None
|
|
except DuplicateResourceTypeError as exc:
|
|
context._ucl_error = exc
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("a DuplicateResourceTypeError should be raised for uncovered lines")
|
|
def step_then_dup_rt_error(context):
|
|
assert isinstance(context._ucl_error, DuplicateResourceTypeError), (
|
|
f"Expected DuplicateResourceTypeError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── ResourceTypeRepository._to_domain equivalence_json ──────────────────────
|
|
|
|
|
|
@given("a resource type row with equivalence_json set to '{json_str}'")
|
|
def step_given_rt_with_equiv(context, json_str):
|
|
# Insert a resource type row directly with equivalence_json set
|
|
session = context._ucl_session_factory()
|
|
now = _now_iso()
|
|
row = ResourceTypeModel(
|
|
name="test/equiv-type",
|
|
namespace="test",
|
|
description="type with equivalence",
|
|
resource_kind="physical",
|
|
sandbox_strategy="none",
|
|
user_addable=True,
|
|
handler_ref=None,
|
|
args_schema_json=None,
|
|
allowed_parent_types_json=None,
|
|
allowed_child_types_json="[]",
|
|
auto_discover_json=None,
|
|
equivalence_json=json_str,
|
|
capabilities_json='{"read": true, "write": true, "sandbox": true, "checkpoint": false}',
|
|
source=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
|
|
|
|
@when('the resource type is retrieved by name "{name}" for uncovered lines')
|
|
def step_when_rt_get_by_name(context, name):
|
|
context._ucl_rt_result = context._ucl_rt_repo.get(name)
|
|
|
|
|
|
@then('the resource type equivalence should contain key "{key}"')
|
|
def step_then_rt_equiv_key(context, key):
|
|
result = context._ucl_rt_result
|
|
assert result is not None, "Resource type was None"
|
|
equiv = result.equivalence
|
|
assert equiv is not None, "Equivalence was None"
|
|
assert key in equiv, f"Expected key '{key}' in equivalence {equiv}"
|
|
|
|
|
|
# ── ResourceRepository helpers ─────────────────────────────────────────────
|
|
|
|
|
|
@given("a resource repository backed by an in-memory database for uncovered lines")
|
|
def step_given_res_repo_inmem(context):
|
|
if not hasattr(context, "_ucl_engine"):
|
|
engine, factory = _make_engine_and_session()
|
|
context._ucl_engine = engine
|
|
context._ucl_session_factory = factory
|
|
context._ucl_rt_repo = ResourceTypeRepository(factory)
|
|
context._ucl_res_repo = ResourceRepository(context._ucl_session_factory)
|
|
|
|
|
|
def _ensure_resource_type(context, type_name: str = "test/default-type"):
|
|
"""Ensure a resource type exists."""
|
|
session = context._ucl_session_factory()
|
|
existing = session.query(ResourceTypeModel).filter_by(name=type_name).first()
|
|
if not existing:
|
|
now = _now_iso()
|
|
ns = type_name.split("/")[0] if "/" in type_name else "builtin"
|
|
row = ResourceTypeModel(
|
|
name=type_name,
|
|
namespace=ns,
|
|
description="test",
|
|
resource_kind="physical",
|
|
sandbox_strategy="none",
|
|
user_addable=True,
|
|
handler_ref=None,
|
|
args_schema_json=None,
|
|
allowed_parent_types_json=None,
|
|
allowed_child_types_json=None,
|
|
auto_discover_json=None,
|
|
capabilities_json='{"read": true, "write": true, "sandbox": true, "checkpoint": false}',
|
|
equivalence_json=None,
|
|
source=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
|
|
|
|
def _create_resource_row(
|
|
context,
|
|
resource_id: str | None = None,
|
|
type_name: str = "test/default-type",
|
|
namespace: str | None = None,
|
|
name: str | None = None,
|
|
sandbox_strategy: str | None = None,
|
|
) -> str:
|
|
"""Insert a resource row directly and return its ID."""
|
|
from ulid import ULID as _ULID
|
|
|
|
if resource_id is None:
|
|
resource_id = str(_ULID())
|
|
_ensure_resource_type(context, type_name)
|
|
session = context._ucl_session_factory()
|
|
now = _now_iso()
|
|
row = ResourceModel(
|
|
resource_id=resource_id,
|
|
namespaced_name=name,
|
|
namespace=namespace,
|
|
type_name=type_name,
|
|
resource_kind="physical",
|
|
location=None,
|
|
description="test resource",
|
|
read_only=False,
|
|
auto_discovered=False,
|
|
sandbox_strategy=sandbox_strategy,
|
|
content_hash=None,
|
|
properties_json=None,
|
|
metadata_json=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
return resource_id
|
|
|
|
|
|
# ── list_resources namespace filter ─────────────────────────────────────────
|
|
|
|
|
|
@given('resources exist in namespace "{ns1}" and "{ns2}" for uncovered lines')
|
|
def step_given_resources_in_namespaces(context, ns1, ns2):
|
|
from ulid import ULID
|
|
|
|
_create_resource_row(context, str(ULID()), namespace=ns1, name=f"{ns1}/res-a")
|
|
_create_resource_row(context, str(ULID()), namespace=ns1, name=f"{ns1}/res-b")
|
|
_create_resource_row(context, str(ULID()), namespace=ns2, name=f"{ns2}/res-a")
|
|
|
|
|
|
@when('resources are listed with namespace "{ns}" for uncovered lines')
|
|
def step_when_list_by_namespace(context, ns):
|
|
context._ucl_listed = context._ucl_res_repo.list_resources(namespace=ns)
|
|
|
|
|
|
@then('only resources from namespace "{ns}" should be returned')
|
|
def step_then_only_namespace_resources(context, ns):
|
|
resources = context._ucl_listed
|
|
assert len(resources) > 0, "Expected at least one resource"
|
|
for r in resources:
|
|
name = r.name or ""
|
|
assert name.startswith(ns), f"Resource '{name}' not in namespace '{ns}'"
|
|
|
|
|
|
# ── unlink_child - parent not found ────────────────────────────────────────
|
|
|
|
|
|
@when(
|
|
'unlink_child is called with non-existent parent "{pid}" and child "{cid}" for uncovered lines'
|
|
)
|
|
def step_when_unlink_parent_missing(context, pid, cid):
|
|
try:
|
|
context._ucl_res_repo.unlink_child(pid, cid)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("a ResourceNotFoundRepoError should be raised for the unlink parent")
|
|
def step_then_unlink_parent_error(context):
|
|
assert isinstance(context._ucl_error, ResourceNotFoundRepoError), (
|
|
f"Expected ResourceNotFoundRepoError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── unlink_child - child not found ────────────────────────────────────────
|
|
|
|
|
|
@given('a resource "{rid}" exists for uncovered unlink tests')
|
|
def step_given_resource_for_unlink(context, rid):
|
|
_create_resource_row(context, rid)
|
|
context._ucl_unlink_parent_id = rid
|
|
|
|
|
|
@when(
|
|
'unlink_child is called with existing parent and non-existent child "{cid}" for uncovered lines'
|
|
)
|
|
def step_when_unlink_child_missing(context, cid):
|
|
try:
|
|
context._ucl_res_repo.unlink_child(context._ucl_unlink_parent_id, cid)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("a ResourceNotFoundRepoError should be raised for the unlink child")
|
|
def step_then_unlink_child_error(context):
|
|
assert isinstance(context._ucl_error, ResourceNotFoundRepoError), (
|
|
f"Expected ResourceNotFoundRepoError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── unlink_child - link not found ─────────────────────────────────────────
|
|
|
|
|
|
@given('two resources "{pid}" and "{cid}" exist but are not linked for uncovered lines')
|
|
def step_given_two_unlinked_resources(context, pid, cid):
|
|
_create_resource_row(context, pid)
|
|
_create_resource_row(context, cid)
|
|
context._ucl_unlink_pid = pid
|
|
context._ucl_unlink_cid = cid
|
|
|
|
|
|
@when('unlink_child is called for "{pid}" and "{cid}" for uncovered lines')
|
|
def step_when_unlink_no_link(context, pid, cid):
|
|
try:
|
|
context._ucl_res_repo.unlink_child(pid, cid)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("a LinkNotFoundError should be raised for uncovered lines")
|
|
def step_then_link_not_found_error(context):
|
|
assert isinstance(context._ucl_error, LinkNotFoundError), (
|
|
f"Expected LinkNotFoundError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── unlink_child - re-raise domain errors ─────────────────────────────────
|
|
|
|
|
|
@when(
|
|
'unlink_child is called with non-existent parent "{pid}" and child "{cid}" for uncovered lines re-raise'
|
|
)
|
|
def step_when_unlink_reraise(context, pid, cid):
|
|
try:
|
|
context._ucl_res_repo.unlink_child(pid, cid)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("a ResourceNotFoundRepoError should be raised and re-raised for uncovered lines")
|
|
def step_then_unlink_reraise(context):
|
|
assert isinstance(context._ucl_error, ResourceNotFoundRepoError), (
|
|
f"Expected ResourceNotFoundRepoError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── auto_discover_children - type_row is None ─────────────────────────────
|
|
|
|
|
|
@given(
|
|
'a resource with type "missing-type" that has no type row in DB for uncovered lines'
|
|
)
|
|
def step_given_res_missing_type(context):
|
|
# Insert resource with a type_name that has no matching ResourceTypeModel
|
|
session = context._ucl_session_factory()
|
|
now = _now_iso()
|
|
row = ResourceModel(
|
|
resource_id="auto-disc-no-type",
|
|
namespaced_name=None,
|
|
namespace=None,
|
|
type_name="missing-type",
|
|
resource_kind="physical",
|
|
location=None,
|
|
description="test",
|
|
read_only=False,
|
|
auto_discovered=False,
|
|
sandbox_strategy=None,
|
|
content_hash=None,
|
|
properties_json=None,
|
|
metadata_json=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
context._ucl_auto_disc_id = "auto-disc-no-type"
|
|
|
|
|
|
@when("auto_discover_children is called for that resource for uncovered lines")
|
|
def step_when_auto_disc_no_type(context):
|
|
context._ucl_auto_disc_result = context._ucl_res_repo.auto_discover_children(
|
|
context._ucl_auto_disc_id
|
|
)
|
|
|
|
|
|
@then("an empty list should be returned from auto_discover for uncovered lines")
|
|
def step_then_auto_disc_empty(context):
|
|
assert context._ucl_auto_disc_result == [], (
|
|
f"Expected empty list, got {context._ucl_auto_disc_result}"
|
|
)
|
|
|
|
|
|
# ── auto_discover_children - null auto_discover_json ──────────────────────
|
|
|
|
|
|
@given("a resource with type that has null auto_discover_json for uncovered lines")
|
|
def step_given_res_null_auto_disc(context):
|
|
_ensure_resource_type(context, "test/null-auto-disc")
|
|
_create_resource_row(context, "auto-disc-null", type_name="test/null-auto-disc")
|
|
context._ucl_auto_disc_id_null = "auto-disc-null"
|
|
|
|
|
|
@when(
|
|
"auto_discover_children is called for that typed resource for uncovered lines null"
|
|
)
|
|
def step_when_auto_disc_null(context):
|
|
context._ucl_auto_disc_result = context._ucl_res_repo.auto_discover_children(
|
|
context._ucl_auto_disc_id_null
|
|
)
|
|
|
|
|
|
# ── auto_discover_children - disabled ─────────────────────────────────────
|
|
|
|
|
|
@given("a resource with type that has auto_discover disabled for uncovered lines")
|
|
def step_given_res_disabled_auto_disc(context):
|
|
session = context._ucl_session_factory()
|
|
now = _now_iso()
|
|
row = ResourceTypeModel(
|
|
name="test/disabled-auto-disc",
|
|
namespace="test",
|
|
description="disabled auto-disc",
|
|
resource_kind="physical",
|
|
sandbox_strategy="none",
|
|
user_addable=True,
|
|
handler_ref=None,
|
|
args_schema_json=None,
|
|
allowed_parent_types_json=None,
|
|
allowed_child_types_json="[]",
|
|
auto_discover_json=json.dumps({"enabled": False, "rules": []}),
|
|
equivalence_json=None,
|
|
capabilities_json='{"read": true, "write": true, "sandbox": true, "checkpoint": false}',
|
|
source=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
_create_resource_row(
|
|
context, "auto-disc-disabled", type_name="test/disabled-auto-disc"
|
|
)
|
|
context._ucl_auto_disc_id_disabled = "auto-disc-disabled"
|
|
|
|
|
|
@when(
|
|
"auto_discover_children is called for that typed resource for uncovered lines disabled"
|
|
)
|
|
def step_when_auto_disc_disabled(context):
|
|
context._ucl_auto_disc_result = context._ucl_res_repo.auto_discover_children(
|
|
context._ucl_auto_disc_id_disabled
|
|
)
|
|
|
|
|
|
# ── auto_discover_children - empty rules ──────────────────────────────────
|
|
|
|
|
|
@given(
|
|
"a resource with type that has auto_discover enabled but empty rules for uncovered lines"
|
|
)
|
|
def step_given_res_empty_rules(context):
|
|
session = context._ucl_session_factory()
|
|
_now_iso()
|
|
row = _make_rt_model_row(
|
|
"test/empty-rules",
|
|
description="empty rules",
|
|
auto_discover_json=json.dumps({"enabled": True, "rules": []}),
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
_create_resource_row(context, "auto-disc-empty-rules", type_name="test/empty-rules")
|
|
context._ucl_auto_disc_id_empty = "auto-disc-empty-rules"
|
|
|
|
|
|
@when(
|
|
"auto_discover_children is called for that typed resource for uncovered lines empty rules"
|
|
)
|
|
def step_when_auto_disc_empty_rules(context):
|
|
context._ucl_auto_disc_result = context._ucl_res_repo.auto_discover_children(
|
|
context._ucl_auto_disc_id_empty
|
|
)
|
|
|
|
|
|
# ── auto_discover_children - child type not in DB ─────────────────────────
|
|
|
|
|
|
@given(
|
|
"a resource with auto_discover rules referencing non-existent child type for uncovered lines"
|
|
)
|
|
def step_given_res_no_child_type(context):
|
|
session = context._ucl_session_factory()
|
|
_now_iso()
|
|
row = _make_rt_model_row(
|
|
"test/no-child-type",
|
|
description="missing child type",
|
|
auto_discover_json=json.dumps(
|
|
{
|
|
"enabled": True,
|
|
"rules": [{"type": "nonexistent/child-type"}],
|
|
}
|
|
),
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
_create_resource_row(context, "auto-disc-no-child", type_name="test/no-child-type")
|
|
context._ucl_auto_disc_id_no_child = "auto-disc-no-child"
|
|
|
|
|
|
@when(
|
|
"auto_discover_children is called for that resource with missing child type for uncovered lines"
|
|
)
|
|
def step_when_auto_disc_no_child(context):
|
|
context._ucl_auto_disc_result = context._ucl_res_repo.auto_discover_children(
|
|
context._ucl_auto_disc_id_no_child
|
|
)
|
|
|
|
|
|
# ── auto_discover_children - child type not in allowed list ───────────────
|
|
|
|
|
|
@given(
|
|
"a resource with auto_discover rules where child type is not in allowed list for uncovered lines"
|
|
)
|
|
def step_given_res_disallowed_child(context):
|
|
session = context._ucl_session_factory()
|
|
_now_iso()
|
|
# Create the child type in DB
|
|
child_type = _make_rt_model_row("test/disallowed-child", description="child type")
|
|
session.add(child_type)
|
|
session.commit()
|
|
|
|
# Create parent type that allows only "test/other-child", not "test/disallowed-child"
|
|
session2 = context._ucl_session_factory()
|
|
parent_type = _make_rt_model_row(
|
|
"test/disallow-parent",
|
|
description="parent type disallowing child",
|
|
allowed_child_types_json=json.dumps(["test/other-child"]),
|
|
auto_discover_json=json.dumps(
|
|
{
|
|
"enabled": True,
|
|
"rules": [{"type": "test/disallowed-child"}],
|
|
}
|
|
),
|
|
)
|
|
session2.add(parent_type)
|
|
session2.commit()
|
|
|
|
_create_resource_row(
|
|
context, "auto-disc-disallowed", type_name="test/disallow-parent"
|
|
)
|
|
context._ucl_auto_disc_id_disallowed = "auto-disc-disallowed"
|
|
|
|
|
|
@when(
|
|
"auto_discover_children is called for that resource with disallowed child type for uncovered lines"
|
|
)
|
|
def step_when_auto_disc_disallowed(context):
|
|
context._ucl_auto_disc_result = context._ucl_res_repo.auto_discover_children(
|
|
context._ucl_auto_disc_id_disallowed
|
|
)
|
|
|
|
|
|
# ── auto_discover_children - successful discovery ─────────────────────────
|
|
|
|
|
|
@given(
|
|
"a resource with auto_discover rules that match an existing child type for uncovered lines"
|
|
)
|
|
def step_given_res_full_discovery(context):
|
|
session = context._ucl_session_factory()
|
|
_now_iso()
|
|
# Create the child type
|
|
existing_ct = (
|
|
session.query(ResourceTypeModel).filter_by(name="test/valid-child").first()
|
|
)
|
|
if not existing_ct:
|
|
child_type = _make_rt_model_row(
|
|
"test/valid-child", description="valid child type"
|
|
)
|
|
session.add(child_type)
|
|
session.commit()
|
|
|
|
session2 = context._ucl_session_factory()
|
|
existing_pt = (
|
|
session2.query(ResourceTypeModel).filter_by(name="test/disc-parent").first()
|
|
)
|
|
if not existing_pt:
|
|
parent_type = _make_rt_model_row(
|
|
"test/disc-parent",
|
|
description="parent for discovery",
|
|
allowed_child_types_json=json.dumps(["test/valid-child"]),
|
|
auto_discover_json=json.dumps(
|
|
{
|
|
"enabled": True,
|
|
"rules": [{"type": "test/valid-child"}],
|
|
}
|
|
),
|
|
)
|
|
session2.add(parent_type)
|
|
session2.commit()
|
|
|
|
_create_resource_row(context, "auto-disc-full", type_name="test/disc-parent")
|
|
context._ucl_auto_disc_full_id = "auto-disc-full"
|
|
|
|
|
|
@when(
|
|
"auto_discover_children is called for that resource for full discovery for uncovered lines"
|
|
)
|
|
def step_when_auto_disc_full(context):
|
|
context._ucl_auto_disc_result = context._ucl_res_repo.auto_discover_children(
|
|
context._ucl_auto_disc_full_id
|
|
)
|
|
|
|
|
|
@then("the discovered children list should not be empty for uncovered lines")
|
|
def step_then_discovered_not_empty(context):
|
|
assert len(context._ucl_auto_disc_result) > 0, (
|
|
"Expected non-empty discovered children"
|
|
)
|
|
|
|
|
|
@then("the child resource should be linked to the parent for uncovered lines")
|
|
def step_then_child_linked(context):
|
|
session = context._ucl_session_factory()
|
|
links = (
|
|
session.query(ResourceLinkModel)
|
|
.filter_by(parent_id=context._ucl_auto_disc_full_id)
|
|
.all()
|
|
)
|
|
assert len(links) > 0, "Expected at least one link from parent to child"
|
|
|
|
|
|
# ── auto_discover_children - OperationalError ─────────────────────────────
|
|
|
|
|
|
@given(
|
|
"a resource repository with session raising OperationalError on auto_discover for uncovered lines"
|
|
)
|
|
def step_given_res_repo_op_error_auto_disc(context):
|
|
call_count = 0
|
|
|
|
def mock_session_factory():
|
|
nonlocal call_count
|
|
call_count += 1
|
|
mock_session = MagicMock(spec=Session)
|
|
# First query returns the resource row
|
|
parent_row = MagicMock()
|
|
parent_row.type_name = "some-type"
|
|
parent_row.resource_id = "res-op-err"
|
|
|
|
type_row = MagicMock()
|
|
type_row.auto_discover_json = json.dumps(
|
|
{"enabled": True, "rules": [{"type": "child-t"}]}
|
|
)
|
|
type_row.allowed_child_types_json = "[]"
|
|
|
|
# query().filter_by().first() returns parent_row first, then type_row, then raises
|
|
query_mock = MagicMock()
|
|
filter_mock = MagicMock()
|
|
filter_mock.first.side_effect = [
|
|
parent_row,
|
|
type_row,
|
|
OperationalError("db fail", {}, None),
|
|
]
|
|
query_mock.filter_by.return_value = filter_mock
|
|
mock_session.query.return_value = query_mock
|
|
mock_session.rollback = MagicMock()
|
|
return mock_session
|
|
|
|
context._ucl_res_repo = ResourceRepository(mock_session_factory)
|
|
context._ucl_auto_disc_op_err_id = "res-op-err"
|
|
|
|
|
|
@when(
|
|
"auto_discover_children is called and an OperationalError occurs for uncovered lines"
|
|
)
|
|
def step_when_auto_disc_op_error(context):
|
|
try:
|
|
context._ucl_res_repo.auto_discover_children(context._ucl_auto_disc_op_err_id)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("a DatabaseError should be raised mentioning auto-discover for uncovered lines")
|
|
def step_then_auto_disc_db_error(context):
|
|
assert isinstance(context._ucl_error, DatabaseError), (
|
|
f"Expected DatabaseError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
assert (
|
|
"auto-discover" in str(context._ucl_error).lower()
|
|
or "auto_discover" in str(context._ucl_error).lower()
|
|
), f"Expected 'auto-discover' in message: {context._ucl_error}"
|
|
|
|
|
|
# ── _get_ancestors - diamond graph ────────────────────────────────────────
|
|
|
|
|
|
@given("resources forming a diamond graph for ancestor traversal for uncovered lines")
|
|
def step_given_diamond_graph(context):
|
|
# Create: A -> B, A -> C, B -> D, C -> D (diamond)
|
|
# _get_ancestors(D) should visit D, B, C, A
|
|
for rid in ["anc-A", "anc-B", "anc-C", "anc-D"]:
|
|
_create_resource_row(context, rid)
|
|
|
|
session = context._ucl_session_factory()
|
|
now = _now_iso()
|
|
links = [
|
|
ResourceLinkModel(parent_id="anc-A", child_id="anc-B", created_at=now),
|
|
ResourceLinkModel(parent_id="anc-A", child_id="anc-C", created_at=now),
|
|
ResourceLinkModel(parent_id="anc-B", child_id="anc-D", created_at=now),
|
|
ResourceLinkModel(parent_id="anc-C", child_id="anc-D", created_at=now),
|
|
]
|
|
for link in links:
|
|
session.add(link)
|
|
session.commit()
|
|
context._ucl_ancestor_id = "anc-D"
|
|
|
|
|
|
@when("_get_ancestors is called on the bottom resource for uncovered lines")
|
|
def step_when_get_ancestors(context):
|
|
session = context._ucl_session_factory()
|
|
context._ucl_ancestors = ResourceRepository._get_ancestors(
|
|
session, context._ucl_ancestor_id
|
|
)
|
|
|
|
|
|
@then(
|
|
"all ancestor IDs should be returned including the bottom resource for uncovered lines"
|
|
)
|
|
def step_then_ancestors_returned(context):
|
|
expected = {"anc-A", "anc-B", "anc-C", "anc-D"}
|
|
assert context._ucl_ancestors == expected, (
|
|
f"Expected {expected}, got {context._ucl_ancestors}"
|
|
)
|
|
|
|
|
|
# ── ResourceRepository._to_domain - sandbox_strategy ─────────────────────
|
|
|
|
|
|
@given('a resource row with sandbox_strategy set to "git_worktree" for uncovered lines')
|
|
def step_given_res_with_sandbox(context):
|
|
from ulid import ULID
|
|
|
|
rid = str(ULID())
|
|
_create_resource_row(
|
|
context,
|
|
rid,
|
|
sandbox_strategy="git_worktree",
|
|
)
|
|
context._ucl_sandbox_id = rid
|
|
|
|
|
|
@when("the resource is retrieved by ID for uncovered lines sandbox")
|
|
def step_when_res_get_sandbox(context):
|
|
context._ucl_sandbox_result = context._ucl_res_repo.get(context._ucl_sandbox_id)
|
|
|
|
|
|
@then("the resource sandbox_strategy should be SandboxStrategy.GIT_WORKTREE")
|
|
def step_then_sandbox_strategy(context):
|
|
from cleveragents.domain.models.core.resource import SandboxStrategy
|
|
|
|
result = context._ucl_sandbox_result
|
|
assert result is not None, "Resource was None"
|
|
assert result.sandbox_strategy == SandboxStrategy.GIT_WORKTREE, (
|
|
f"Expected GIT_WORKTREE, got {result.sandbox_strategy}"
|
|
)
|
|
|
|
|
|
# ── ToolRepository helpers ────────────────────────────────────────────────
|
|
|
|
|
|
@given("a tool repository backed by an in-memory database for uncovered lines")
|
|
def step_given_tool_repo_inmem(context):
|
|
engine, factory = _make_engine_and_session()
|
|
context._ucl_tool_engine = engine
|
|
context._ucl_tool_session_factory = factory
|
|
context._ucl_tool_repo = ToolRepository(factory)
|
|
|
|
|
|
def _make_tool_domain(
|
|
name: str = "test/my-tool",
|
|
tool_type: str = "tool",
|
|
source: str = "builtin",
|
|
input_schema: dict | None = None,
|
|
output_schema: dict | None = None,
|
|
resource_slots: list | None = None,
|
|
) -> Any:
|
|
"""Create a mock tool domain object."""
|
|
tool = MagicMock()
|
|
tool.name = name
|
|
tool.description = "A test tool"
|
|
|
|
# tool_type enum-like
|
|
tt = MagicMock()
|
|
tt.value = tool_type
|
|
tool.tool_type = tt
|
|
|
|
# source enum-like
|
|
src = MagicMock()
|
|
src.value = source
|
|
tool.source = src
|
|
|
|
tool.input_schema = input_schema
|
|
tool.output_schema = output_schema
|
|
tool.config_yaml = None
|
|
tool.tool_id = ""
|
|
|
|
cap = MagicMock()
|
|
cap.read_only = False
|
|
cap.writes = True
|
|
cap.checkpointable = False
|
|
cap.side_effects = False
|
|
tool.capability = cap
|
|
|
|
tool.resource_slots = resource_slots or []
|
|
return tool
|
|
|
|
|
|
# ── ToolRepository.add - invalid tool_type ────────────────────────────────
|
|
|
|
|
|
@given('a tool domain object with tool_type "bogus" for uncovered lines')
|
|
def step_given_tool_bad_type(context):
|
|
context._ucl_tool = _make_tool_domain(tool_type="bogus")
|
|
|
|
|
|
@when("the tool is added to the repository for uncovered lines invalid type")
|
|
def step_when_tool_add_invalid(context):
|
|
try:
|
|
context._ucl_tool_repo.add(context._ucl_tool)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("an InvalidToolTypeError should be raised for uncovered lines")
|
|
def step_then_invalid_tool_type(context):
|
|
assert isinstance(context._ucl_error, InvalidToolTypeError), (
|
|
f"Expected InvalidToolTypeError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── ToolRepository.add - schemas + resource slots ─────────────────────────
|
|
|
|
|
|
@given("a tool domain object with schemas and resource slots for uncovered lines")
|
|
def step_given_tool_with_schemas(context):
|
|
slot = MagicMock()
|
|
slot.name = "my_slot"
|
|
slot.resource_type = "test/resource-type"
|
|
|
|
binding = MagicMock()
|
|
binding.value = "contextual"
|
|
slot.binding = binding
|
|
|
|
access = MagicMock()
|
|
access.value = "read_only"
|
|
slot.access = access
|
|
|
|
context._ucl_tool = _make_tool_domain(
|
|
name="test/schema-tool",
|
|
input_schema={"type": "object", "properties": {"x": {"type": "string"}}},
|
|
output_schema={"type": "object", "properties": {"y": {"type": "integer"}}},
|
|
resource_slots=[slot],
|
|
)
|
|
|
|
|
|
@when("the tool is added to the repository for uncovered lines with schemas")
|
|
def step_when_tool_add_schemas(context):
|
|
context._ucl_tool_id = context._ucl_tool_repo.add(context._ucl_tool)
|
|
|
|
|
|
@then(
|
|
"the tool should be retrievable and have schemas and bindings for uncovered lines"
|
|
)
|
|
def step_then_tool_schemas_bindings(context):
|
|
session = context._ucl_tool_session_factory()
|
|
row = session.query(ToolModel).filter_by(name=context._ucl_tool_id).first()
|
|
assert row is not None, "Tool row not found"
|
|
assert row.input_schema_json is not None, "input_schema_json should not be None"
|
|
assert row.output_schema_json is not None, "output_schema_json should not be None"
|
|
|
|
bindings = (
|
|
session.query(ToolBindingModel).filter_by(tool_name=context._ucl_tool_id).all()
|
|
)
|
|
assert len(bindings) == 1, f"Expected 1 binding, got {len(bindings)}"
|
|
assert bindings[0].slot_name == "my_slot"
|
|
assert bindings[0].binding_mode == "contextual"
|
|
|
|
|
|
# ── ToolRepository.get - by ULID ──────────────────────────────────────────
|
|
|
|
|
|
@given("a tool has been persisted and its ULID captured for uncovered lines")
|
|
def step_given_tool_persisted(context):
|
|
tool = _make_tool_domain(name="test/get-tool")
|
|
context._ucl_persisted_tool_id = context._ucl_tool_repo.add(tool)
|
|
|
|
|
|
@when("the tool is fetched by ULID for uncovered lines")
|
|
def step_when_tool_get_by_ulid(context):
|
|
context._ucl_tool_result = context._ucl_tool_repo.get(
|
|
context._ucl_persisted_tool_id
|
|
)
|
|
|
|
|
|
@then("the tool domain object should be returned for uncovered lines")
|
|
def step_then_tool_returned(context):
|
|
assert context._ucl_tool_result is not None, "Tool was None"
|
|
result = context._ucl_tool_result
|
|
name = result["name"] if isinstance(result, dict) else result.name
|
|
assert name == "test/get-tool"
|
|
|
|
|
|
@when('a tool is fetched by ULID "{ulid}" for uncovered lines')
|
|
def step_when_tool_get_unknown(context, ulid):
|
|
context._ucl_tool_result = context._ucl_tool_repo.get(ulid)
|
|
|
|
|
|
@then("None should be returned for the tool get for uncovered lines")
|
|
def step_then_tool_none(context):
|
|
assert context._ucl_tool_result is None, (
|
|
f"Expected None, got {context._ucl_tool_result}"
|
|
)
|
|
|
|
|
|
# ── ToolRepository._to_domain - input/output schema ──────────────────────
|
|
|
|
|
|
@given("a tool with input_schema and output_schema stored as JSON for uncovered lines")
|
|
def step_given_tool_with_json_schemas(context):
|
|
tool = _make_tool_domain(
|
|
name="test/json-schema-tool",
|
|
input_schema={"type": "object"},
|
|
output_schema={"type": "array"},
|
|
)
|
|
context._ucl_tool_repo.add(tool)
|
|
context._ucl_schema_tool_name = "test/json-schema-tool"
|
|
|
|
|
|
@when("the tool is retrieved by name for uncovered lines schemas")
|
|
def step_when_tool_get_by_name_schemas(context):
|
|
context._ucl_schema_tool = context._ucl_tool_repo.get_by_name(
|
|
context._ucl_schema_tool_name
|
|
)
|
|
|
|
|
|
@then("the tool should have parsed input_schema and output_schema for uncovered lines")
|
|
def step_then_tool_parsed_schemas(context):
|
|
import json as _json
|
|
|
|
tool = context._ucl_schema_tool
|
|
assert tool is not None, "Tool was None"
|
|
# HEAD's to_domain() returns a dict with JSON strings
|
|
if isinstance(tool, dict):
|
|
input_raw = tool.get("input_schema_json")
|
|
output_raw = tool.get("output_schema_json")
|
|
assert input_raw is not None, "input_schema_json should not be None"
|
|
assert output_raw is not None, "output_schema_json should not be None"
|
|
input_schema = (
|
|
_json.loads(input_raw) if isinstance(input_raw, str) else input_raw
|
|
)
|
|
output_schema = (
|
|
_json.loads(output_raw) if isinstance(output_raw, str) else output_raw
|
|
)
|
|
else:
|
|
input_schema = tool.input_schema
|
|
output_schema = tool.output_schema
|
|
assert isinstance(input_schema, dict), f"Expected dict, got {type(input_schema)}"
|
|
assert isinstance(output_schema, dict), f"Expected dict, got {type(output_schema)}"
|
|
|
|
|
|
# ── ValidationAttachment - duplicate detection ────────────────────────────
|
|
|
|
|
|
@given(
|
|
"a validation attachment repository backed by an in-memory database for uncovered lines"
|
|
)
|
|
def step_given_va_repo(context):
|
|
engine, factory = _make_engine_and_session()
|
|
context._ucl_va_engine = engine
|
|
context._ucl_va_session_factory = factory
|
|
context._ucl_va_repo = ValidationAttachmentRepository(factory)
|
|
# We need a resource to attach to
|
|
_ensure_va_resource(context, factory)
|
|
|
|
|
|
def _ensure_va_resource(context, factory):
|
|
"""Create a resource to use in validation attachment tests."""
|
|
session = factory()
|
|
now = _now_iso()
|
|
# Resource type first
|
|
rt = _make_rt_model_row("test/va-type", description="for VA tests")
|
|
session.add(rt)
|
|
session.flush()
|
|
for rid in ["res-1", "res-2"]:
|
|
res = ResourceModel(
|
|
resource_id=rid,
|
|
namespaced_name=None,
|
|
namespace=None,
|
|
type_name="test/va-type",
|
|
resource_kind="physical",
|
|
location=None,
|
|
description="VA test resource",
|
|
read_only=False,
|
|
auto_discovered=False,
|
|
sandbox_strategy=None,
|
|
content_hash=None,
|
|
properties_json=None,
|
|
metadata_json=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(res)
|
|
session.commit()
|
|
|
|
|
|
@given(
|
|
'a validation "{vname}" is already attached to resource "{rid}" for uncovered lines'
|
|
)
|
|
def step_given_va_attached(context, vname, rid):
|
|
context._ucl_va_repo.attach(vname, rid, "required")
|
|
context._ucl_va_rid = rid
|
|
context._ucl_va_vname = vname
|
|
|
|
|
|
@when(
|
|
'the same validation "{vname}" is attached again to resource "{rid}" for uncovered lines'
|
|
)
|
|
def step_when_va_dup_attach(context, vname, rid):
|
|
try:
|
|
context._ucl_va_repo.attach(vname, rid, "required")
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then(
|
|
"a DuplicateValidationAttachmentError should be raised for uncovered lines attach"
|
|
)
|
|
def step_then_va_dup_error(context):
|
|
assert isinstance(context._ucl_error, DuplicateValidationAttachmentError), (
|
|
f"Expected DuplicateValidationAttachmentError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── ValidationAttachment - re-raise ───────────────────────────────────────
|
|
|
|
|
|
@when(
|
|
'the same validation "{vname}" is attached again to resource "{rid}" for uncovered lines re-raise'
|
|
)
|
|
def step_when_va_dup_reraise(context, vname, rid):
|
|
try:
|
|
context._ucl_va_repo.attach(vname, rid, "required")
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("the DuplicateValidationAttachmentError should propagate for uncovered lines")
|
|
def step_then_va_dup_propagate(context):
|
|
assert isinstance(context._ucl_error, DuplicateValidationAttachmentError), (
|
|
f"Expected DuplicateValidationAttachmentError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── AutomationProfileRepository ──────────────────────────────────────────
|
|
|
|
|
|
@given(
|
|
"an automation profile repository backed by an in-memory database for uncovered lines"
|
|
)
|
|
def step_given_ap_repo(context):
|
|
engine, factory = _make_engine_and_session()
|
|
context._ucl_ap_engine = engine
|
|
context._ucl_ap_session_factory = factory
|
|
context._ucl_ap_repo = AutomationProfileRepository(factory)
|
|
|
|
|
|
@when('a profile is fetched by name "{name}" for uncovered lines')
|
|
def step_when_ap_get(context, name):
|
|
context._ucl_ap_result = context._ucl_ap_repo.get_by_name(name)
|
|
|
|
|
|
@then("None should be returned for the profile get for uncovered lines")
|
|
def step_then_ap_none(context):
|
|
assert context._ucl_ap_result is None, (
|
|
f"Expected None, got {context._ucl_ap_result}"
|
|
)
|
|
|
|
|
|
# ── get_by_name returns domain ────────────────────────────────────────────
|
|
|
|
|
|
@given('a profile "{name}" has been upserted for uncovered lines')
|
|
def step_given_ap_upserted(context, name):
|
|
profile = _make_automation_profile(name)
|
|
context._ucl_ap_repo.upsert(profile)
|
|
|
|
|
|
@then("the profile domain object should be returned for uncovered lines")
|
|
def step_then_ap_domain(context):
|
|
assert context._ucl_ap_result is not None, "Profile was None"
|
|
assert context._ucl_ap_result.name == "test-profile"
|
|
|
|
|
|
# ── list_all ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
@given('profiles "{name1}" and "{name2}" have been upserted for uncovered lines')
|
|
def step_given_multiple_profiles(context, name1, name2):
|
|
for name in [name1, name2]:
|
|
profile = _make_automation_profile(name)
|
|
context._ucl_ap_repo.upsert(profile)
|
|
|
|
|
|
@when("all profiles are listed for uncovered lines")
|
|
def step_when_ap_list(context):
|
|
context._ucl_ap_list = context._ucl_ap_repo.list_all()
|
|
|
|
|
|
@then("two profile domain objects should be returned for uncovered lines")
|
|
def step_then_ap_list_count(context):
|
|
assert len(context._ucl_ap_list) == 2, (
|
|
f"Expected 2, got {len(context._ucl_ap_list)}"
|
|
)
|
|
|
|
|
|
# ── upsert update ────────────────────────────────────────────────────────
|
|
|
|
|
|
@when(
|
|
'the profile "{name}" is upserted again with changed description for uncovered lines'
|
|
)
|
|
def step_when_ap_update(context, name):
|
|
profile = _make_automation_profile(name, description="updated desc")
|
|
context._ucl_ap_repo.upsert(profile)
|
|
|
|
|
|
@then("the profile should have the updated description for uncovered lines")
|
|
def step_then_ap_updated_desc(context):
|
|
result = context._ucl_ap_repo.get_by_name("update-prof")
|
|
assert result is not None, "Profile was None"
|
|
assert result.description == "updated desc", (
|
|
f"Expected 'updated desc', got '{result.description}'"
|
|
)
|
|
|
|
|
|
# ── upsert schema version mismatch ──────────────────────────────────────
|
|
|
|
|
|
@given(
|
|
'a profile "{name}" has been upserted with schema_version "{ver}" for uncovered lines'
|
|
)
|
|
def step_given_ap_versioned(context, name, ver):
|
|
profile = _make_automation_profile(name, schema_version=ver)
|
|
context._ucl_ap_repo.upsert(profile)
|
|
|
|
|
|
@when(
|
|
'the profile "{name}" is upserted with expected_schema_version "{ver}" for uncovered lines'
|
|
)
|
|
def step_when_ap_version_mismatch(context, name, ver):
|
|
profile = _make_automation_profile(name, schema_version="1.0")
|
|
try:
|
|
context._ucl_ap_repo.upsert(profile, expected_schema_version=ver)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("an AutomationProfileSchemaVersionError should be raised for uncovered lines")
|
|
def step_then_ap_version_error(context):
|
|
assert isinstance(context._ucl_error, AutomationProfileSchemaVersionError), (
|
|
f"Expected AutomationProfileSchemaVersionError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── upsert new insert ────────────────────────────────────────────────────
|
|
|
|
|
|
@when('a new profile "{name}" is upserted for uncovered lines')
|
|
def step_when_ap_new_insert(context, name):
|
|
profile = _make_automation_profile(name, description="new profile")
|
|
context._ucl_ap_repo.upsert(profile)
|
|
|
|
|
|
@then('the profile "{name}" should be retrievable for uncovered lines')
|
|
def step_then_ap_retrievable(context, name):
|
|
result = context._ucl_ap_repo.get_by_name(name)
|
|
assert result is not None, f"Profile '{name}' was None"
|
|
assert result.name == name
|
|
|
|
|
|
# ── upsert IntegrityError ────────────────────────────────────────────────
|
|
|
|
|
|
@given(
|
|
"an automation profile repository with session raising IntegrityError on flush for uncovered lines"
|
|
)
|
|
def step_given_ap_repo_integrity_error(context):
|
|
def mock_session_factory():
|
|
mock_session = MagicMock(spec=Session)
|
|
query_mock = MagicMock()
|
|
filter_mock = MagicMock()
|
|
filter_mock.first.return_value = None # No existing row
|
|
query_mock.filter_by.return_value = filter_mock
|
|
mock_session.query.return_value = query_mock
|
|
mock_session.add = MagicMock()
|
|
mock_session.flush.side_effect = IntegrityError("UNIQUE constraint", {}, None)
|
|
mock_session.rollback = MagicMock()
|
|
return mock_session
|
|
|
|
context._ucl_ap_repo = AutomationProfileRepository(mock_session_factory)
|
|
|
|
|
|
@when("a profile is upserted and IntegrityError occurs for uncovered lines")
|
|
def step_when_ap_integrity_error(context):
|
|
profile = _make_automation_profile("integrity-fail")
|
|
try:
|
|
context._ucl_ap_repo.upsert(profile)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("a DuplicateAutomationProfileError should be raised for uncovered lines")
|
|
def step_then_ap_dup_error(context):
|
|
assert isinstance(context._ucl_error, DuplicateAutomationProfileError), (
|
|
f"Expected DuplicateAutomationProfileError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── delete not found ─────────────────────────────────────────────────────
|
|
|
|
|
|
@when('a profile "{name}" is deleted for uncovered lines')
|
|
def step_when_ap_delete(context, name):
|
|
try:
|
|
context._ucl_ap_repo.delete(name)
|
|
context._ucl_error = None
|
|
except Exception as exc:
|
|
context._ucl_error = exc
|
|
|
|
|
|
@then("an AutomationProfileNotFoundError should be raised for uncovered lines")
|
|
def step_then_ap_not_found(context):
|
|
assert isinstance(context._ucl_error, AutomationProfileNotFoundError), (
|
|
f"Expected AutomationProfileNotFoundError, got {type(context._ucl_error)}: {context._ucl_error}"
|
|
)
|
|
|
|
|
|
# ── delete success ────────────────────────────────────────────────────────
|
|
|
|
|
|
@when('the profile "{name}" is deleted for uncovered lines')
|
|
def step_when_ap_delete_success(context, name):
|
|
context._ucl_ap_repo.delete(name)
|
|
|
|
|
|
@then('the profile "{name}" should no longer exist for uncovered lines')
|
|
def step_then_ap_deleted(context, name):
|
|
result = context._ucl_ap_repo.get_by_name(name)
|
|
assert result is None, f"Profile '{name}' still exists"
|
|
|
|
|
|
# ── _to_domain full fields ───────────────────────────────────────────────
|
|
|
|
|
|
@given('a profile "{name}" with all fields set has been upserted for uncovered lines')
|
|
def step_given_ap_full_fields(context, name):
|
|
profile = _make_automation_profile(
|
|
name,
|
|
description="full fields profile",
|
|
schema_version="2.0",
|
|
decompose_task=0.1,
|
|
create_tool=0.2,
|
|
select_tool=0.3,
|
|
edit_code=0.4,
|
|
execute_command=0.5,
|
|
create_file=0.6,
|
|
delete_content=0.7,
|
|
access_network=0.8,
|
|
install_dependency=0.9,
|
|
modify_config=0.15,
|
|
approve_plan=0.25,
|
|
require_sandbox=False,
|
|
require_checkpoints=False,
|
|
allow_unsafe_tools=True,
|
|
)
|
|
context._ucl_ap_repo.upsert(profile)
|
|
context._ucl_ap_full = profile
|
|
|
|
|
|
@then("all profile fields should match the original values for uncovered lines")
|
|
def step_then_ap_all_fields(context):
|
|
result = context._ucl_ap_result
|
|
orig = context._ucl_ap_full
|
|
assert result is not None, "Profile was None"
|
|
assert result.name == orig.name
|
|
assert result.description == orig.description
|
|
assert result.schema_version == orig.schema_version
|
|
assert abs(result.decompose_task - orig.decompose_task) < 0.001
|
|
assert abs(result.create_tool - orig.create_tool) < 0.001
|
|
assert abs(result.select_tool - orig.select_tool) < 0.001
|
|
assert abs(result.edit_code - orig.edit_code) < 0.001
|
|
assert abs(result.execute_command - orig.execute_command) < 0.001
|
|
assert abs(result.create_file - orig.create_file) < 0.001
|
|
assert abs(result.delete_content - orig.delete_content) < 0.001
|
|
assert abs(result.access_network - orig.access_network) < 0.001
|
|
assert abs(result.install_dependency - orig.install_dependency) < 0.001
|
|
assert abs(result.modify_config - orig.modify_config) < 0.001
|
|
assert abs(result.approve_plan - orig.approve_plan) < 0.001
|
|
assert result.safety.require_sandbox == orig.safety.require_sandbox
|
|
assert result.safety.require_checkpoints == orig.safety.require_checkpoints
|
|
assert result.safety.allow_unsafe_tools == orig.safety.allow_unsafe_tools
|
|
|
|
|
|
# ── _from_domain + _update_row roundtrip ──────────────────────────────────
|
|
|
|
|
|
@given(
|
|
'a profile "{name}" with specific field values has been upserted for uncovered lines'
|
|
)
|
|
def step_given_ap_specific_fields(context, name):
|
|
profile = _make_automation_profile(
|
|
name,
|
|
description="original",
|
|
decompose_task=0.1,
|
|
create_tool=0.2,
|
|
require_sandbox=True,
|
|
allow_unsafe_tools=False,
|
|
)
|
|
context._ucl_ap_repo.upsert(profile)
|
|
|
|
|
|
@when(
|
|
'the profile "{name}" is upserted again with different field values for uncovered lines'
|
|
)
|
|
def step_when_ap_update_fields(context, name):
|
|
profile = _make_automation_profile(
|
|
name,
|
|
description="updated roundtrip",
|
|
decompose_task=0.9,
|
|
create_tool=0.8,
|
|
select_tool=0.7,
|
|
edit_code=0.6,
|
|
execute_command=0.5,
|
|
create_file=0.4,
|
|
delete_content=0.3,
|
|
access_network=0.2,
|
|
install_dependency=0.1,
|
|
modify_config=0.05,
|
|
approve_plan=0.15,
|
|
require_sandbox=False,
|
|
require_checkpoints=False,
|
|
allow_unsafe_tools=True,
|
|
)
|
|
context._ucl_ap_repo.upsert(profile)
|
|
context._ucl_ap_updated_vals = profile
|
|
|
|
|
|
@then('the profile "{name}" should have the new field values for uncovered lines')
|
|
def step_then_ap_new_values(context, name):
|
|
result = context._ucl_ap_repo.get_by_name(name)
|
|
expected = context._ucl_ap_updated_vals
|
|
assert result is not None, f"Profile '{name}' was None"
|
|
assert result.description == expected.description
|
|
assert abs(result.decompose_task - expected.decompose_task) < 0.001
|
|
assert abs(result.create_tool - expected.create_tool) < 0.001
|
|
assert abs(result.select_tool - expected.select_tool) < 0.001
|
|
assert abs(result.edit_code - expected.edit_code) < 0.001
|
|
assert abs(result.execute_command - expected.execute_command) < 0.001
|
|
assert abs(result.create_file - expected.create_file) < 0.001
|
|
assert abs(result.delete_content - expected.delete_content) < 0.001
|
|
assert abs(result.access_network - expected.access_network) < 0.001
|
|
assert abs(result.install_dependency - expected.install_dependency) < 0.001
|
|
assert abs(result.modify_config - expected.modify_config) < 0.001
|
|
assert abs(result.approve_plan - expected.approve_plan) < 0.001
|
|
assert result.safety.require_sandbox == expected.safety.require_sandbox
|
|
assert result.safety.require_checkpoints == expected.safety.require_checkpoints
|
|
assert result.safety.allow_unsafe_tools == expected.safety.allow_unsafe_tools
|