khyari hamza
5d5d3fde2c
Add AsyncResourceTracker (core/async_cleanup.py) providing a central registry for async resources with timeout-bounded close_all(), async context manager support, and a __del__ finalizer that logs leaked resources by name. Enhance LangGraphBridge with cleanup_tasks_async() that awaits in-flight tasks with a deadline instead of fire-and-forget cancel(). Add cancellation_reasons dict to trace why tasks were cancelled. Add StateManager.close() to properly release checkpoint file handles and complete the RxPY BehaviorSubject. Add AcpEventQueue.close() to dispose all subscriptions. Includes 14 Behave scenarios (67 steps), Robot integration tests, ASV benchmarks, and docs/reference/async_safety.md. ISSUES CLOSED: #321
136 lines
6.2 KiB
Plaintext
136 lines
6.2 KiB
Plaintext
*** Settings ***
|
|
Documentation Integration smoke tests for async resource cleanup (#321).
|
|
... Validates that AsyncResourceTracker, enhanced bridge cleanup,
|
|
... AcpEventQueue.close(), and StateManager.close() work end-to-end
|
|
... by executing small Python driver scripts in a subprocess.
|
|
|
|
Library OperatingSystem
|
|
Library Process
|
|
Library String
|
|
Resource ${CURDIR}/common.resource
|
|
|
|
Test Setup Setup Async Cleanup Test Environment
|
|
Test Teardown Cleanup Async Cleanup Test Environment
|
|
|
|
*** Variables ***
|
|
${PYTHON} python
|
|
${WORKSPACE_ROOT} ${CURDIR}/..
|
|
${TEST_FILE} ${EMPTY}
|
|
${TEST_OUTPUT} ${EMPTY}
|
|
${TIMEOUT} 30s
|
|
|
|
*** Test Cases ***
|
|
Test AsyncResourceTracker Register And Close
|
|
[Documentation] Register a resource and close it via close_all
|
|
${script} = Catenate SEPARATOR=\n
|
|
... import sys, asyncio
|
|
... sys.path.insert(0, '${WORKSPACE_ROOT}/src')
|
|
... from cleveragents.core.async_cleanup import AsyncResourceTracker
|
|
...
|
|
... class FakeResource:
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}def __init__(self): self.closed = False
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}async def close(self): self.closed = True
|
|
...
|
|
... async def main():
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}tracker = AsyncResourceTracker()
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}res = FakeResource()
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}tracker.register("test-res", res)
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}assert tracker.open_count == 1
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}await tracker.close_all()
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}assert res.closed
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}assert tracker.open_count == 0
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}print("PASS: register and close", flush=True)
|
|
...
|
|
... asyncio.run(main())
|
|
Create File ${TEST_FILE} ${script}
|
|
${result} = Run Process ${PYTHON} ${TEST_FILE}
|
|
... stdout=${TEST_OUTPUT} stderr=STDOUT timeout=${TIMEOUT}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
${output} = Get File ${TEST_OUTPUT}
|
|
Should Contain ${output} PASS: register and close
|
|
|
|
Test AsyncResourceTracker Timeout Warning
|
|
[Documentation] Verify forced-termination warning for slow resource
|
|
${script} = Catenate SEPARATOR=\n
|
|
... import sys, asyncio, logging
|
|
... sys.path.insert(0, '${WORKSPACE_ROOT}/src')
|
|
... from cleveragents.core.async_cleanup import AsyncResourceTracker
|
|
...
|
|
... handler = logging.StreamHandler(sys.stdout)
|
|
... handler.setLevel(logging.WARNING)
|
|
... logging.getLogger("cleveragents.core.async_cleanup").addHandler(handler)
|
|
...
|
|
... class SlowResource:
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}async def close(self): await asyncio.sleep(10)
|
|
...
|
|
... async def main():
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}tracker = AsyncResourceTracker()
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}tracker.register("slow", SlowResource())
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}await tracker.close_all(timeout=0.05)
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}assert len(tracker.timed_out_resources) == 1
|
|
... ${SPACE}${SPACE}${SPACE}${SPACE}print("PASS: timeout warning", flush=True)
|
|
...
|
|
... asyncio.run(main())
|
|
Create File ${TEST_FILE} ${script}
|
|
${result} = Run Process ${PYTHON} ${TEST_FILE}
|
|
... stdout=${TEST_OUTPUT} stderr=STDOUT timeout=${TIMEOUT}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
${output} = Get File ${TEST_OUTPUT}
|
|
Should Contain ${output} PASS: timeout warning
|
|
|
|
Test AcpEventQueue Close
|
|
[Documentation] Verify AcpEventQueue.close() removes subscriptions
|
|
${script} = Catenate SEPARATOR=\n
|
|
... import sys
|
|
... sys.path.insert(0, '${WORKSPACE_ROOT}/src')
|
|
... from cleveragents.acp.events import AcpEventQueue
|
|
...
|
|
... q = AcpEventQueue()
|
|
... q.subscribe_local(lambda e: None)
|
|
... q.subscribe_local(lambda e: None)
|
|
... assert len(q._subscriptions) == 2
|
|
... q.close()
|
|
... assert len(q._subscriptions) == 0
|
|
... print("PASS: event queue close", flush=True)
|
|
Create File ${TEST_FILE} ${script}
|
|
${result} = Run Process ${PYTHON} ${TEST_FILE}
|
|
... stdout=${TEST_OUTPUT} stderr=STDOUT timeout=${TIMEOUT}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
${output} = Get File ${TEST_OUTPUT}
|
|
Should Contain ${output} PASS: event queue close
|
|
|
|
Test StateManager Close
|
|
[Documentation] Verify StateManager.close() marks manager as closed
|
|
${script} = Catenate SEPARATOR=\n
|
|
... import sys, tempfile
|
|
... sys.path.insert(0, '${WORKSPACE_ROOT}/src')
|
|
... from pathlib import Path
|
|
... from cleveragents.langgraph.state import StateManager, GraphState
|
|
...
|
|
... d = tempfile.mkdtemp()
|
|
... mgr = StateManager(initial_state=GraphState(), checkpoint_dir=Path(d))
|
|
... assert not mgr.is_closed
|
|
... mgr.close()
|
|
... assert mgr.is_closed
|
|
... print("PASS: state manager close", flush=True)
|
|
Create File ${TEST_FILE} ${script}
|
|
${result} = Run Process ${PYTHON} ${TEST_FILE}
|
|
... stdout=${TEST_OUTPUT} stderr=STDOUT timeout=${TIMEOUT}
|
|
Should Be Equal As Integers ${result.rc} 0
|
|
${output} = Get File ${TEST_OUTPUT}
|
|
Should Contain ${output} PASS: state manager close
|
|
|
|
*** Keywords ***
|
|
Setup Async Cleanup Test Environment
|
|
[Documentation] Create temp dir for test scripts
|
|
${temp_dir} = Evaluate tempfile.mkdtemp() modules=tempfile
|
|
Set Test Variable ${TEMP_DIR} ${temp_dir}
|
|
Set Test Variable ${TEST_FILE} ${temp_dir}/async_cleanup_test.py
|
|
Set Test Variable ${TEST_OUTPUT} ${temp_dir}/async_cleanup_output.txt
|
|
|
|
Cleanup Async Cleanup Test Environment
|
|
[Documentation] Remove temp dir
|
|
Run Keyword And Ignore Error Remove File ${TEST_FILE}
|
|
Run Keyword And Ignore Error Remove File ${TEST_OUTPUT}
|
|
Run Keyword And Ignore Error Remove Directory ${TEMP_DIR} recursive=True
|