160 lines
7.4 KiB
Plaintext
160 lines
7.4 KiB
Plaintext
*** Settings ***
|
|
Documentation Integration test: Plan actor integration for strategize/execute phases
|
|
Library OperatingSystem
|
|
Library Process
|
|
Library Collections
|
|
|
|
*** Variables ***
|
|
${PYTHON} python3
|
|
|
|
*** Test Cases ***
|
|
Strategize Stub Produces Decisions
|
|
[Documentation] Run strategize stub and verify decisions are produced.
|
|
${result}= Run Process ${PYTHON} -c
|
|
... ${STRATEGIZE_SCRIPT}
|
|
Log ${result.stdout}
|
|
Log ${result.stderr}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
Should Contain ${result.stdout} PASS:decisions=2
|
|
Should Contain ${result.stdout} PASS:root_id_set=True
|
|
|
|
Execute Stub Captures ChangeSet
|
|
[Documentation] Run execute stub and verify changeset is captured.
|
|
${result}= Run Process ${PYTHON} -c
|
|
... ${EXECUTE_SCRIPT}
|
|
Log ${result.stdout}
|
|
Log ${result.stderr}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
Should Contain ${result.stdout} PASS:changeset_id_set=True
|
|
Should Contain ${result.stdout} PASS:execute_complete=True
|
|
|
|
Full Lifecycle Strategize Then Execute
|
|
[Documentation] Run full strategize-then-execute lifecycle.
|
|
${result}= Run Process ${PYTHON} -c
|
|
... ${LIFECYCLE_SCRIPT}
|
|
Log ${result.stdout}
|
|
Log ${result.stderr}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
Should Contain ${result.stdout} PASS:strategize_ok
|
|
Should Contain ${result.stdout} PASS:execute_ok
|
|
Should Contain ${result.stdout} PASS:decision_root_set
|
|
|
|
Strategize Streaming Events Emitted
|
|
[Documentation] Verify strategize emits streaming events.
|
|
${result}= Run Process ${PYTHON} -c
|
|
... ${STREAM_SCRIPT}
|
|
Log ${result.stdout}
|
|
Log ${result.stderr}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
Should Contain ${result.stdout} PASS:strategize_started
|
|
Should Contain ${result.stdout} PASS:strategize_complete
|
|
|
|
Phase Guard Prevents Invalid Execute
|
|
[Documentation] Verify execute is blocked when plan is not ready.
|
|
${result}= Run Process ${PYTHON} -c
|
|
... ${GUARD_SCRIPT}
|
|
Log ${result.stdout}
|
|
Log ${result.stderr}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
Should Contain ${result.stdout} PASS:guard_blocked
|
|
|
|
*** Variables ***
|
|
${STRATEGIZE_SCRIPT} SEPARATOR=\n
|
|
... import os, sys
|
|
... sys.path.insert(0, os.path.join(os.getcwd(), "src"))
|
|
... from cleveragents.application.services.plan_executor import StrategizeStubActor, StrategyDecision
|
|
... from cleveragents.domain.models.core.plan import PlanInvariant, InvariantSource
|
|
... from ulid import ULID
|
|
... actor = StrategizeStubActor()
|
|
... result = actor.execute(
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}plan_id=str(ULID()),
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}definition_of_done="Tests pass" + chr(10) + "Coverage met",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}invariants=[PlanInvariant(text="No regressions", source=InvariantSource.ACTION)],
|
|
... )
|
|
... print(f"PASS:decisions={len(result.decisions)}")
|
|
... print(f"PASS:root_id_set={result.decision_root_id is not None}")
|
|
|
|
${EXECUTE_SCRIPT} SEPARATOR=\n
|
|
... import os, sys
|
|
... sys.path.insert(0, os.path.join(os.getcwd(), "src"))
|
|
... from cleveragents.application.services.plan_executor import ExecuteStubActor, StrategyDecision
|
|
... from ulid import ULID
|
|
... actor = ExecuteStubActor()
|
|
... decisions = [
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}StrategyDecision(decision_id=str(ULID()), step_text="Step 1", sequence=0),
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}StrategyDecision(decision_id=str(ULID()), step_text="Step 2", sequence=1),
|
|
... ]
|
|
... result = actor.execute(plan_id=str(ULID()), decisions=decisions)
|
|
... print(f"PASS:changeset_id_set={result.changeset_id is not None}")
|
|
... print(f"PASS:execute_complete=True")
|
|
|
|
${LIFECYCLE_SCRIPT} SEPARATOR=\n
|
|
... import os, sys
|
|
... sys.path.insert(0, os.path.join(os.getcwd(), "src"))
|
|
... from cleveragents.config.settings import Settings
|
|
... from cleveragents.application.services.plan_lifecycle_service import PlanLifecycleService
|
|
... from cleveragents.application.services.plan_executor import PlanExecutor
|
|
... from cleveragents.tool.registry import ToolRegistry
|
|
... from cleveragents.tool.runner import ToolRunner
|
|
... settings = Settings()
|
|
... lifecycle = PlanLifecycleService(settings=settings)
|
|
... runner = ToolRunner(registry=ToolRegistry())
|
|
... executor = PlanExecutor(lifecycle_service=lifecycle, tool_runner=runner)
|
|
... action = lifecycle.create_action(
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}name="local/robot-test",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}description="Robot test action",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}definition_of_done="Write tests" + chr(10) + "Run linter",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}strategy_actor="local/stub",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}execution_actor="local/stub",
|
|
... )
|
|
... plan = lifecycle.use_action(action_name=str(action.namespaced_name))
|
|
... plan_id = plan.identity.plan_id
|
|
... s_result = executor.run_strategize(plan_id)
|
|
... print("PASS:strategize_ok")
|
|
... lifecycle.execute_plan(plan_id)
|
|
... e_result = executor.run_execute(plan_id)
|
|
... print("PASS:execute_ok")
|
|
... plan = lifecycle.get_plan(plan_id)
|
|
... if plan.decision_root_id:
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}print("PASS:decision_root_set")
|
|
|
|
${STREAM_SCRIPT} SEPARATOR=\n
|
|
... import os, sys
|
|
... sys.path.insert(0, os.path.join(os.getcwd(), "src"))
|
|
... from cleveragents.application.services.plan_executor import StrategizeStubActor
|
|
... from ulid import ULID
|
|
... events = []
|
|
... def callback(event_type, data):
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}events.append(event_type)
|
|
... actor = StrategizeStubActor()
|
|
... actor.execute(plan_id=str(ULID()), definition_of_done="Test", stream_callback=callback)
|
|
... for e in events:
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}print(f"PASS:{e}")
|
|
|
|
${GUARD_SCRIPT} SEPARATOR=\n
|
|
... import os, sys
|
|
... sys.path.insert(0, os.path.join(os.getcwd(), "src"))
|
|
... from cleveragents.config.settings import Settings
|
|
... from cleveragents.application.services.plan_lifecycle_service import PlanLifecycleService
|
|
... from cleveragents.application.services.plan_executor import PlanExecutor
|
|
... from cleveragents.tool.registry import ToolRegistry
|
|
... from cleveragents.tool.runner import ToolRunner
|
|
... from cleveragents.core.exceptions import PlanError
|
|
... settings = Settings()
|
|
... lifecycle = PlanLifecycleService(settings=settings)
|
|
... runner = ToolRunner(registry=ToolRegistry())
|
|
... executor = PlanExecutor(lifecycle_service=lifecycle, tool_runner=runner)
|
|
... action = lifecycle.create_action(
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}name="local/guard-test",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}description="Guard test",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}definition_of_done="Test",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}strategy_actor="local/stub",
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}execution_actor="local/stub",
|
|
... )
|
|
... plan = lifecycle.use_action(action_name=str(action.namespaced_name))
|
|
... try:
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}executor.run_execute(plan.identity.plan_id)
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}print("FAIL:guard_not_triggered")
|
|
... except PlanError:
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}print("PASS:guard_blocked")
|