diff --git a/features/steps/tdd_actors_loader_lock_filter_steps.py b/features/steps/tdd_actors_loader_lock_filter_steps.py index fc8ec7d67..45905824d 100644 --- a/features/steps/tdd_actors_loader_lock_filter_steps.py +++ b/features/steps/tdd_actors_loader_lock_filter_steps.py @@ -9,20 +9,18 @@ from __future__ import annotations import tempfile import threading import time -from contextlib import ExitStack from pathlib import Path from typing import Any -import yaml from behave import given, then, when # type: ignore[attr-defined] from cleveragents.actor.loader import ActorLoader -from cleveragents.core.exceptions import ValidationError def _make_actor_yaml(name: str, provider: str = "openai", model: str = "gpt-4") -> str: return f"""name: {name} type: llm +description: Test actor {name} provider: {provider} model: {model} tools: [] @@ -55,9 +53,9 @@ def _setup_actor_dir() -> tuple[str, Path]: @given("a temporary directory with multiple namespace groups of actor YAML files") -def step_impl(context: Any) -> None: +def step_given_actor_dir(context: Any) -> None: """Set up temp directory with actors from two namespaces.""" - context._actor_tmp, context._actor_root = _setup_actor_dir() + context._actor_tmp, context._actor_root = _setup_actor_dir() # type: ignore[attr-defined] @given("the loader discovers all actors") @@ -68,11 +66,12 @@ def step_when_discover_all(context: Any) -> None: @given("I create an ActorLoader with initial actors from multiple namespaces") -def step_impl(context: Any) -> None: +@when("I create an ActorLoader with initial actors from multiple namespaces") +def step_given_loader_with_initial(context: Any) -> None: """Set up a loader and seed it with actors.""" loader = ActorLoader(search_roots=[context._actor_root]) # type: ignore[attr-defined] loader.discover() - context.actor_loader = loader + context.actor_loader = loader # type: ignore[attr-defined] # ------- When steps -------------------------------------------------------- @@ -83,15 +82,15 @@ def step_when_create_loader(context: Any) -> None: root: Path = context._actor_root # type: ignore[attr-defined] loader = ActorLoader(search_roots=[root]) loader.discover() - context.actor_loader = loader + context.actor_loader = loader # type: ignore[attr-defined] @when( "another thread concurrently adds and removes YAML files " - "(triggering ``discover()'')" + "(triggering ``discover()``)" ) def step_when_concurrent_modifications(context: Any) -> None: - """Start a background thread that modifies actor files.""" + """Define a background thread that mutates actor files.""" root: Path = context._actor_root # type: ignore[attr-defined] def _concurrent_worker() -> None: @@ -99,19 +98,20 @@ def step_when_concurrent_modifications(context: Any) -> None: try: for i in range(10): tmp_fpath = root / f"conc_{i}.yaml" - tmp_fpath.write_text( - _make_actor_yaml(f"conc/ns{i % 2}/concurrent_{i}") - ) + tmp_fpath.write_text(_make_actor_yaml(f"conc/concurrent_{i}")) loader = ActorLoader(search_roots=[root]) loader.discover() time.sleep(0.01) if tmp_fpath.exists(): tmp_fpath.unlink(missing_ok=True) - except Exception as exc: # noqa: BLE001 - context._concurrent_errors = getattr(context, "_concurrent_errors", []) # type: ignore[attr-defined] - context._concurrent_errors.append(exc) # type: ignore[attr-defined] + except Exception as exc: + errors = getattr(context, "_concurrent_errors", []) # type: ignore[attr-defined] + errors.append(exc) + context._concurrent_errors = errors # type: ignore[attr-defined] - context._worker_thread = threading.Thread(target=_concurrent_worker, daemon=True) # type: ignore[attr-defined] + context._worker_thread = threading.Thread( # type: ignore[attr-defined] + target=_concurrent_worker, daemon=True + ) @when("a worker repeatedly calls ``list_actors(namespace=...)`` in parallel") @@ -120,27 +120,18 @@ def step_when_parallel_list(context: Any) -> None: loader: ActorLoader = context.actor_loader # type: ignore[attr-defined] def _list_worker() -> None: - try: - for _ in range(20): - _ = loader.list_actors(namespace="local") - _ = loader.list_actors(namespace="remote") - time.sleep(0.01) - except Exception as exc: # noqa: BLE001 - context._list_errors = getattr(context, "_list_errors", []) # type: ignore[attr-defined] - context._list_errors.append(exc) # type: ignore[attr-defined] - - errors_list: list[Exception] = [] - - def _error_collector() -> None: try: for _ in range(20): loader.list_actors(namespace="local") + loader.list_actors(namespace="remote") time.sleep(0.01) - except Exception as exc: # noqa: BLE001 - errors_list.append(exc) + except Exception as exc: + errors = getattr(context, "_list_errors", []) # type: ignore[attr-defined] + errors.append(exc) + context._list_errors = errors # type: ignore[attr-defined] - workers = [] - start_thread = context._worker_thread if hasattr(context, "_worker_thread") else None # type: ignore[attr-defined] + workers: list[threading.Thread] = [] + start_thread = getattr(context, "_worker_thread", None) # type: ignore[attr-defined] for _ in range(5): t = threading.Thread(target=_list_worker, daemon=True) @@ -152,12 +143,13 @@ def step_when_parallel_list(context: Any) -> None: for t in workers: t.join(timeout=10) + if start_thread is not None: + start_thread.join(timeout=10) - # Check both error collectors - collected_errors = [*getattr(context, "_list_errors", [])] + errors_list # type: ignore[attr-defined] - if hasattr(context, "_concurrent_errors"): # type: ignore[attr-defined] - collected_errors.extend(context._concurrent_errors) # type: ignore[attr-defined] - + collected_errors: list[Exception] = [ + *getattr(context, "_list_errors", []), # type: ignore[attr-defined] + *getattr(context, "_concurrent_errors", []), # type: ignore[attr-defined] + ] context._final_concurrent_errors = collected_errors # type: ignore[attr-defined] @@ -167,7 +159,9 @@ def step_when_parallel_list(context: Any) -> None: @then('filtering by namespace "local" should return only local actors') def step_then_local_only(context: Any) -> None: result = context.actor_loader.list_actors(namespace="local") # type: ignore[attr-defined] - assert len(result) == 2, f"Expected 2 local actors, got {len(result)}: {[a.name for a in result]}" + assert len(result) == 2, ( + f"Expected 2 local actors, got {len(result)}: {[a.name for a in result]}" + ) names = {a.name for a in result} assert names == {"local/alpha", "local/beta"} @@ -175,12 +169,14 @@ def step_then_local_only(context: Any) -> None: @then('filtering by namespace "remote" should return only remote actors') def step_then_remote_only(context: Any) -> None: result = context.actor_loader.list_actors(namespace="remote") # type: ignore[attr-defined] - assert len(result) == 2, f"Expected 2 remote actors, got {len(result)}: {[a.name for a in result]}" + assert len(result) == 2, ( + f"Expected 2 remote actors, got {len(result)}: {[a.name for a in result]}" + ) names = {a.name for a in result} assert names == {"remote/gamma", "remote/delta"} -@then('calling ``list_actors()`` without a filter should return every actor') +@then("calling ``list_actors()`` without a filter should return every actor") def step_then_all_actors(context: Any) -> None: result = context.actor_loader.list_actors() # type: ignore[attr-defined] assert len(result) == 4, f"Expected 4 total actors, got {len(result)}" @@ -195,9 +191,6 @@ def step_then_no_errors(context: Any) -> None: @then("every returned result should be internally consistent") def step_then_consistent(context: Any) -> None: """Every thread-safe list_actors call returns a stable snapshot.""" - # If we reach here without an assertion error from "no exceptions", - # consistency is implied. Verify the namespace filter returned valid data - # by checking that all results share the expected structure. loader: ActorLoader = context.actor_loader # type: ignore[attr-defined] result = loader.list_actors(namespace="local") for actor in result: diff --git a/features/tdd_actors_loader_lock_filter.feature b/features/tdd_actors_loader_lock_filter.feature index 43b455768..6c7c63598 100644 --- a/features/tdd_actors_loader_lock_filter.feature +++ b/features/tdd_actors_loader_lock_filter.feature @@ -1,4 +1,4 @@ -@tdd_issue @issue_8660 +@tdd_issue @tdd_issue_8588 Feature: TDD Issue #8660 — ActorLoader.list_actors namespace filter inside lock As a developer diff --git a/tests/actor/test_loader_list_actors_thread_safety.py b/tests/actor/test_loader_list_actors_thread_safety.py index b0cd79314..862221300 100644 --- a/tests/actor/test_loader_list_actors_thread_safety.py +++ b/tests/actor/test_loader_list_actors_thread_safety.py @@ -13,14 +13,13 @@ import threading import time from pathlib import Path -import yaml - from cleveragents.actor.loader import ActorLoader def _make_actor_yaml(name: str, provider: str = "openai", model: str = "gpt-4") -> str: return f"""name: {name} type: llm +description: Test actor {name} provider: {provider} model: {model} tools: [] @@ -60,7 +59,9 @@ class TestListActorsThreadSafety: try: for i in range(20): extra_file = root / f"extra_{i}.yaml" - extra_file.write_text(_make_actor_yaml(f"conc/ns{i % 2}/thread_{i}")) + extra_file.write_text( + _make_actor_yaml(f"conc/thread_{i}") + ) loader.discover() time.sleep(0.01) if extra_file.exists(): @@ -115,9 +116,9 @@ class TestListActorsThreadSafety: def _list_worker() -> None: try: for _ in range(30): - all_actors = loader.list_actors() - local = loader.list_actors(namespace="local") - remote = loader.list_actors(namespace="remote") + loader.list_actors() + loader.list_actors(namespace="local") + loader.list_actors(namespace="remote") time.sleep(0.01) except Exception as exc: errors.append(exc) @@ -141,7 +142,9 @@ class TestListActorsThreadSafety: def _add_actor() -> None: try: for i in range(20): - (root / f"new_{i}.yaml").write_text(_make_actor_yaml(f"conc/new_{i}")) + (root / f"new_{i}.yaml").write_text( + _make_actor_yaml(f"conc/new_{i}") + ) loader.discover() time.sleep(0.01) if (root / f"new_{i}.yaml").exists(): @@ -172,4 +175,5 @@ class TestListActorsThreadSafety: if __name__ == "__main__": # pragma: no cover import pytest + pytest.main([__file__, "-v"])