"""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, root_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, effective_profile_snapshot="{}", 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, root_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"]), effective_profile_snapshot="{}", 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, root_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), effective_profile_snapshot="{}", 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, root_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), effective_profile_snapshot="{}", 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, root_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", effective_profile_snapshot="{}", 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