forked from HAL9000/cleveragents-core
Luis Mendes
ab911dbdc4
The format_output() function returned a string that callers passed to Rich console.print(), which wraps long lines at terminal width. This injected literal newline characters into JSON string values (e.g. in definition_of_done fields), producing invalid JSON that downstream parsers could not decode (JSONDecodeError: Invalid control character). For machine-readable formats (json, yaml, plain), format_output() now writes the rendered output directly to sys.stdout and returns an empty string. This preserves the exact serialization from json.dumps/ yaml.dump without Rich text processing artifacts. Refs: #746
1188 lines
43 KiB
Python
1188 lines
43 KiB
Python
"""Step definitions for plan_apply_service_coverage feature.
|
|
|
|
Covers all code paths in plan_apply_service.py including helper functions,
|
|
service methods, edge cases, and error branches.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
|
|
from cleveragents.application.services.plan_apply_service import (
|
|
_OP_COLORS,
|
|
ApplyOutcome,
|
|
ApplyResult,
|
|
PlanApplyService,
|
|
_build_artifacts_dict,
|
|
_operation_label,
|
|
_render_diff_json,
|
|
_render_diff_plain,
|
|
_render_diff_rich,
|
|
)
|
|
from cleveragents.core.exceptions import PlanError, ValidationError
|
|
from cleveragents.domain.models.core.change import (
|
|
ChangeEntry,
|
|
ChangeOperation,
|
|
SpecChangeSet,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
|
|
__all__: list[str] = []
|
|
|
|
# Plan ID constant for tests
|
|
_PLAN_ID = "01PLANTEST000000000000001"
|
|
_CHANGESET_ID = "01CSTEST0000000000000001"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Stub / helper factories
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
class _StubPhase:
|
|
"""Stub for PlanPhase with a .value attribute."""
|
|
|
|
def __init__(self, phase: PlanPhase) -> None:
|
|
"""Initialize with the real enum member."""
|
|
self._phase = phase
|
|
self.value = phase.value
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
"""Compare against PlanPhase or another _StubPhase."""
|
|
if isinstance(other, PlanPhase):
|
|
return self._phase == other
|
|
if isinstance(other, _StubPhase):
|
|
return self._phase == other._phase
|
|
return NotImplemented
|
|
|
|
|
|
class _StubState:
|
|
"""Stub for ProcessingState with a .value attribute."""
|
|
|
|
def __init__(self, state: ProcessingState) -> None:
|
|
"""Initialize with the real enum member."""
|
|
self._state = state
|
|
self.value = state.value
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
"""Compare against ProcessingState or another _StubState."""
|
|
if isinstance(other, ProcessingState):
|
|
return self._state == other
|
|
if isinstance(other, _StubState):
|
|
return self._state == other._state
|
|
return NotImplemented
|
|
|
|
|
|
def _make_mock_plan(
|
|
*,
|
|
plan_id: str = _PLAN_ID,
|
|
phase: PlanPhase = PlanPhase.EXECUTE,
|
|
state: ProcessingState = ProcessingState.COMPLETE,
|
|
changeset_id: str | None = _CHANGESET_ID,
|
|
validation_summary: dict[str, Any] | None = None,
|
|
is_terminal: bool = False,
|
|
error_details: dict[str, str] | None = None,
|
|
sandbox_refs: list[str] | None = None,
|
|
) -> MagicMock:
|
|
"""Create a mock Plan object with sensible defaults."""
|
|
plan = MagicMock()
|
|
plan.identity.plan_id = plan_id
|
|
plan.phase = _StubPhase(phase)
|
|
plan.processing_state = _StubState(state)
|
|
plan.changeset_id = changeset_id
|
|
plan.validation_summary = validation_summary
|
|
plan.is_terminal = is_terminal
|
|
plan.error_details = error_details
|
|
plan.timestamps = PlanTimestamps()
|
|
plan.sandbox_refs = sandbox_refs or ["sandbox-ref-1"]
|
|
return plan
|
|
|
|
|
|
def _make_changeset(
|
|
plan_id: str = _PLAN_ID,
|
|
changeset_id: str = _CHANGESET_ID,
|
|
entries: list[ChangeEntry] | None = None,
|
|
) -> SpecChangeSet:
|
|
"""Create a SpecChangeSet with optional entries."""
|
|
cs = SpecChangeSet(changeset_id=changeset_id, plan_id=plan_id)
|
|
if entries:
|
|
for entry in entries:
|
|
cs.add_change(entry)
|
|
return cs
|
|
|
|
|
|
def _make_modify_entry(
|
|
plan_id: str = _PLAN_ID,
|
|
path: str = "src/app.py",
|
|
before_hash: str | None = "abcdef0123456789",
|
|
after_hash: str | None = "123456abcdefghij",
|
|
) -> ChangeEntry:
|
|
"""Create a modify ChangeEntry."""
|
|
return ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="RES001",
|
|
tool_name="builtin/test-tool",
|
|
operation=ChangeOperation.MODIFY,
|
|
path=path,
|
|
before_hash=before_hash,
|
|
after_hash=after_hash,
|
|
)
|
|
|
|
|
|
def _make_create_entry(
|
|
plan_id: str = _PLAN_ID,
|
|
path: str = "src/new_file.py",
|
|
after_hash: str | None = "aaaaaa1111111111",
|
|
) -> ChangeEntry:
|
|
"""Create a create ChangeEntry."""
|
|
return ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="RES002",
|
|
tool_name="builtin/test-tool",
|
|
operation=ChangeOperation.CREATE,
|
|
path=path,
|
|
before_hash=None,
|
|
after_hash=after_hash,
|
|
)
|
|
|
|
|
|
def _make_delete_entry(
|
|
plan_id: str = _PLAN_ID,
|
|
path: str = "src/old_file.py",
|
|
before_hash: str | None = "bbbbbb2222222222",
|
|
) -> ChangeEntry:
|
|
"""Create a delete ChangeEntry."""
|
|
return ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="RES003",
|
|
tool_name="builtin/test-tool",
|
|
operation=ChangeOperation.DELETE,
|
|
path=path,
|
|
before_hash=before_hash,
|
|
after_hash=None,
|
|
)
|
|
|
|
|
|
def _make_rename_entry(
|
|
plan_id: str = _PLAN_ID,
|
|
path: str = "src/renamed.py",
|
|
) -> ChangeEntry:
|
|
"""Create a rename ChangeEntry."""
|
|
return ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="RES004",
|
|
tool_name="builtin/test-tool",
|
|
operation=ChangeOperation.RENAME,
|
|
path=path,
|
|
before_hash=None,
|
|
after_hash=None,
|
|
)
|
|
|
|
|
|
def _make_lifecycle_mock() -> MagicMock:
|
|
"""Create a mock PlanLifecycleService."""
|
|
lifecycle = MagicMock()
|
|
lifecycle._commit_plan = MagicMock()
|
|
lifecycle.complete_apply = MagicMock()
|
|
lifecycle.constrain_apply = MagicMock()
|
|
lifecycle.fail_apply = MagicMock()
|
|
return lifecycle
|
|
|
|
|
|
def _make_service(
|
|
lifecycle: MagicMock | None = None,
|
|
plan: MagicMock | None = None,
|
|
changeset_store: Any | None = None,
|
|
) -> PlanApplyService:
|
|
"""Create a PlanApplyService with mocked dependencies."""
|
|
lc = lifecycle or _make_lifecycle_mock()
|
|
if plan is not None:
|
|
lc.get_plan.return_value = plan
|
|
svc = PlanApplyService(
|
|
lifecycle_service=lc,
|
|
changeset_store=changeset_store,
|
|
)
|
|
return svc
|
|
|
|
|
|
# ======================================================================
|
|
# _operation_label steps
|
|
# ======================================================================
|
|
|
|
|
|
@when('pas_cov I call _operation_label with "{op}"')
|
|
def step_call_operation_label(context: Context, op: str) -> None:
|
|
"""Call _operation_label with the given operation string."""
|
|
context.pas_label_result = _operation_label(op)
|
|
|
|
|
|
@then('pas_cov the label should be "{expected}"')
|
|
def step_label_should_be(context: Context, expected: str) -> None:
|
|
"""Assert _operation_label returned the expected string."""
|
|
assert context.pas_label_result == expected, (
|
|
f"Expected label '{expected}', got '{context.pas_label_result}'"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# SpecChangeSet construction steps (shared)
|
|
# ======================================================================
|
|
|
|
|
|
@given("pas_cov a SpecChangeSet with no entries")
|
|
def step_changeset_no_entries(context: Context) -> None:
|
|
"""Create an empty SpecChangeSet."""
|
|
context.pas_changeset = _make_changeset()
|
|
|
|
|
|
@given("pas_cov a SpecChangeSet with a modify entry having both hashes")
|
|
def step_changeset_modify_both_hashes(context: Context) -> None:
|
|
"""Create a SpecChangeSet with one modify entry that has both hashes."""
|
|
entry = _make_modify_entry()
|
|
context.pas_changeset = _make_changeset(entries=[entry])
|
|
|
|
|
|
@given("pas_cov a SpecChangeSet with a create entry having only after_hash")
|
|
def step_changeset_create_after_hash(context: Context) -> None:
|
|
"""Create a SpecChangeSet with one create entry that has only after_hash."""
|
|
entry = _make_create_entry()
|
|
context.pas_changeset = _make_changeset(entries=[entry])
|
|
|
|
|
|
@given("pas_cov a SpecChangeSet with a delete entry having only before_hash")
|
|
def step_changeset_delete_before_hash(context: Context) -> None:
|
|
"""Create a SpecChangeSet with one delete entry that has only before_hash."""
|
|
entry = _make_delete_entry()
|
|
context.pas_changeset = _make_changeset(entries=[entry])
|
|
|
|
|
|
@given('pas_cov a SpecChangeSet with an entry having operation "{op}"')
|
|
def step_changeset_with_operation(context: Context, op: str) -> None:
|
|
"""Create a SpecChangeSet with an entry having the given operation."""
|
|
if op == "rename":
|
|
entry = _make_rename_entry()
|
|
elif op == "create":
|
|
entry = _make_create_entry()
|
|
elif op == "delete":
|
|
entry = _make_delete_entry()
|
|
else:
|
|
entry = _make_modify_entry()
|
|
context.pas_changeset = _make_changeset(entries=[entry])
|
|
|
|
|
|
# ======================================================================
|
|
# _render_diff_plain steps
|
|
# ======================================================================
|
|
|
|
|
|
@when("pas_cov I render diff plain")
|
|
def step_render_diff_plain(context: Context) -> None:
|
|
"""Call _render_diff_plain with the context changeset."""
|
|
context.pas_diff_result = _render_diff_plain(context.pas_changeset)
|
|
|
|
|
|
@then('pas_cov the plain diff should be "{expected}"')
|
|
def step_plain_diff_exact(context: Context, expected: str) -> None:
|
|
"""Assert the plain diff output matches exactly."""
|
|
assert context.pas_diff_result == expected, (
|
|
f"Expected '{expected}', got '{context.pas_diff_result}'"
|
|
)
|
|
|
|
|
|
@then('pas_cov the plain diff should contain "{text}"')
|
|
def step_plain_diff_contains(context: Context, text: str) -> None:
|
|
"""Assert the plain diff output contains the given text."""
|
|
assert text in context.pas_diff_result, (
|
|
f"Expected plain diff to contain '{text}', got:\n{context.pas_diff_result}"
|
|
)
|
|
|
|
|
|
@then('pas_cov the plain diff should not contain "{text}"')
|
|
def step_plain_diff_not_contains(context: Context, text: str) -> None:
|
|
"""Assert the plain diff output does not contain the given text."""
|
|
assert text not in context.pas_diff_result, (
|
|
f"Expected plain diff NOT to contain '{text}', got:\n{context.pas_diff_result}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _render_diff_rich steps
|
|
# ======================================================================
|
|
|
|
|
|
@when("pas_cov I render diff rich")
|
|
def step_render_diff_rich(context: Context) -> None:
|
|
"""Call _render_diff_rich with the context changeset."""
|
|
context.pas_diff_result = _render_diff_rich(context.pas_changeset)
|
|
|
|
|
|
@then('pas_cov the rich diff should contain "{text}"')
|
|
def step_rich_diff_contains(context: Context, text: str) -> None:
|
|
"""Assert the rich diff output contains the given text."""
|
|
assert text in context.pas_diff_result, (
|
|
f"Expected rich diff to contain '{text}', got:\n{context.pas_diff_result}"
|
|
)
|
|
|
|
|
|
@then('pas_cov the rich diff should not contain "{text}"')
|
|
def step_rich_diff_not_contains(context: Context, text: str) -> None:
|
|
"""Assert the rich diff output does not contain the given text."""
|
|
assert text not in context.pas_diff_result, (
|
|
f"Expected rich diff NOT to contain '{text}', got:\n{context.pas_diff_result}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _render_diff_json steps
|
|
# ======================================================================
|
|
|
|
|
|
@when("pas_cov I render diff json")
|
|
def step_render_diff_json(context: Context) -> None:
|
|
"""Call _render_diff_json with the context changeset."""
|
|
context.pas_json_result = _render_diff_json(context.pas_changeset)
|
|
|
|
|
|
@then('pas_cov the json diff should have key "{key}"')
|
|
def step_json_diff_has_key(context: Context, key: str) -> None:
|
|
"""Assert the JSON diff dict has the given key."""
|
|
assert key in context.pas_json_result, (
|
|
f"Expected JSON diff to have key '{key}', keys: {list(context.pas_json_result.keys())}"
|
|
)
|
|
|
|
|
|
@then('pas_cov the json diff should have key "{key}" with value {value}')
|
|
def step_json_diff_key_value(context: Context, key: str, value: str) -> None:
|
|
"""Assert the JSON diff dict has the given key with the given value."""
|
|
assert key in context.pas_json_result, f"Expected key '{key}' in JSON diff"
|
|
actual = context.pas_json_result[key]
|
|
expected = int(value) if value.isdigit() else value
|
|
assert actual == expected, f"Expected '{key}' to be {expected}, got {actual}"
|
|
|
|
|
|
@then("pas_cov the json diff entries should have length {n}")
|
|
def step_json_diff_entries_length(context: Context, n: str) -> None:
|
|
"""Assert the JSON diff entries list has the given length."""
|
|
entries = context.pas_json_result["entries"]
|
|
assert len(entries) == int(n), f"Expected {n} entries, got {len(entries)}"
|
|
|
|
|
|
@then('pas_cov the json diff first entry should have "{key}" equal to "{value}"')
|
|
def step_json_diff_first_entry_value(context: Context, key: str, value: str) -> None:
|
|
"""Assert the first entry in JSON diff has the given key/value."""
|
|
entry = context.pas_json_result["entries"][0]
|
|
assert entry[key] == value, (
|
|
f"Expected entry['{key}'] == '{value}', got '{entry[key]}'"
|
|
)
|
|
|
|
|
|
@then('pas_cov the json diff first entry should have key "{key}"')
|
|
def step_json_diff_first_entry_has_key(context: Context, key: str) -> None:
|
|
"""Assert the first entry in JSON diff has the given key."""
|
|
entry = context.pas_json_result["entries"][0]
|
|
assert key in entry, (
|
|
f"Expected entry to have key '{key}', keys: {list(entry.keys())}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _build_artifacts_dict steps
|
|
# ======================================================================
|
|
|
|
|
|
@given('pas_cov a mock plan with changeset_id "{cs_id}"')
|
|
def step_mock_plan_with_changeset_id(context: Context, cs_id: str) -> None:
|
|
"""Create a mock plan with the given changeset_id."""
|
|
context.pas_plan = _make_mock_plan(changeset_id=cs_id)
|
|
|
|
|
|
@given('pas_cov a mock plan with changeset_id "{cs_id}" and validation summary')
|
|
def step_mock_plan_with_validation(context: Context, cs_id: str) -> None:
|
|
"""Create a mock plan with changeset_id and validation summary."""
|
|
context.pas_plan = _make_mock_plan(
|
|
changeset_id=cs_id,
|
|
validation_summary={"total": 3, "required_passed": 2, "required_failed": 1},
|
|
)
|
|
|
|
|
|
@given("pas_cov a mock plan with apply summary in error_details")
|
|
def step_mock_plan_with_apply_summary(context: Context) -> None:
|
|
"""Create a mock plan with apply summary in error_details."""
|
|
context.pas_plan = _make_mock_plan(
|
|
changeset_id=_CHANGESET_ID,
|
|
validation_summary=None,
|
|
error_details={
|
|
"apply_files_changed": "5",
|
|
"apply_validations_run": "3",
|
|
},
|
|
)
|
|
|
|
|
|
@when("pas_cov I build artifacts dict")
|
|
def step_build_artifacts_dict(context: Context) -> None:
|
|
"""Call _build_artifacts_dict with plan and changeset."""
|
|
context.pas_artifacts = _build_artifacts_dict(
|
|
context.pas_plan,
|
|
context.pas_changeset,
|
|
)
|
|
|
|
|
|
@when("pas_cov I build artifacts dict with no changeset")
|
|
def step_build_artifacts_dict_no_changeset(context: Context) -> None:
|
|
"""Call _build_artifacts_dict with plan and None changeset."""
|
|
context.pas_artifacts = _build_artifacts_dict(context.pas_plan, None)
|
|
|
|
|
|
@then('pas_cov the artifacts dict should have key "{key}"')
|
|
def step_artifacts_dict_has_key(context: Context, key: str) -> None:
|
|
"""Assert artifacts dict contains the given key."""
|
|
assert key in context.pas_artifacts, (
|
|
f"Expected artifacts to have key '{key}', keys: {list(context.pas_artifacts.keys())}"
|
|
)
|
|
|
|
|
|
@then("pas_cov the artifacts dict files_changed list should have length {n}")
|
|
def step_artifacts_files_changed_length(context: Context, n: str) -> None:
|
|
"""Assert the files_changed list has the expected length."""
|
|
fc = context.pas_artifacts["files_changed"]
|
|
assert len(fc) == int(n), f"Expected files_changed length {n}, got {len(fc)}"
|
|
|
|
|
|
@then("pas_cov the artifacts dict changeset_summary should be null")
|
|
def step_artifacts_changeset_summary_null(context: Context) -> None:
|
|
"""Assert changeset_summary is None in artifacts dict."""
|
|
assert context.pas_artifacts["changeset_summary"] is None, (
|
|
f"Expected changeset_summary to be None, got {context.pas_artifacts['changeset_summary']}"
|
|
)
|
|
|
|
|
|
@then('pas_cov the artifacts dict apply_summary files_changed should be "{value}"')
|
|
def step_artifacts_apply_summary_files(context: Context, value: str) -> None:
|
|
"""Assert apply_summary.files_changed has the expected value."""
|
|
summary = context.pas_artifacts["apply_summary"]
|
|
assert summary["files_changed"] == value, (
|
|
f"Expected files_changed '{value}', got '{summary['files_changed']}'"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# PlanApplyService initialization steps
|
|
# ======================================================================
|
|
|
|
|
|
@when("pas_cov I create PlanApplyService with None lifecycle service")
|
|
def step_create_service_none_lifecycle(context: Context) -> None:
|
|
"""Attempt to create PlanApplyService with None lifecycle."""
|
|
context.pas_error = None
|
|
try:
|
|
PlanApplyService(lifecycle_service=None) # type: ignore[arg-type]
|
|
except ValidationError as exc:
|
|
context.pas_error = exc
|
|
|
|
|
|
@then("pas_cov a ValidationError should be raised")
|
|
def step_validation_error_raised(context: Context) -> None:
|
|
"""Assert that a ValidationError was raised."""
|
|
assert context.pas_error is not None, "Expected ValidationError but none was raised"
|
|
assert isinstance(context.pas_error, ValidationError), (
|
|
f"Expected ValidationError, got {type(context.pas_error).__name__}"
|
|
)
|
|
|
|
|
|
@given("pas_cov a mock lifecycle service")
|
|
def step_mock_lifecycle(context: Context) -> None:
|
|
"""Create a mock lifecycle service on context."""
|
|
context.pas_lifecycle = _make_lifecycle_mock()
|
|
|
|
|
|
@given("pas_cov a mock changeset store")
|
|
def step_mock_changeset_store(context: Context) -> None:
|
|
"""Create a mock changeset store on context."""
|
|
context.pas_store = MagicMock()
|
|
|
|
|
|
@when("pas_cov I create PlanApplyService with valid lifecycle and no store")
|
|
def step_create_service_no_store(context: Context) -> None:
|
|
"""Create PlanApplyService with valid lifecycle and no store."""
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=context.pas_lifecycle,
|
|
changeset_store=None,
|
|
)
|
|
|
|
|
|
@when("pas_cov I create PlanApplyService with valid lifecycle and store")
|
|
def step_create_service_with_store(context: Context) -> None:
|
|
"""Create PlanApplyService with valid lifecycle and store."""
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=context.pas_lifecycle,
|
|
changeset_store=context.pas_store,
|
|
)
|
|
|
|
|
|
@then("pas_cov the service should be created successfully")
|
|
def step_service_created(context: Context) -> None:
|
|
"""Assert the PlanApplyService instance was created."""
|
|
assert context.pas_service is not None, "Service should not be None"
|
|
assert isinstance(context.pas_service, PlanApplyService), (
|
|
f"Expected PlanApplyService, got {type(context.pas_service).__name__}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# diff method steps
|
|
# ======================================================================
|
|
|
|
|
|
@given("pas_cov a service with a plan that has no changeset_id")
|
|
def step_service_plan_no_changeset(context: Context) -> None:
|
|
"""Create a service with a plan whose changeset_id is None."""
|
|
plan = _make_mock_plan(changeset_id=None)
|
|
context.pas_plan = plan
|
|
context.pas_service = _make_service(plan=plan)
|
|
|
|
|
|
@given("pas_cov a service with a plan that has a changeset with entries")
|
|
def step_service_plan_with_changeset_entries(context: Context) -> None:
|
|
"""Create a service with a plan whose changeset has entries in the store."""
|
|
plan = _make_mock_plan(changeset_id=_CHANGESET_ID)
|
|
context.pas_plan = plan
|
|
cs = _make_changeset(entries=[_make_modify_entry()])
|
|
store = MagicMock()
|
|
store.get.return_value = cs
|
|
context.pas_changeset = cs
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
)
|
|
|
|
|
|
@given("pas_cov a service with a plan that has an empty changeset")
|
|
def step_service_plan_empty_changeset(context: Context) -> None:
|
|
"""Create a service with a plan whose changeset has no entries."""
|
|
plan = _make_mock_plan(changeset_id=_CHANGESET_ID)
|
|
context.pas_plan = plan
|
|
cs = _make_changeset()
|
|
store = MagicMock()
|
|
store.get.return_value = cs
|
|
context.pas_changeset = cs
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
)
|
|
|
|
|
|
@when("pas_cov I call diff on the service")
|
|
def step_call_diff_default(context: Context) -> None:
|
|
"""Call diff on the service with default format."""
|
|
context.pas_error = None
|
|
try:
|
|
context.pas_diff_result = context.pas_service.diff(_PLAN_ID)
|
|
except PlanError as exc:
|
|
context.pas_error = exc
|
|
|
|
|
|
@when('pas_cov I call diff with format "{fmt}"')
|
|
def step_call_diff_format(context: Context, fmt: str) -> None:
|
|
"""Call diff on the service with specified format."""
|
|
context.pas_error = None
|
|
try:
|
|
context.pas_diff_result = context.pas_service.diff(_PLAN_ID, fmt=fmt)
|
|
except PlanError as exc:
|
|
context.pas_error = exc
|
|
|
|
|
|
@then('pas_cov a PlanError should be raised with message containing "{text}"')
|
|
def step_plan_error_raised_with_message(context: Context, text: str) -> None:
|
|
"""Assert a PlanError was raised with the given text in its message."""
|
|
assert context.pas_error is not None, "Expected PlanError but none was raised"
|
|
assert isinstance(context.pas_error, PlanError), (
|
|
f"Expected PlanError, got {type(context.pas_error).__name__}"
|
|
)
|
|
assert text in str(context.pas_error), (
|
|
f"Expected message to contain '{text}', got: {context.pas_error}"
|
|
)
|
|
|
|
|
|
@then('pas_cov the diff result should contain "{text}"')
|
|
def step_diff_result_contains(context: Context, text: str) -> None:
|
|
"""Assert the diff result contains the given text."""
|
|
assert text in context.pas_diff_result, (
|
|
f"Expected diff result to contain '{text}', got:\n{context.pas_diff_result}"
|
|
)
|
|
|
|
|
|
@then('pas_cov the diff result should not contain "{text}"')
|
|
def step_diff_result_not_contains(context: Context, text: str) -> None:
|
|
"""Assert the diff result does not contain the given text."""
|
|
assert text not in context.pas_diff_result, (
|
|
f"Expected diff result NOT to contain '{text}'"
|
|
)
|
|
|
|
|
|
@then("pas_cov the diff result should be valid JSON")
|
|
def step_diff_result_is_json(context: Context) -> None:
|
|
"""Assert the diff result is valid JSON."""
|
|
try:
|
|
parsed = json.loads(context.pas_diff_result)
|
|
assert isinstance(parsed, dict), "Expected JSON object"
|
|
except json.JSONDecodeError as exc:
|
|
assert False, f"Diff result is not valid JSON: {exc}" # noqa: B011
|
|
|
|
|
|
# ======================================================================
|
|
# artifacts method steps
|
|
# ======================================================================
|
|
|
|
|
|
@when('pas_cov I call artifacts with format "{fmt}"')
|
|
def step_call_artifacts_format(context: Context, fmt: str) -> None:
|
|
"""Call artifacts on the service with specified format."""
|
|
from contextlib import redirect_stdout
|
|
from io import StringIO
|
|
|
|
buf = StringIO()
|
|
with redirect_stdout(buf):
|
|
result = context.pas_service.artifacts(_PLAN_ID, fmt=fmt)
|
|
context.pas_artifacts_result = result or buf.getvalue().rstrip("\n")
|
|
|
|
|
|
@then('pas_cov the artifacts result should contain "{text}"')
|
|
def step_artifacts_result_contains(context: Context, text: str) -> None:
|
|
"""Assert artifacts result contains the given text."""
|
|
assert text in context.pas_artifacts_result, (
|
|
f"Expected artifacts to contain '{text}', got:\n{context.pas_artifacts_result}"
|
|
)
|
|
|
|
|
|
@then("pas_cov the artifacts result should be a non-empty string")
|
|
def step_artifacts_result_non_empty(context: Context) -> None:
|
|
"""Assert artifacts result is a non-empty string."""
|
|
assert isinstance(context.pas_artifacts_result, str), "Expected string"
|
|
assert len(context.pas_artifacts_result) > 0, "Expected non-empty string"
|
|
|
|
|
|
# ======================================================================
|
|
# persist_apply_summary steps
|
|
# ======================================================================
|
|
|
|
|
|
@given("pas_cov a service with a plan that has no error_details")
|
|
def step_service_plan_no_error_details(context: Context) -> None:
|
|
"""Create a service with a plan that has no error_details."""
|
|
plan = _make_mock_plan(error_details=None)
|
|
context.pas_plan = plan
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
lifecycle.fail_apply.return_value = _make_mock_plan(
|
|
state=ProcessingState.ERRORED, is_terminal=True
|
|
)
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
)
|
|
|
|
|
|
@given("pas_cov a service with a plan that has existing error_details")
|
|
def step_service_plan_existing_error_details(context: Context) -> None:
|
|
"""Create a service with a plan that has pre-existing error_details."""
|
|
plan = _make_mock_plan(
|
|
error_details={"existing_key": "existing_value"},
|
|
)
|
|
context.pas_plan = plan
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
lifecycle.fail_apply.return_value = _make_mock_plan(
|
|
state=ProcessingState.ERRORED, is_terminal=True
|
|
)
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
)
|
|
|
|
|
|
@when(
|
|
"pas_cov I persist apply summary with {files} files and {validations} validations"
|
|
)
|
|
def step_persist_apply_summary(context: Context, files: str, validations: str) -> None:
|
|
"""Call persist_apply_summary on the service."""
|
|
context.pas_returned_plan = context.pas_service.persist_apply_summary(
|
|
plan_id=_PLAN_ID,
|
|
files_changed=int(files),
|
|
validations_run=int(validations),
|
|
)
|
|
|
|
|
|
@then('pas_cov the plan error_details should contain "{key}" with value "{value}"')
|
|
def step_plan_error_details_value(context: Context, key: str, value: str) -> None:
|
|
"""Assert the plan error_details has the given key with expected value."""
|
|
details = context.pas_plan.error_details
|
|
assert details is not None, "error_details should not be None"
|
|
assert key in details, f"Expected key '{key}' in error_details, got: {details}"
|
|
assert details[key] == value, (
|
|
f"Expected error_details['{key}'] == '{value}', got '{details[key]}'"
|
|
)
|
|
|
|
|
|
@then('pas_cov the plan error_details should contain key "{key}"')
|
|
def step_plan_error_details_has_key(context: Context, key: str) -> None:
|
|
"""Assert the plan error_details has the given key."""
|
|
details = context.pas_plan.error_details
|
|
assert details is not None, "error_details should not be None"
|
|
assert key in details, (
|
|
f"Expected key '{key}' in error_details, keys: {list(details.keys())}"
|
|
)
|
|
|
|
|
|
@then("pas_cov the lifecycle _commit_plan should have been called")
|
|
def step_commit_plan_called(context: Context) -> None:
|
|
"""Assert lifecycle._commit_plan was called."""
|
|
context.pas_lifecycle._commit_plan.assert_called()
|
|
|
|
|
|
# ======================================================================
|
|
# handle_merge_failure steps
|
|
# ======================================================================
|
|
|
|
|
|
@when('pas_cov I handle merge failure with conflict "{details}"')
|
|
def step_handle_merge_failure(context: Context, details: str) -> None:
|
|
"""Call handle_merge_failure on the service."""
|
|
context.pas_returned_plan = context.pas_service.handle_merge_failure(
|
|
plan_id=_PLAN_ID,
|
|
conflict_details=details,
|
|
)
|
|
|
|
|
|
@then("pas_cov the lifecycle fail_apply should have been called")
|
|
def step_fail_apply_called(context: Context) -> None:
|
|
"""Assert lifecycle.fail_apply was called."""
|
|
context.pas_lifecycle.fail_apply.assert_called()
|
|
|
|
|
|
@then("pas_cov the returned plan should reflect errored state")
|
|
def step_returned_plan_errored(context: Context) -> None:
|
|
"""Assert the returned plan is in errored state."""
|
|
assert context.pas_returned_plan is not None, "Returned plan should not be None"
|
|
|
|
|
|
@then('pas_cov the committed plan error_details should contain "{key}"')
|
|
def step_committed_error_details_contains(context: Context, key: str) -> None:
|
|
"""Assert the plan error_details set before commit contains the key."""
|
|
# The error_details were set on the mock plan directly
|
|
details = context.pas_plan.error_details
|
|
assert details is not None, "error_details should not be None"
|
|
assert key in details, (
|
|
f"Expected key '{key}' in committed error_details, keys: {list(details.keys())}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# guard_empty_changeset steps
|
|
# ======================================================================
|
|
|
|
|
|
@when("pas_cov I call guard_empty_changeset")
|
|
def step_call_guard(context: Context) -> None:
|
|
"""Call guard_empty_changeset on the service."""
|
|
context.pas_error = None
|
|
try:
|
|
context.pas_guard_result = context.pas_service.guard_empty_changeset(_PLAN_ID)
|
|
except PlanError as exc:
|
|
context.pas_error = exc
|
|
|
|
|
|
@when("pas_cov I call guard_empty_changeset with allow_empty true")
|
|
def step_call_guard_allow_empty(context: Context) -> None:
|
|
"""Call guard_empty_changeset with allow_empty=True."""
|
|
context.pas_error = None
|
|
try:
|
|
context.pas_guard_result = context.pas_service.guard_empty_changeset(
|
|
_PLAN_ID, allow_empty=True
|
|
)
|
|
except PlanError as exc:
|
|
context.pas_error = exc
|
|
|
|
|
|
@then("pas_cov the guard should return true")
|
|
def step_guard_returns_true(context: Context) -> None:
|
|
"""Assert the guard returned True."""
|
|
assert context.pas_guard_result is True, (
|
|
f"Expected guard to return True, got {context.pas_guard_result}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _extract_validation_counts steps
|
|
# ======================================================================
|
|
|
|
|
|
@when("pas_cov I extract validation counts from None")
|
|
def step_extract_counts_none(context: Context) -> None:
|
|
"""Call _extract_validation_counts with None."""
|
|
context.pas_counts = PlanApplyService._extract_validation_counts(None)
|
|
|
|
|
|
@when(
|
|
"pas_cov I extract validation counts from full summary {p} passed {f} failed {t} total"
|
|
)
|
|
def step_extract_counts_full(context: Context, p: str, f: str, t: str) -> None:
|
|
"""Call _extract_validation_counts with a full summary dict."""
|
|
vs = {
|
|
"required_passed": int(p),
|
|
"required_failed": int(f),
|
|
"total": int(t),
|
|
}
|
|
context.pas_counts = PlanApplyService._extract_validation_counts(vs)
|
|
|
|
|
|
@when("pas_cov I extract validation counts from partial summary {p} passed {f} failed")
|
|
def step_extract_counts_no_total(context: Context, p: str, f: str) -> None:
|
|
"""Call _extract_validation_counts with no total key."""
|
|
vs = {
|
|
"required_passed": int(p),
|
|
"required_failed": int(f),
|
|
}
|
|
context.pas_counts = PlanApplyService._extract_validation_counts(vs)
|
|
|
|
|
|
@then("pas_cov the counts should be {p} passed {f} failed {t} total")
|
|
def step_assert_counts(context: Context, p: str, f: str, t: str) -> None:
|
|
"""Assert the extracted validation counts match."""
|
|
expected = (int(p), int(f), int(t))
|
|
assert context.pas_counts == expected, (
|
|
f"Expected counts {expected}, got {context.pas_counts}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _resolve_changeset steps
|
|
# ======================================================================
|
|
|
|
|
|
@when("pas_cov I call _resolve_changeset")
|
|
def step_call_resolve_changeset(context: Context) -> None:
|
|
"""Call _resolve_changeset on the service."""
|
|
plan = context.pas_plan
|
|
context.pas_resolved_cs = context.pas_service._resolve_changeset(plan)
|
|
|
|
|
|
@then("pas_cov the resolved changeset should be None")
|
|
def step_resolved_cs_none(context: Context) -> None:
|
|
"""Assert _resolve_changeset returned None."""
|
|
assert context.pas_resolved_cs is None, (
|
|
f"Expected None, got {context.pas_resolved_cs}"
|
|
)
|
|
|
|
|
|
@then("pas_cov the resolved changeset should have entries")
|
|
def step_resolved_cs_has_entries(context: Context) -> None:
|
|
"""Assert _resolve_changeset returned a changeset with entries."""
|
|
assert context.pas_resolved_cs is not None, "Resolved changeset should not be None"
|
|
assert len(context.pas_resolved_cs.entries) > 0, (
|
|
f"Expected entries, got {len(context.pas_resolved_cs.entries)}"
|
|
)
|
|
|
|
|
|
@then("pas_cov the resolved changeset should have no entries")
|
|
def step_resolved_cs_no_entries(context: Context) -> None:
|
|
"""Assert _resolve_changeset returned a changeset with no entries."""
|
|
assert context.pas_resolved_cs is not None, "Resolved changeset should not be None"
|
|
assert len(context.pas_resolved_cs.entries) == 0, (
|
|
f"Expected 0 entries, got {len(context.pas_resolved_cs.entries)}"
|
|
)
|
|
|
|
|
|
@then("pas_cov the resolved changeset should have the plan changeset_id")
|
|
def step_resolved_cs_has_changeset_id(context: Context) -> None:
|
|
"""Assert the resolved changeset has the plan's changeset_id."""
|
|
assert context.pas_resolved_cs.changeset_id == context.pas_plan.changeset_id, (
|
|
f"Expected changeset_id '{context.pas_plan.changeset_id}', "
|
|
f"got '{context.pas_resolved_cs.changeset_id}'"
|
|
)
|
|
|
|
|
|
@given(
|
|
"pas_cov a service with a plan that has changeset_id and store returns a changeset"
|
|
)
|
|
def step_service_store_returns_changeset(context: Context) -> None:
|
|
"""Create a service where the store returns a populated changeset."""
|
|
plan = _make_mock_plan(changeset_id=_CHANGESET_ID)
|
|
context.pas_plan = plan
|
|
cs = _make_changeset(entries=[_make_modify_entry()])
|
|
store = MagicMock()
|
|
store.get.return_value = cs
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
)
|
|
|
|
|
|
@given("pas_cov a service with a plan that has changeset_id but store returns None")
|
|
def step_service_store_returns_none(context: Context) -> None:
|
|
"""Create a service where the store returns None (miss)."""
|
|
plan = _make_mock_plan(changeset_id=_CHANGESET_ID)
|
|
context.pas_plan = plan
|
|
store = MagicMock()
|
|
store.get.return_value = None
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
)
|
|
|
|
|
|
@given("pas_cov a service with a plan that has changeset_id and no store")
|
|
def step_service_no_store(context: Context) -> None:
|
|
"""Create a service with no changeset store configured."""
|
|
plan = _make_mock_plan(changeset_id=_CHANGESET_ID)
|
|
context.pas_plan = plan
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# cleanup_changeset steps
|
|
# ======================================================================
|
|
|
|
|
|
@given("pas_cov a service with a mock lifecycle")
|
|
def step_service_mock_lifecycle(context: Context) -> None:
|
|
"""Create a basic service with a mock lifecycle."""
|
|
lifecycle = _make_lifecycle_mock()
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
)
|
|
|
|
|
|
@when("pas_cov I call cleanup_changeset with empty plan_id")
|
|
def step_cleanup_empty_plan_id(context: Context) -> None:
|
|
"""Call cleanup_changeset with empty plan_id."""
|
|
context.pas_error = None
|
|
try:
|
|
context.pas_service.cleanup_changeset("")
|
|
except ValidationError as exc:
|
|
context.pas_error = exc
|
|
|
|
|
|
@given("pas_cov a service with a store that has delete_for_plan returning {n}")
|
|
def step_service_store_with_delete(context: Context, n: str) -> None:
|
|
"""Create a service whose store.delete_for_plan returns n."""
|
|
store = MagicMock()
|
|
store.delete_for_plan.return_value = int(n)
|
|
lifecycle = _make_lifecycle_mock()
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
)
|
|
|
|
|
|
@when('pas_cov I call cleanup_changeset with plan_id "{plan_id}"')
|
|
def step_cleanup_with_plan_id(context: Context, plan_id: str) -> None:
|
|
"""Call cleanup_changeset with the given plan_id."""
|
|
context.pas_cleanup_result = context.pas_service.cleanup_changeset(plan_id)
|
|
|
|
|
|
@then("pas_cov cleanup should return {n}")
|
|
def step_cleanup_returns(context: Context, n: str) -> None:
|
|
"""Assert cleanup_changeset returned the expected count."""
|
|
assert context.pas_cleanup_result == int(n), (
|
|
f"Expected cleanup to return {n}, got {context.pas_cleanup_result}"
|
|
)
|
|
|
|
|
|
@given("pas_cov a service with no changeset store")
|
|
def step_service_no_changeset_store(context: Context) -> None:
|
|
"""Create a service with changeset_store=None."""
|
|
lifecycle = _make_lifecycle_mock()
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
)
|
|
|
|
|
|
@given("pas_cov a service with a store that lacks delete_for_plan")
|
|
def step_service_store_no_delete(context: Context) -> None:
|
|
"""Create a service whose store has no delete_for_plan attribute."""
|
|
|
|
class _StubStore:
|
|
"""Stub store without delete_for_plan method."""
|
|
|
|
def get(self, changeset_id: str) -> None:
|
|
"""Return None for any lookup."""
|
|
return None
|
|
|
|
lifecycle = _make_lifecycle_mock()
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=_StubStore(),
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# apply_with_validation_gate: exception branches
|
|
# ======================================================================
|
|
|
|
|
|
@given("pas_cov a service where constrain_apply raises PlanError")
|
|
def step_service_constrain_raises(context: Context) -> None:
|
|
"""Create a service where constrain_apply raises PlanError."""
|
|
plan = _make_mock_plan(
|
|
changeset_id=_CHANGESET_ID,
|
|
is_terminal=False,
|
|
)
|
|
context.pas_plan = plan
|
|
cs = _make_changeset(entries=[_make_modify_entry()])
|
|
store = MagicMock()
|
|
store.get.return_value = cs
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
lifecycle.constrain_apply.side_effect = PlanError("Not in Apply phase")
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
)
|
|
|
|
|
|
@given("pas_cov the plan has failed validations")
|
|
def step_plan_failed_validations(context: Context) -> None:
|
|
"""Set the plan's validation_summary to have failed validations."""
|
|
context.pas_plan.validation_summary = {
|
|
"total": 3,
|
|
"required_passed": 1,
|
|
"required_failed": 2,
|
|
}
|
|
|
|
|
|
@given("pas_cov a service where complete_apply raises PlanError")
|
|
def step_service_complete_raises(context: Context) -> None:
|
|
"""Create a service where complete_apply raises PlanError."""
|
|
plan = _make_mock_plan(
|
|
changeset_id=_CHANGESET_ID,
|
|
is_terminal=False,
|
|
)
|
|
context.pas_plan = plan
|
|
cs = _make_changeset(entries=[_make_modify_entry()])
|
|
store = MagicMock()
|
|
store.get.return_value = cs
|
|
lifecycle = _make_lifecycle_mock()
|
|
lifecycle.get_plan.return_value = plan
|
|
lifecycle.complete_apply.side_effect = PlanError("Not in Apply phase")
|
|
context.pas_lifecycle = lifecycle
|
|
context.pas_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
)
|
|
|
|
|
|
@given("pas_cov the plan has passing validations and entries")
|
|
def step_plan_passing_validations(context: Context) -> None:
|
|
"""Set the plan's validation_summary to all passing."""
|
|
context.pas_plan.validation_summary = {
|
|
"total": 2,
|
|
"required_passed": 2,
|
|
"required_failed": 0,
|
|
}
|
|
|
|
|
|
@when("pas_cov I call apply_with_validation_gate")
|
|
def step_call_apply_gate(context: Context) -> None:
|
|
"""Call apply_with_validation_gate on the service."""
|
|
context.pas_apply_result = context.pas_service.apply_with_validation_gate(
|
|
plan_id=_PLAN_ID,
|
|
)
|
|
|
|
|
|
@then('pas_cov the result outcome should be "{outcome}"')
|
|
def step_result_outcome(context: Context, outcome: str) -> None:
|
|
"""Assert the apply result outcome matches."""
|
|
expected = ApplyOutcome(outcome)
|
|
assert context.pas_apply_result.outcome == expected, (
|
|
f"Expected outcome '{outcome}', got '{context.pas_apply_result.outcome}'"
|
|
)
|
|
|
|
|
|
@then('pas_cov the result message should contain "{text}"')
|
|
def step_result_message_contains(context: Context, text: str) -> None:
|
|
"""Assert the apply result message contains the given text."""
|
|
assert text in context.pas_apply_result.message, (
|
|
f"Expected message to contain '{text}', got: {context.pas_apply_result.message}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# ApplyOutcome and ApplyResult model steps
|
|
# ======================================================================
|
|
|
|
|
|
@then('pas_cov ApplyOutcome should have value "{value}"')
|
|
def step_apply_outcome_value(context: Context, value: str) -> None:
|
|
"""Assert ApplyOutcome enum contains the given value."""
|
|
outcome = ApplyOutcome(value)
|
|
assert outcome.value == value, (
|
|
f"Expected ApplyOutcome value '{value}', got '{outcome.value}'"
|
|
)
|
|
|
|
|
|
@when("pas_cov I create an ApplyResult with whitespace in message")
|
|
def step_create_apply_result_whitespace(context: Context) -> None:
|
|
"""Create an ApplyResult with whitespace in message field."""
|
|
context.pas_apply_result_model = ApplyResult(
|
|
outcome=ApplyOutcome.APPLIED,
|
|
plan_id=_PLAN_ID,
|
|
message=" hello world ",
|
|
)
|
|
|
|
|
|
@then("pas_cov the ApplyResult message should be stripped")
|
|
def step_apply_result_stripped(context: Context) -> None:
|
|
"""Assert the ApplyResult message had whitespace stripped."""
|
|
assert context.pas_apply_result_model.message == "hello world", (
|
|
f"Expected 'hello world', got '{context.pas_apply_result_model.message}'"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _OP_COLORS steps
|
|
# ======================================================================
|
|
|
|
|
|
@then('pas_cov OP_COLORS should map "{key}" to "{value}"')
|
|
def step_op_colors_map(context: Context, key: str, value: str) -> None:
|
|
"""Assert _OP_COLORS maps the key to the expected value."""
|
|
assert key in _OP_COLORS, (
|
|
f"Expected key '{key}' in _OP_COLORS, keys: {list(_OP_COLORS.keys())}"
|
|
)
|
|
assert _OP_COLORS[key] == value, (
|
|
f"Expected _OP_COLORS['{key}'] == '{value}', got '{_OP_COLORS[key]}'"
|
|
)
|