style(a2a): apply ruff format to integration test helper

Apply ruff format to robot/helper_a2a_session_plan_lifecycle_integration.py
to fix CI lint failure. Changes are purely formatting (whitespace, line
wrapping) with no logic modifications.

ISSUES CLOSED: #10032
This commit is contained in:
2026-04-22 23:00:56 +00:00
committed by Forgejo
parent d306e6e33c
commit 05b58630bf
@@ -146,9 +146,7 @@ def _session_full_lifecycle() -> None:
sessions = session_svc.list()
assert len(sessions) >= 1, f"Expected at least 1 session, got {len(sessions)}"
session_ids = [s.session_id for s in sessions]
assert session_id in session_ids, (
f"Created session {session_id} not in list"
)
assert session_id in session_ids, f"Created session {session_id} not in list"
# 3. Show (get) session by ID
fetched = session_svc.get(session_id)
@@ -278,26 +276,30 @@ def _event_queue_receives_events() -> None:
assert sub_id, "Expected a subscription ID"
# Publish events simulating plan lifecycle transitions
event_queue.publish(A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "strategize", "state": "queued"},
))
event_queue.publish(A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "execute", "state": "processing"},
))
event_queue.publish(A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "apply", "state": "applied"},
))
event_queue.publish(
A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "strategize", "state": "queued"},
)
)
event_queue.publish(
A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "execute", "state": "processing"},
)
)
event_queue.publish(
A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "apply", "state": "applied"},
)
)
# Verify subscriber received all events
assert len(received_events) == 3, (
f"Expected 3 events, got {len(received_events)}"
)
assert len(received_events) == 3, f"Expected 3 events, got {len(received_events)}"
assert received_events[0].event_type == "TaskStatusUpdateEvent"
assert received_events[0].plan_id == "test-plan-001"
assert received_events[0].data["phase"] == "strategize"
@@ -312,11 +314,13 @@ def _event_queue_receives_events() -> None:
removed = event_queue.unsubscribe(sub_id)
assert removed is True
event_queue.publish(A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "done"},
))
event_queue.publish(
A2aEvent(
event_type="TaskStatusUpdateEvent",
plan_id="test-plan-001",
data={"phase": "done"},
)
)
assert len(received_events) == 3, "Should not receive events after unsubscribe"
# Verify event queue integration with A2A facade
@@ -511,9 +515,7 @@ def _guard_denylist() -> None:
# Tool in allowlist should be allowed
result5 = profile_with_allowlist.check_guard(tool_name="read_file")
assert result5.allowed is True, (
"Expected 'read_file' to be allowed (in allowlist)"
)
assert result5.allowed is True, "Expected 'read_file' to be allowed (in allowlist)"
print("guard-denylist-ok")
@@ -565,9 +567,7 @@ def _plan_rollback_session_consistent() -> None:
assert failed_plan.is_errored
# ERRORED is terminal per spec (is_terminal returns True for ERRORED)
assert failed_plan.is_terminal is True, (
"ERRORED state should be terminal per spec"
)
assert failed_plan.is_terminal is True, "ERRORED state should be terminal per spec"
# Verify session is still accessible (consistent state)
session_still_exists = session_svc.get(session.session_id)