forked from HAL9000/cleveragents-core
942 lines
33 KiB
Python
942 lines
33 KiB
Python
"""Step definitions for database model coverage boost tests.
|
|
|
|
Targets uncovered branches in LifecycleActionModel, LifecyclePlanModel,
|
|
and NamespacedProjectModel to_domain/from_domain methods.
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from behave import given, then, when
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from ulid import ULID
|
|
|
|
from cleveragents.infrastructure.database.models import (
|
|
ActionArgumentModel,
|
|
Base,
|
|
LifecycleActionModel,
|
|
LifecyclePlanModel,
|
|
NamespacedProjectModel,
|
|
PlanArgumentModel,
|
|
ProjectResourceLinkModel,
|
|
ResourceModel,
|
|
ResourceTypeModel,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Valid ULIDs for tests (generated via python-ulid)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
ULID_PLAN_1 = str(ULID())
|
|
ULID_PLAN_2 = str(ULID())
|
|
ULID_RESOURCE_1 = str(ULID())
|
|
ULID_RESOURCE_2 = str(ULID())
|
|
ULID_LINK_1 = str(ULID())
|
|
ULID_LINK_2 = str(ULID())
|
|
|
|
NOW_ISO = "2025-06-01T12:00:00"
|
|
LATER_ISO = "2025-06-01T13:00:00"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the coverage boost database is ready")
|
|
def step_coverage_boost_db_ready(context):
|
|
"""Set up an in-memory database with all lifecycle tables."""
|
|
context.cb_engine = create_engine("sqlite:///:memory:")
|
|
context.CbSessionLocal = sessionmaker(
|
|
bind=context.cb_engine, autoflush=False, autocommit=False
|
|
)
|
|
Base.metadata.create_all(context.cb_engine)
|
|
|
|
|
|
@given("a coverage boost database session is open")
|
|
def step_coverage_boost_session_open(context):
|
|
"""Open a database session for coverage boost tests."""
|
|
context.cb_session = context.CbSessionLocal()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _ensure_action_exists(session, name="local/cov-action"):
|
|
"""Ensure the FK-target action row exists."""
|
|
existing = (
|
|
session.query(LifecycleActionModel).filter_by(namespaced_name=name).first()
|
|
)
|
|
if existing:
|
|
return
|
|
session.add(
|
|
LifecycleActionModel(
|
|
namespaced_name=name,
|
|
namespace="local",
|
|
name="cov-action",
|
|
description="Coverage action",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
state="available",
|
|
reusable=True,
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
|
|
def _ensure_resource_type_exists(session, type_name="builtin/git-checkout"):
|
|
"""Ensure the FK-target resource_type row exists."""
|
|
existing = session.query(ResourceTypeModel).filter_by(name=type_name).first()
|
|
if existing:
|
|
return
|
|
session.add(
|
|
ResourceTypeModel(
|
|
name=type_name,
|
|
namespace="builtin",
|
|
resource_kind="physical",
|
|
user_addable=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
|
|
def _ensure_resource_exists(session, resource_id, type_name="builtin/git-checkout"):
|
|
"""Ensure the FK-target resource row exists."""
|
|
_ensure_resource_type_exists(session, type_name)
|
|
existing = session.query(ResourceModel).filter_by(resource_id=resource_id).first()
|
|
if existing:
|
|
return
|
|
session.add(
|
|
ResourceModel(
|
|
resource_id=resource_id,
|
|
type_name=type_name,
|
|
resource_kind="physical",
|
|
read_only=False,
|
|
auto_discovered=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
|
|
# ===================================================================
|
|
# LifecycleActionModel.to_domain: default_value_json branch
|
|
# ===================================================================
|
|
|
|
|
|
@given("an action model exists with arguments that have default values set")
|
|
def step_action_model_with_default_values(context):
|
|
"""Create an action model with arguments having default_value_json set."""
|
|
model = LifecycleActionModel(
|
|
namespaced_name="local/default-val-action",
|
|
namespace="local",
|
|
name="default-val-action",
|
|
description="Action with arg defaults",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
state="available",
|
|
reusable=True,
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
# Argument WITH a default value
|
|
model.arguments_rel.append(
|
|
ActionArgumentModel(
|
|
name="threshold",
|
|
arg_type="integer",
|
|
requirement="optional",
|
|
description="Coverage threshold",
|
|
default_value_json=json.dumps(80),
|
|
position=0,
|
|
)
|
|
)
|
|
# Argument WITHOUT a default value
|
|
model.arguments_rel.append(
|
|
ActionArgumentModel(
|
|
name="target",
|
|
arg_type="string",
|
|
requirement="required",
|
|
description="Target module",
|
|
default_value_json=None,
|
|
position=1,
|
|
)
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_action_model = model
|
|
|
|
|
|
@when("the action model is converted to a domain object")
|
|
def step_action_model_to_domain(context):
|
|
"""Call to_domain on the action model."""
|
|
context.cb_action_domain = context.cb_action_model.to_domain()
|
|
|
|
|
|
@then("the domain action argument should have the correct default value")
|
|
def step_verify_arg_default_value(context):
|
|
"""Verify the argument with default_value_json has its value deserialized."""
|
|
args = context.cb_action_domain.arguments
|
|
threshold_arg = next(a for a in args if a.name == "threshold")
|
|
assert threshold_arg.default_value == 80
|
|
|
|
|
|
@then("the domain action argument without a default should have None")
|
|
def step_verify_arg_no_default(context):
|
|
"""Verify the argument without default_value_json has None."""
|
|
args = context.cb_action_domain.arguments
|
|
target_arg = next(a for a in args if a.name == "target")
|
|
assert target_arg.default_value is None
|
|
|
|
|
|
# ===================================================================
|
|
# LifecycleActionModel.to_domain: inputs_schema_json branch
|
|
# ===================================================================
|
|
|
|
|
|
@given("an action model exists with inputs_schema_json populated")
|
|
def step_action_model_with_inputs_schema(context):
|
|
"""Create an action model with a non-null inputs_schema_json."""
|
|
schema = {"type": "object", "properties": {"name": {"type": "string"}}}
|
|
model = LifecycleActionModel(
|
|
namespaced_name="local/schema-action",
|
|
namespace="local",
|
|
name="schema-action",
|
|
description="Action with inputs schema",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
state="available",
|
|
reusable=True,
|
|
read_only=False,
|
|
inputs_schema_json=json.dumps(schema),
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_action_model = model
|
|
|
|
|
|
@then("the domain action should have the parsed inputs_schema dict")
|
|
def step_verify_inputs_schema_parsed(context):
|
|
"""Verify inputs_schema is a dict parsed from JSON."""
|
|
schema = context.cb_action_domain.inputs_schema
|
|
assert isinstance(schema, dict)
|
|
assert schema["type"] == "object"
|
|
assert "properties" in schema
|
|
|
|
|
|
# ===================================================================
|
|
# LifecycleActionModel.from_domain: inputs_schema not None
|
|
# ===================================================================
|
|
|
|
|
|
@given("an action domain object with inputs_schema set")
|
|
def step_action_domain_with_inputs_schema(context):
|
|
"""Prepare an Action domain object that has inputs_schema."""
|
|
from cleveragents.domain.models.core.action import Action, ActionState
|
|
from cleveragents.domain.models.core.plan import NamespacedName
|
|
|
|
context.cb_action_domain_input = Action(
|
|
namespaced_name=NamespacedName(namespace="local", name="schema-from"),
|
|
description="Action with schema",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
inputs_schema={"type": "object", "required": ["name"]},
|
|
state=ActionState.AVAILABLE,
|
|
created_at=datetime(2025, 6, 1, 12, 0, 0),
|
|
updated_at=datetime(2025, 6, 1, 13, 0, 0),
|
|
)
|
|
|
|
|
|
@when("the action domain object is converted to a database model")
|
|
def step_action_domain_to_model(context):
|
|
"""Call from_domain on the action domain object."""
|
|
context.cb_action_model_result = LifecycleActionModel.from_domain(
|
|
context.cb_action_domain_input
|
|
)
|
|
|
|
|
|
@then("the database model should have inputs_schema_json as a JSON string")
|
|
def step_verify_model_inputs_schema_json(context):
|
|
"""Verify inputs_schema_json is a non-null JSON string."""
|
|
raw = context.cb_action_model_result.inputs_schema_json
|
|
assert raw is not None
|
|
parsed = json.loads(raw)
|
|
assert parsed["type"] == "object"
|
|
assert "required" in parsed
|
|
|
|
|
|
# ===================================================================
|
|
# LifecycleActionModel.from_domain: default_value not None
|
|
# ===================================================================
|
|
|
|
|
|
@given("an action domain object with arguments that have default values")
|
|
def step_action_domain_with_arg_defaults(context):
|
|
"""Prepare an Action with arguments, some having default_value."""
|
|
from cleveragents.domain.models.core.action import (
|
|
Action,
|
|
ActionArgument,
|
|
ActionState,
|
|
ArgumentRequirement,
|
|
ArgumentType,
|
|
)
|
|
from cleveragents.domain.models.core.plan import NamespacedName
|
|
|
|
context.cb_action_domain_input = Action(
|
|
namespaced_name=NamespacedName(namespace="local", name="defaults-from"),
|
|
description="Action with arg defaults",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
arguments=[
|
|
ActionArgument(
|
|
name="threshold",
|
|
arg_type=ArgumentType.INTEGER,
|
|
requirement=ArgumentRequirement.OPTIONAL,
|
|
description="Coverage threshold",
|
|
default_value=80,
|
|
),
|
|
ActionArgument(
|
|
name="target",
|
|
arg_type=ArgumentType.STRING,
|
|
requirement=ArgumentRequirement.REQUIRED,
|
|
description="Target module",
|
|
default_value=None,
|
|
),
|
|
],
|
|
state=ActionState.AVAILABLE,
|
|
created_at=datetime(2025, 6, 1, 12, 0, 0),
|
|
updated_at=datetime(2025, 6, 1, 13, 0, 0),
|
|
)
|
|
|
|
|
|
@then("the database model arguments should have default_value_json set")
|
|
def step_verify_model_arg_default_json(context):
|
|
"""Verify argument with default_value has default_value_json serialized."""
|
|
args = context.cb_action_model_result.arguments_rel
|
|
threshold_arg = next(a for a in args if a.name == "threshold")
|
|
assert threshold_arg.default_value_json is not None
|
|
assert json.loads(threshold_arg.default_value_json) == 80
|
|
|
|
|
|
@then("the argument without a default should have null default_value_json")
|
|
def step_verify_model_arg_no_default_json(context):
|
|
"""Verify argument without default_value has null default_value_json."""
|
|
args = context.cb_action_model_result.arguments_rel
|
|
target_arg = next(a for a in args if a.name == "target")
|
|
assert target_arg.default_value_json is None
|
|
|
|
|
|
# ===================================================================
|
|
# LifecycleActionModel.from_domain: invariants loop body
|
|
# ===================================================================
|
|
|
|
|
|
@given("an action domain object with invariants defined")
|
|
def step_action_domain_with_invariants(context):
|
|
"""Prepare an Action with non-empty invariants list."""
|
|
from cleveragents.domain.models.core.action import Action, ActionState
|
|
from cleveragents.domain.models.core.plan import NamespacedName
|
|
|
|
context.cb_action_domain_input = Action(
|
|
namespaced_name=NamespacedName(namespace="local", name="inv-action"),
|
|
description="Action with invariants",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
invariants=["No breaking changes", "Maintain backward compatibility"],
|
|
state=ActionState.AVAILABLE,
|
|
created_at=datetime(2025, 6, 1, 12, 0, 0),
|
|
updated_at=datetime(2025, 6, 1, 13, 0, 0),
|
|
)
|
|
|
|
|
|
@then("the database model should have invariant child records")
|
|
def step_verify_model_has_invariants(context):
|
|
"""Verify the model has invariant child records."""
|
|
invs = context.cb_action_model_result.invariants_rel
|
|
assert len(invs) == 2
|
|
|
|
|
|
@then("each invariant child record should have correct text and position")
|
|
def step_verify_invariant_child_records(context):
|
|
"""Verify invariant child records have correct text and position."""
|
|
invs = context.cb_action_model_result.invariants_rel
|
|
assert invs[0].invariant_text == "No breaking changes"
|
|
assert invs[0].position == 0
|
|
assert invs[1].invariant_text == "Maintain backward compatibility"
|
|
assert invs[1].position == 1
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.to_domain: automation_profile not None
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan model exists with automation_profile JSON set")
|
|
def step_plan_model_with_automation_profile(context):
|
|
"""Create a plan model with automation_profile as JSON string."""
|
|
_ensure_action_exists(context.cb_session)
|
|
profile_json = json.dumps(
|
|
{
|
|
"profile_name": "trusted",
|
|
"provenance": "action",
|
|
}
|
|
)
|
|
model = LifecyclePlanModel(
|
|
plan_id=ULID_PLAN_1,
|
|
action_name="local/cov-action",
|
|
namespaced_name="local/profile-plan",
|
|
namespace="local",
|
|
phase="strategize",
|
|
processing_state="queued",
|
|
attempt=1,
|
|
description="Plan with automation profile",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
automation_profile=profile_json,
|
|
reusable=True,
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_plan_model = model
|
|
|
|
|
|
@when("the plan model is converted to a domain object")
|
|
def step_plan_model_to_domain(context):
|
|
"""Call to_domain on the plan model."""
|
|
context.cb_plan_domain = context.cb_plan_model.to_domain()
|
|
|
|
|
|
@then("the domain plan should have an automation profile with the correct name")
|
|
def step_verify_plan_automation_profile_name(context):
|
|
"""Verify automation_profile is parsed with correct profile_name."""
|
|
ap = context.cb_plan_domain.automation_profile
|
|
assert ap is not None
|
|
assert ap.profile_name == "trusted"
|
|
|
|
|
|
@then("the domain plan automation profile should have the correct provenance")
|
|
def step_verify_plan_automation_profile_provenance(context):
|
|
"""Verify automation_profile has correct provenance."""
|
|
from cleveragents.domain.models.core.plan import AutomationProfileProvenance
|
|
|
|
ap = context.cb_plan_domain.automation_profile
|
|
assert ap.provenance == AutomationProfileProvenance.ACTION
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.to_domain: sandbox_refs_json
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan model exists with sandbox_refs_json populated")
|
|
def step_plan_model_with_sandbox_refs(context):
|
|
"""Create a plan model with sandbox_refs_json as a JSON list."""
|
|
_ensure_action_exists(context.cb_session)
|
|
model = LifecyclePlanModel(
|
|
plan_id=ULID_PLAN_2,
|
|
action_name="local/cov-action",
|
|
namespaced_name="local/sandbox-plan",
|
|
namespace="local",
|
|
phase="execute",
|
|
processing_state="processing",
|
|
attempt=1,
|
|
description="Plan with sandbox refs",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
sandbox_refs_json=json.dumps(["sandbox-ref-1", "sandbox-ref-2"]),
|
|
reusable=True,
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_plan_model = model
|
|
|
|
|
|
@then("the domain plan should have the parsed sandbox refs list")
|
|
def step_verify_plan_sandbox_refs(context):
|
|
"""Verify sandbox_refs is a list parsed from JSON."""
|
|
refs = context.cb_plan_domain.sandbox_refs
|
|
assert refs == ["sandbox-ref-1", "sandbox-ref-2"]
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.to_domain: validation_summary_json
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan model exists with validation_summary_json populated")
|
|
def step_plan_model_with_validation_summary(context):
|
|
"""Create a plan model with validation_summary_json set."""
|
|
_ensure_action_exists(context.cb_session)
|
|
summary = {"tests_passed": 42, "tests_failed": 0, "coverage": 95.5}
|
|
# Use a fresh ULID to avoid PK conflict
|
|
plan_id = str(ULID())
|
|
model = LifecyclePlanModel(
|
|
plan_id=plan_id,
|
|
action_name="local/cov-action",
|
|
namespaced_name="local/validation-plan",
|
|
namespace="local",
|
|
phase="apply",
|
|
processing_state="queued",
|
|
attempt=1,
|
|
description="Plan with validation summary",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
validation_summary_json=json.dumps(summary),
|
|
reusable=True,
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_plan_model = model
|
|
|
|
|
|
@then("the domain plan should have the parsed validation summary dict")
|
|
def step_verify_plan_validation_summary(context):
|
|
"""Verify validation_summary is a dict parsed from JSON."""
|
|
vs = context.cb_plan_domain.validation_summary
|
|
assert isinstance(vs, dict)
|
|
assert vs["tests_passed"] == 42
|
|
assert vs["coverage"] == 95.5
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.to_domain: error_details_json
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan model exists with error_details_json populated")
|
|
def step_plan_model_with_error_details(context):
|
|
"""Create a plan model with error_details_json set."""
|
|
_ensure_action_exists(context.cb_session)
|
|
error_details = {"error_type": "ValidationError", "trace": "line 42"}
|
|
plan_id = str(ULID())
|
|
model = LifecyclePlanModel(
|
|
plan_id=plan_id,
|
|
action_name="local/cov-action",
|
|
namespaced_name="local/error-plan",
|
|
namespace="local",
|
|
phase="apply",
|
|
processing_state="errored",
|
|
attempt=1,
|
|
description="Plan with error details",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
error_message="Something went wrong",
|
|
error_details_json=json.dumps(error_details),
|
|
reusable=True,
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_plan_model = model
|
|
|
|
|
|
@then("the domain plan should have the parsed error details dict")
|
|
def step_verify_plan_error_details(context):
|
|
"""Verify error_details is a dict parsed from JSON."""
|
|
ed = context.cb_plan_domain.error_details
|
|
assert isinstance(ed, dict)
|
|
assert ed["error_type"] == "ValidationError"
|
|
assert ed["trace"] == "line 42"
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.to_domain: arguments from child table
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan model exists with argument child records")
|
|
def step_plan_model_with_arguments(context):
|
|
"""Create a plan model with PlanArgumentModel child records."""
|
|
_ensure_action_exists(context.cb_session)
|
|
plan_id = str(ULID())
|
|
model = LifecyclePlanModel(
|
|
plan_id=plan_id,
|
|
action_name="local/cov-action",
|
|
namespaced_name="local/args-plan",
|
|
namespace="local",
|
|
phase="strategize",
|
|
processing_state="queued",
|
|
attempt=1,
|
|
description="Plan with arguments",
|
|
definition_of_done="Tests pass",
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
reusable=True,
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
model.arguments_rel.append(
|
|
PlanArgumentModel(
|
|
name="coverage",
|
|
value_json=json.dumps(80),
|
|
value_type="integer",
|
|
position=0,
|
|
)
|
|
)
|
|
model.arguments_rel.append(
|
|
PlanArgumentModel(
|
|
name="framework",
|
|
value_json=json.dumps("pytest"),
|
|
value_type="string",
|
|
position=1,
|
|
)
|
|
)
|
|
model.arguments_rel.append(
|
|
PlanArgumentModel(
|
|
name="verbose",
|
|
value_json=None,
|
|
value_type="boolean",
|
|
position=2,
|
|
)
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_plan_model = model
|
|
|
|
|
|
@then("the domain plan should have the arguments dict populated")
|
|
def step_verify_plan_arguments_dict(context):
|
|
"""Verify the arguments dict has entries from child records."""
|
|
args = context.cb_plan_domain.arguments
|
|
assert args["coverage"] == 80
|
|
assert args["framework"] == "pytest"
|
|
assert args["verbose"] is None
|
|
|
|
|
|
@then("the domain plan should have arguments_order populated")
|
|
def step_verify_plan_arguments_order(context):
|
|
"""Verify arguments_order contains the argument names in order."""
|
|
order = context.cb_plan_domain.arguments_order
|
|
assert order == ["coverage", "framework", "verbose"]
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.from_domain: project_links loop
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan domain object with project links defined")
|
|
def step_plan_domain_with_project_links(context):
|
|
"""Prepare a Plan domain object with project_links populated."""
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
ProjectLink,
|
|
)
|
|
|
|
context.cb_plan_domain_input = Plan(
|
|
identity=PlanIdentity(plan_id=str(ULID())),
|
|
namespaced_name=NamespacedName(namespace="local", name="pl-plan"),
|
|
action_name="local/cov-action",
|
|
description="Plan with project links",
|
|
definition_of_done="Tests pass",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
project_links=[
|
|
ProjectLink(project_name="local/api-service", alias="api", read_only=False),
|
|
ProjectLink(project_name="local/web-app", alias="web", read_only=True),
|
|
],
|
|
timestamps=PlanTimestamps(
|
|
created_at=datetime(2025, 6, 1, 12, 0, 0),
|
|
updated_at=datetime(2025, 6, 1, 13, 0, 0),
|
|
),
|
|
tags=["test"],
|
|
)
|
|
|
|
|
|
@when("the plan domain object is converted to a database model via from_domain")
|
|
def step_plan_domain_to_model(context):
|
|
"""Call from_domain on the plan domain object."""
|
|
context.cb_plan_model_result = LifecyclePlanModel.from_domain(
|
|
context.cb_plan_domain_input
|
|
)
|
|
|
|
|
|
@then("the database plan model should have project link child records")
|
|
def step_verify_plan_model_project_links(context):
|
|
"""Verify the model has project link child records."""
|
|
links = context.cb_plan_model_result.project_links_rel
|
|
assert len(links) == 2
|
|
|
|
|
|
@then("each project link child record should have the correct project name")
|
|
def step_verify_project_link_names(context):
|
|
"""Verify each project link child record has the correct project name."""
|
|
links = context.cb_plan_model_result.project_links_rel
|
|
names = [pl.project_name for pl in links]
|
|
assert "local/api-service" in names
|
|
assert "local/web-app" in names
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.from_domain: invariants loop
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan domain object with plan invariants defined")
|
|
def step_plan_domain_with_invariants(context):
|
|
"""Prepare a Plan domain object with invariants list."""
|
|
from cleveragents.domain.models.core.plan import (
|
|
InvariantSource,
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanInvariant,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
|
|
context.cb_plan_domain_input = Plan(
|
|
identity=PlanIdentity(plan_id=str(ULID())),
|
|
namespaced_name=NamespacedName(namespace="local", name="inv-plan"),
|
|
action_name="local/cov-action",
|
|
description="Plan with invariants",
|
|
definition_of_done="Tests pass",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
invariants=[
|
|
PlanInvariant(text="No breaking changes", source=InvariantSource.ACTION),
|
|
PlanInvariant(text="Keep API stable", source=InvariantSource.PROJECT),
|
|
],
|
|
timestamps=PlanTimestamps(
|
|
created_at=datetime(2025, 6, 1, 12, 0, 0),
|
|
updated_at=datetime(2025, 6, 1, 13, 0, 0),
|
|
),
|
|
tags=["test"],
|
|
)
|
|
|
|
|
|
@then("the database plan model should have invariant child records with source")
|
|
def step_verify_plan_model_invariants(context):
|
|
"""Verify the model has invariant child records."""
|
|
invs = context.cb_plan_model_result.invariants_rel
|
|
assert len(invs) == 2
|
|
|
|
|
|
@then("each plan invariant child record should have correct text and source")
|
|
def step_verify_plan_invariant_details(context):
|
|
"""Verify invariant child records have correct text and source_scope."""
|
|
invs = context.cb_plan_model_result.invariants_rel
|
|
assert invs[0].invariant_text == "No breaking changes"
|
|
assert invs[0].source_scope == "action"
|
|
assert invs[0].position == 0
|
|
assert invs[1].invariant_text == "Keep API stable"
|
|
assert invs[1].source_scope == "project"
|
|
assert invs[1].position == 1
|
|
|
|
|
|
# ===================================================================
|
|
# LifecyclePlanModel.from_domain: arguments loop
|
|
# ===================================================================
|
|
|
|
|
|
@given("a plan domain object with arguments and arguments_order defined")
|
|
def step_plan_domain_with_arguments(context):
|
|
"""Prepare a Plan domain object with arguments dict and arguments_order."""
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
|
|
context.cb_plan_domain_input = Plan(
|
|
identity=PlanIdentity(plan_id=str(ULID())),
|
|
namespaced_name=NamespacedName(namespace="local", name="args-plan"),
|
|
action_name="local/cov-action",
|
|
description="Plan with arguments",
|
|
definition_of_done="Tests pass",
|
|
phase=PlanPhase.STRATEGIZE,
|
|
processing_state=ProcessingState.QUEUED,
|
|
strategy_actor="local/strategy",
|
|
execution_actor="local/executor",
|
|
arguments={"coverage": 80, "framework": "pytest"},
|
|
arguments_order=["coverage", "framework"],
|
|
timestamps=PlanTimestamps(
|
|
created_at=datetime(2025, 6, 1, 12, 0, 0),
|
|
updated_at=datetime(2025, 6, 1, 13, 0, 0),
|
|
),
|
|
tags=["test"],
|
|
)
|
|
|
|
|
|
@then("the database plan model should have argument child records")
|
|
def step_verify_plan_model_arguments(context):
|
|
"""Verify the model has argument child records."""
|
|
args = context.cb_plan_model_result.arguments_rel
|
|
assert len(args) == 2
|
|
|
|
|
|
@then("each plan argument child record should have correct name and value")
|
|
def step_verify_plan_argument_details(context):
|
|
"""Verify argument child records have correct name and value_json."""
|
|
args = context.cb_plan_model_result.arguments_rel
|
|
coverage_arg = next(a for a in args if a.name == "coverage")
|
|
framework_arg = next(a for a in args if a.name == "framework")
|
|
assert json.loads(coverage_arg.value_json) == 80
|
|
assert json.loads(framework_arg.value_json) == "pytest"
|
|
assert coverage_arg.position == 0
|
|
assert framework_arg.position == 1
|
|
|
|
|
|
# ===================================================================
|
|
# NamespacedProjectModel.to_domain: resource_links loop
|
|
# ===================================================================
|
|
|
|
|
|
@given("a project model exists with resource link child records")
|
|
def step_project_model_with_resource_links(context):
|
|
"""Create a project model with ProjectResourceLinkModel children."""
|
|
_ensure_resource_exists(context.cb_session, ULID_RESOURCE_1)
|
|
_ensure_resource_exists(context.cb_session, ULID_RESOURCE_2)
|
|
|
|
model = NamespacedProjectModel(
|
|
namespaced_name="local/linked-project",
|
|
namespace="local",
|
|
description="Project with resource links",
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.flush()
|
|
|
|
link1 = ProjectResourceLinkModel(
|
|
link_id=ULID_LINK_1,
|
|
project_name="local/linked-project",
|
|
resource_id=ULID_RESOURCE_1,
|
|
alias="main-repo",
|
|
read_only=False,
|
|
created_at=NOW_ISO,
|
|
)
|
|
link2 = ProjectResourceLinkModel(
|
|
link_id=ULID_LINK_2,
|
|
project_name="local/linked-project",
|
|
resource_id=ULID_RESOURCE_2,
|
|
alias="docs-repo",
|
|
read_only=True,
|
|
created_at=NOW_ISO,
|
|
)
|
|
context.cb_session.add_all([link1, link2])
|
|
context.cb_session.commit()
|
|
|
|
# Re-fetch to ensure relationships are loaded
|
|
context.cb_project_model = (
|
|
context.cb_session.query(NamespacedProjectModel)
|
|
.filter_by(namespaced_name="local/linked-project")
|
|
.first()
|
|
)
|
|
|
|
|
|
@when("the project model is converted to a domain object")
|
|
def step_project_model_to_domain(context):
|
|
"""Call to_domain on the project model."""
|
|
context.cb_project_domain = context.cb_project_model.to_domain()
|
|
|
|
|
|
@then("the domain project should have linked resources populated")
|
|
def step_verify_project_linked_resources(context):
|
|
"""Verify the project has linked_resources populated."""
|
|
lr = context.cb_project_domain.linked_resources
|
|
assert len(lr) == 2
|
|
|
|
|
|
@then("each linked resource should have the correct resource id and alias")
|
|
def step_verify_linked_resource_details(context):
|
|
"""Verify each linked resource has correct resource_id and alias."""
|
|
lr = context.cb_project_domain.linked_resources
|
|
ids = {r.resource_id for r in lr}
|
|
assert ULID_RESOURCE_1 in ids
|
|
assert ULID_RESOURCE_2 in ids
|
|
aliases = {r.alias for r in lr}
|
|
assert "main-repo" in aliases
|
|
assert "docs-repo" in aliases
|
|
|
|
|
|
# ===================================================================
|
|
# NamespacedProjectModel.to_domain: context_policy_json parsing
|
|
# ===================================================================
|
|
|
|
|
|
@given("a project model exists with context_policy_json populated")
|
|
def step_project_model_with_context_policy(context):
|
|
"""Create a project model with non-null context_policy_json."""
|
|
policy = {
|
|
"max_file_size": 500000,
|
|
"indexing_strategy": "semantic",
|
|
"summarize": False,
|
|
}
|
|
model = NamespacedProjectModel(
|
|
namespaced_name="local/ctx-project",
|
|
namespace="local",
|
|
description="Project with context policy",
|
|
context_policy_json=json.dumps(policy),
|
|
created_at=NOW_ISO,
|
|
updated_at=LATER_ISO,
|
|
tags_json="[]",
|
|
)
|
|
context.cb_session.add(model)
|
|
context.cb_session.commit()
|
|
context.cb_project_model = model
|
|
|
|
|
|
@then("the domain project should have a context config with custom values")
|
|
def step_verify_project_context_config(context):
|
|
"""Verify context_config is parsed from JSON with custom values."""
|
|
cc = context.cb_project_domain.context_config
|
|
assert cc.max_file_size == 500000
|
|
assert cc.indexing_strategy == "semantic"
|
|
assert cc.summarize is False
|