HAL9000/temp
0
0
Fork 0
forked from HAL9000/cleveragents-core
Code Pull Requests Activity
Files
master
temp/features/steps/database_models_coverage_r2_steps.py
T
CoreRasurae a4a6b061a6 fix(db): align v3_plans schema with specification DDL 
Aligned v3_plans table with specification DDL:

1. Added effective_profile_snapshot column (TEXT NOT NULL) for
   storing frozen JSON snapshot of automation profile at plan
   creation time.  Added Pydantic field_validator ensuring the
   value is well-formed JSON.  Validator catches RecursionError
   for deeply nested JSON, consistent with automation_profile
   deserialization hardening.  Validator error message uses
   length-only to avoid potential information disclosure.
   Documented that the default "{}" exists for backward
   compatibility; new plans should explicitly set the snapshot.

2. Made root_plan_id NOT NULL — root plans self-reference their
   own plan_id, child plans reference the root ancestor.  Added
   explicit ondelete="RESTRICT" FK policy for consistency with
   other FKs in the model.  Documented known FK policy drift
   between ORM model (RESTRICT) and migrated databases (retained
   SET NULL) in the migration; data integrity is preserved by
   the NOT NULL constraint regardless.  Moved root_plan_id
   self-reference resolution into a PlanIdentity model_validator
   so the domain model is consistent with the DB NOT NULL
   constraint before and after persistence (previously the
   resolution only happened in from_domain(), creating an
   asymmetry where root_plan_id was None in-memory but non-null
   after round-tripping through the database).

3. Made automation_profile NOT NULL with default "balanced".

4. Documented intentional deviation: phase default is "action"
   (code) vs "strategize" (spec) because the Action phase was
   added as a pre-Strategize setup step.

5. Created Alembic migration with backfill logic for existing
   rows.  Root-ancestor backfill uses level-by-level propagation
   with a parent-readiness guard to correctly resolve plans at
   arbitrary hierarchy depth (3+ levels).  Added safety bound
   (max 100 iterations) with logged error on exhaustion to guard
   against cycles in parent_plan_id.  Merged batch_alter_table
   operations to avoid redundant full-table copies in SQLite
   batch mode.  Migration backfill also handles empty-string
   automation_profile values.  Documented downgrade limitation
   (backfill is not reversible).  Orphan-row fallback now logs
   affected row count at WARNING level.  Migration cycle-detection
   now logs affected plan_id values before the orphan fallback
   overwrites them.  All migration SQL uses sa.text() for
   consistency with SQLAlchemy best practices.

6. Hardened automation_profile deserialization in to_domain() to
   catch ValueError (invalid StrEnum provenance), Pydantic
   ValidationError, and RecursionError (deeply nested JSON) in
   addition to JSONDecodeError and KeyError, preventing
   unreadable plans from corrupted DB rows.  Applied the same
   defensive deserialization pattern to effective_profile_snapshot
   in to_domain(): corrupted JSON falls back to '{}' with a
   WARNING log instead of crashing the read path.  Added TypeError
   to the effective_profile_snapshot exception list in to_domain()
   for consistency with the Pydantic validator.  Logging of
   unparseable values uses length only to avoid potential
   information disclosure.

7. Used explicit None check (is not None) instead of truthiness
   for root_plan_id resolution in from_domain(), for
   effective_profile_snapshot in to_domain(), and in
   _serialize_automation_profile() for consistency.

8. Documented intentional column naming conventions vs spec DDL
   (e.g. automation_profile vs automation_profile_name, *_actor
   vs *_actor_name, processing_state vs state, v3_plans vs
   plans).  Documented the semantic difference: automation_profile
   stores either a bare name or structured JSON with provenance,
   whereas the spec automation_profile_name stores a plain name.

9. Fixed benchmark plan constructors
   (plan_phase_migration_bench.py) that were missing the now-
   required root_plan_id and effective_profile_snapshot fields.

10. Replaced defensive getattr() with direct attribute access for
    effective_profile_snapshot in from_domain() and update(),
    since the field is now defined on the Plan domain model.

11. Fixed Any type annotation in test helper _make_plan() to use
    AutomationProfileRef | None for proper type safety.

12. Added BDD scenarios for PlanIdentity self-reference
    resolution, NULL effective_profile_snapshot constraint
    enforcement, valid-JSON-missing-profile_name-key
    deserialization, invalid-JSON and empty-string snapshot
    rejection by Pydantic validator, and corrupted
    effective_profile_snapshot DB fallback in to_domain().

13. Extracted default automation profile name to a module-level
    constant (DEFAULT_AUTOMATION_PROFILE) to reduce sentinel
    duplication across models.py and repositories.py.

14. Centralised automation-profile serialisation into
    LifecyclePlanModel._serialize_automation_profile() to
    eliminate duplication between from_domain() and
    LifecyclePlanRepository.update().

15. Fixed to_domain() root_plan_id type cast from str | None
    to str, reflecting the NOT NULL column constraint.

16. Added PlanIdentity model_validator that resolves None
    root_plan_id to plan_id at domain construction time, ensuring
    the domain model honours the spec DDL NOT NULL constraint
    regardless of persistence state.  Simplified from_domain()
    root resolution accordingly.

ISSUES CLOSED: #921
2026-03-30 23:40:36 +01:00

615 lines
20 KiB
Python
Raw Permalink Blame History

"""Step definitions for database_models_coverage_r2.feature.
Targets uncovered lines in
src/cleveragents/infrastructure/database/models.py
identified from build/coverage.xml (round 2).
"""
from __future__ import annotations
import json
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any
from behave import given, then, when # type: ignore[import-untyped]
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
NOW = datetime(2025, 6, 1, 12, 0, 0, tzinfo=UTC)
NOW_ISO = NOW.isoformat()
def _make_ulid(seed: int = 1) -> str:
"""Return a fake but valid-length 26-char ULID."""
return f"01ABCDEFGH{seed:016d}"
def _minimal_plan_timestamps() -> SimpleNamespace:
return SimpleNamespace(
created_at=NOW,
updated_at=NOW,
strategize_started_at=None,
strategize_completed_at=None,
execute_started_at=None,
execute_completed_at=None,
apply_started_at=None,
applied_at=None,
)
def _minimal_plan_identity(plan_id: str | None = None) -> SimpleNamespace:
return SimpleNamespace(
plan_id=plan_id or _make_ulid(1),
parent_plan_id=None,
root_plan_id=None,
attempt=1,
)
def _minimal_namespaced_name() -> SimpleNamespace:
return SimpleNamespace(
server=None,
namespace="local",
name="test-action",
__str__=lambda self: f"{self.namespace}/{self.name}",
)
# ---------------------------------------------------------------------------
# LifecycleActionModel — safety_profile round-trip (lines 376, 427)
# ---------------------------------------------------------------------------
@given("a LifecycleActionModel with safety_profile_json populated")
def step_action_model_with_safety(context: Any) -> None:
from cleveragents.infrastructure.database.models import (
LifecycleActionModel,
)
safety_data = {
"require_sandbox": True,
"require_checkpoints": False,
"allow_unsafe_tools": False,
}
model = LifecycleActionModel(
namespaced_name="local/test-action",
namespace="local",
name="test-action",
description="desc",
definition_of_done="dod",
strategy_actor="strat",
execution_actor="exec",
reusable=True,
read_only=False,
state="available",
tags_json="[]",
created_at=NOW_ISO,
updated_at=NOW_ISO,
safety_profile_json=json.dumps(safety_data),
)
# Ensure relationship lists exist
model.arguments_rel = []
model.invariants_rel = []
context.action_model = model
@when("I convert the action model to domain via to_domain")
def step_action_to_domain(context: Any) -> None:
context.domain_action = context.action_model.to_domain()
@then("the domain action has a non-null safety_profile object")
def step_check_safety_profile_not_none(context: Any) -> None:
sp = context.domain_action.safety_profile
assert sp is not None, "safety_profile should be deserialized"
assert sp.require_sandbox is True
@given("an Action domain object with a SafetyProfile attached")
def step_action_domain_with_safety(context: Any) -> None:
from cleveragents.domain.models.core.safety_profile import SafetyProfile
context.domain_action_input = SimpleNamespace(
namespaced_name=_minimal_namespaced_name(),
description="desc",
long_description=None,
definition_of_done="dod",
strategy_actor="strat",
execution_actor="exec",
review_actor=None,
apply_actor=None,
estimation_actor=None,
invariant_actor=None,
automation_profile=None,
safety_profile=SafetyProfile(
require_sandbox=True,
require_checkpoints=False,
allow_unsafe_tools=True,
),
reusable=True,
read_only=False,
inputs_schema=None,
state=SimpleNamespace(value="available"),
created_by=None,
tags=[],
arguments=[],
invariants=[],
created_at=NOW,
updated_at=NOW,
)
@when("I convert the action to a model via from_domain")
def step_action_from_domain(context: Any) -> None:
from cleveragents.infrastructure.database.models import (
LifecycleActionModel,
)
context.action_model_out = LifecycleActionModel.from_domain(
context.domain_action_input,
)
@then("the model safety_profile_json contains serialized SafetyProfile data")
def step_check_safety_json(context: Any) -> None:
raw = context.action_model_out.safety_profile_json
assert raw is not None, "safety_profile_json must not be None"
parsed = json.loads(raw)
assert parsed["require_sandbox"] is True
assert parsed["allow_unsafe_tools"] is True
# ---------------------------------------------------------------------------
# LifecycleActionModel — string arg_type/requirement (lines 466, 469)
# ---------------------------------------------------------------------------
@given(
"an Action domain object whose arguments have plain-string arg_type and requirement"
)
def step_action_plain_string_args(context: Any) -> None:
arg = SimpleNamespace(
name="my_arg",
arg_type="string", # plain string, no .value
requirement="required", # plain string, no .value
description="a test arg",
default_value=None,
min_value=None,
max_value=None,
validation_pattern=None,
)
context.domain_action_input = SimpleNamespace(
namespaced_name=_minimal_namespaced_name(),
description="desc",
long_description=None,
definition_of_done="dod",
strategy_actor="strat",
execution_actor="exec",
review_actor=None,
apply_actor=None,
estimation_actor=None,
invariant_actor=None,
automation_profile=None,
safety_profile=None,
reusable=True,
read_only=False,
inputs_schema=None,
state=SimpleNamespace(value="available"),
created_by=None,
tags=[],
arguments=[arg],
invariants=[],
created_at=NOW,
updated_at=NOW,
)
@when("I convert that action to a model via from_domain")
def step_that_action_from_domain(context: Any) -> None:
from cleveragents.infrastructure.database.models import (
LifecycleActionModel,
)
context.action_model_out = LifecycleActionModel.from_domain(
context.domain_action_input,
)
@then("the model argument rows use the plain-string values directly")
def step_check_arg_strings(context: Any) -> None:
args = context.action_model_out.arguments_rel
assert len(args) == 1
assert args[0].arg_type == "string"
assert args[0].requirement == "required"
# ---------------------------------------------------------------------------
# LifecyclePlanModel — string processing_state (line 882)
# ---------------------------------------------------------------------------
def _base_plan_ns(
*,
processing_state: Any = "queued",
automation_profile: AutomationProfileRef | None = None,
namespaced_name: Any = None,
error_details: Any = None,
) -> SimpleNamespace:
"""Build a minimal Plan-like namespace for from_domain tests."""
ns_name = namespaced_name or _minimal_namespaced_name()
return SimpleNamespace(
identity=_minimal_plan_identity(),
namespaced_name=ns_name,
action_name="local/test-action",
description="plan desc",
definition_of_done="dod",
phase=SimpleNamespace(value="action"),
processing_state=processing_state,
automation_profile=automation_profile,
strategy_actor=None,
execution_actor=None,
review_actor=None,
apply_actor=None,
estimation_actor=None,
invariant_actor=None,
execution_environment=None,
execution_env_priority=None,
project_links=[],
invariants=[],
arguments={},
arguments_order=[],
changeset_id=None,
sandbox_refs=[],
validation_summary=None,
decision_root_id=None,
timestamps=_minimal_plan_timestamps(),
error_message=None,
error_details=error_details,
created_by=None,
tags=[],
reusable=True,
read_only=False,
effective_profile_snapshot="{}",
)
@given("a Plan domain object whose processing_state is a plain string")
def step_plan_string_state(context: Any) -> None:
context.plan_domain = _base_plan_ns(processing_state="complete")
@when("I convert the plan to a model via from_domain")
def step_plan_from_domain(context: Any) -> None:
from cleveragents.infrastructure.database.models import (
LifecyclePlanModel,
)
context.plan_model_out = LifecyclePlanModel.from_domain(context.plan_domain)
@then("the model processing_state equals the plain string value")
def step_check_plan_state(context: Any) -> None:
assert context.plan_model_out.processing_state == "complete"
# ---------------------------------------------------------------------------
# LifecyclePlanModel — automation_profile (line 890)
# ---------------------------------------------------------------------------
@given("a Plan domain object with a non-null automation_profile")
def step_plan_with_automation_profile(context: Any) -> None:
profile = AutomationProfileRef(
profile_name="strict",
provenance=AutomationProfileProvenance.ACTION,
)
context.plan_domain = _base_plan_ns(automation_profile=profile)
@then("the model automation_profile column contains JSON with profile_name")
def step_check_automation_profile_json(context: Any) -> None:
raw = context.plan_model_out.automation_profile
assert raw is not None
parsed = json.loads(raw)
assert parsed["profile_name"] == "strict"
# ---------------------------------------------------------------------------
# LifecyclePlanModel — fallback namespace (line 906)
# ---------------------------------------------------------------------------
@given("a Plan domain object whose namespaced_name has no namespace attribute")
def step_plan_no_namespace(context: Any) -> None:
# Use a plain string that has no .namespace attribute
context.plan_domain = _base_plan_ns(namespaced_name="local/test-plan")
@then('the model namespace equals "local"')
def step_check_namespace_local(context: Any) -> None:
assert context.plan_model_out.namespace == "local"
# ---------------------------------------------------------------------------
# LifecyclePlanModel — error_details (line 923)
# ---------------------------------------------------------------------------
@given("a Plan domain object with non-null error_details")
def step_plan_with_error_details(context: Any) -> None:
context.plan_domain = _base_plan_ns(
error_details={"code": "E001", "msg": "something failed"},
)
@then("the model error_details_json is a JSON string of the details dict")
def step_check_error_details_json(context: Any) -> None:
raw = context.plan_model_out.error_details_json
assert raw is not None
parsed = json.loads(raw)
assert parsed["code"] == "E001"
assert parsed["msg"] == "something failed"
# ---------------------------------------------------------------------------
# SkillModel — include with overrides (line 2327)
# ---------------------------------------------------------------------------
@given("a Skill domain object with an include that has overrides")
def step_skill_with_include_overrides(context: Any) -> None:
include = SimpleNamespace(
name="other-skill/base",
overrides={"timeout": 120},
)
context.skill_domain = SimpleNamespace(
name="local/my-skill",
description="a skill",
tool_refs=[],
includes=[include],
anonymous_tools=[],
mcp_servers=[],
agent_skills=[],
overrides={},
version=None,
)
@when("I convert the skill to a model via from_domain")
def step_skill_from_domain(context: Any) -> None:
from cleveragents.infrastructure.database.models import SkillModel
context.skill_model_out = SkillModel.from_domain(context.skill_domain)
@then("the include item_config contains the serialized overrides")
def step_check_include_overrides(context: Any) -> None:
items = context.skill_model_out.items_rel
include_items = [i for i in items if i.item_type == "include"]
assert len(include_items) == 1
config = json.loads(include_items[0].item_config)
assert config["overrides"]["timeout"] == 120
# ---------------------------------------------------------------------------
# DecisionModel — string decision_type (line 2698)
# ---------------------------------------------------------------------------
@given("a Decision domain object whose decision_type is a plain string")
def step_decision_string_type(context: Any) -> None:
from cleveragents.domain.models.core.decision import (
ContextSnapshot,
)
context.decision_domain = SimpleNamespace(
decision_id=_make_ulid(10),
plan_id=_make_ulid(11),
parent_decision_id=None,
sequence_number=1,
decision_type="strategy_choice", # plain string, no .value
question="Which strategy?",
chosen_option="Option A",
alternatives_considered=["Option B"],
confidence_score=0.9,
context_snapshot=ContextSnapshot(
hot_context_hash="abc",
hot_context_ref="ref1",
relevant_resources=[],
actor_state_ref="actor1",
),
rationale="because",
actor_reasoning=None,
downstream_decision_ids=[],
downstream_plan_ids=[],
artifacts_produced=[],
created_at=NOW,
is_correction=False,
corrects_decision_id=None,
correction_reason=None,
superseded_by=None,
)
@when("I convert the decision to a model via from_domain")
def step_decision_from_domain(context: Any) -> None:
from cleveragents.infrastructure.database.models import DecisionModel
context.decision_model_out = DecisionModel.from_domain(
context.decision_domain,
)
@then("the model decision_type equals the plain string")
def step_check_decision_type_string(context: Any) -> None:
assert context.decision_model_out.decision_type == "strategy_choice"
# ---------------------------------------------------------------------------
# CheckpointModel — to_domain (lines 2823-2852)
# ---------------------------------------------------------------------------
@given("a CheckpointModel with metadata_json containing reason and phase")
def step_checkpoint_model_with_meta(context: Any) -> None:
from cleveragents.infrastructure.database.models import CheckpointModel
context.checkpoint_model = CheckpointModel(
checkpoint_id=_make_ulid(20),
plan_id=_make_ulid(21),
sandbox_ref="abc123commit",
decision_id=None,
checkpoint_type="pre_write",
resource_id=None,
filesystem_path="checkpoints/cp1",
size_bytes=1024,
created_at=NOW_ISO,
metadata_json=json.dumps(
{
"reason": "before write",
"source_tool": "file_write",
"phase": "execute",
}
),
)
@when("I convert the checkpoint model to domain via to_domain")
def step_checkpoint_to_domain(context: Any) -> None:
context.domain_checkpoint = context.checkpoint_model.to_domain()
@then("the domain checkpoint has the correct metadata fields")
def step_check_checkpoint_metadata(context: Any) -> None:
cp = context.domain_checkpoint
assert cp.checkpoint_id == _make_ulid(20)
assert cp.plan_id == _make_ulid(21)
assert cp.sandbox_ref == "abc123commit"
assert cp.checkpoint_type == "pre_write"
assert cp.filesystem_path == "checkpoints/cp1"
assert cp.size_bytes == 1024
assert cp.metadata.reason == "before write"
assert cp.metadata.source_tool == "file_write"
assert cp.metadata.phase == "execute"
@given("a CheckpointModel with invalid JSON in metadata_json")
def step_checkpoint_bad_json(context: Any) -> None:
from cleveragents.infrastructure.database.models import CheckpointModel
context.checkpoint_model = CheckpointModel(
checkpoint_id=_make_ulid(30),
plan_id=_make_ulid(31),
sandbox_ref="deadbeef",
checkpoint_type="manual",
filesystem_path="",
size_bytes=None,
created_at=NOW_ISO,
metadata_json="NOT VALID JSON {{{",
)
@then("the domain checkpoint metadata is empty defaults")
def step_check_empty_metadata(context: Any) -> None:
cp = context.domain_checkpoint
assert cp.metadata.reason == ""
assert cp.metadata.source_tool == ""
assert cp.metadata.phase == ""
# ---------------------------------------------------------------------------
# CheckpointModel — from_domain (lines 2865-2879)
# ---------------------------------------------------------------------------
@given("a Checkpoint domain object with full metadata")
def step_checkpoint_domain(context: Any) -> None:
from cleveragents.domain.models.core.checkpoint import (
Checkpoint,
CheckpointMetadata,
)
context.checkpoint_domain = Checkpoint(
checkpoint_id=_make_ulid(40),
plan_id=_make_ulid(41),
sandbox_ref="commitsha",
decision_id=_make_ulid(42),
checkpoint_type="post_step",
resource_id=_make_ulid(43),
filesystem_path="cp/path",
size_bytes=2048,
created_at=NOW,
metadata=CheckpointMetadata(
reason="post step save",
source_tool="apply_tool",
phase="apply",
),
)
@when("I convert the checkpoint to a model via from_domain")
def step_checkpoint_from_domain(context: Any) -> None:
from cleveragents.infrastructure.database.models import CheckpointModel
context.checkpoint_model_out = CheckpointModel.from_domain(
context.checkpoint_domain,
)
@then("the model has correct checkpoint_id plan_id and metadata_json")
def step_check_checkpoint_model_fields(context: Any) -> None:
m = context.checkpoint_model_out
assert m.checkpoint_id == _make_ulid(40)
assert m.plan_id == _make_ulid(41)
assert m.sandbox_ref == "commitsha"
assert m.decision_id == _make_ulid(42)
assert m.checkpoint_type == "post_step"
assert m.resource_id == _make_ulid(43)
assert m.filesystem_path == "cp/path"
assert m.size_bytes == 2048
assert m.created_at == NOW_ISO
raw = m.metadata_json
assert raw is not None
parsed = json.loads(raw)
assert parsed["reason"] == "post step save"
assert parsed["source_tool"] == "apply_tool"
assert parsed["phase"] == "apply"
# ---------------------------------------------------------------------------
# get_session (lines 2907-2908)
# ---------------------------------------------------------------------------
@given("an in-memory SQLAlchemy engine")
def step_create_engine(context: Any) -> None:
from cleveragents.infrastructure.database.models import init_database
context.engine = init_database("sqlite:///:memory:")
@when("I call get_session with that engine")
def step_call_get_session(context: Any) -> None:
from cleveragents.infrastructure.database.models import get_session
context.session = get_session(context.engine)
@then("I receive a valid SQLAlchemy session object")
def step_check_session(context: Any) -> None:
from sqlalchemy.orm import Session
assert isinstance(context.session, Session)
# Verify it can execute a simple query
result = context.session.execute(__import__("sqlalchemy").text("SELECT 1"))
assert result.scalar() == 1
context.session.close()
View Git Blame Copy Permalink