forked from CopilotKit/CopilotKit
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconftest.py
More file actions
116 lines (87 loc) · 3.85 KB
/
Copy pathconftest.py
File metadata and controls
116 lines (87 loc) · 3.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""Pytest configuration for strands showcase unit tests.
Ensures ``src/`` (so ``agents.agent`` imports) is on ``sys.path``.
Shared tools are accessible via the ``tools`` symlink at the project root.
Also installs minimal stubs for ``ag_ui_strands`` and ``strands`` so the
unit tests can run in environments where those heavy runtime deps aren't
installed. Tests that need real behavior monkey-patch the specific
symbols they touch.
"""
import os
import sys
import types
_HERE = os.path.dirname(__file__)
_PKG_ROOT = os.path.abspath(os.path.join(_HERE, "..", ".."))
# src/ holds agent_server.py and agents/
sys.path.insert(0, os.path.join(_PKG_ROOT, "src"))
# tools/ symlink at project root points to shared/python/tools
sys.path.insert(0, _PKG_ROOT)
class _Permissive:
"""Base stub that accepts any ``__init__`` args and exposes a writable
``_agents_by_thread`` attribute so ``build_showcase_agent`` can swap
the per-thread dict in place."""
def __init__(self, *args, **kwargs):
self.args = args
self.kwargs = kwargs
self._agents_by_thread: dict = {}
def _install_stub_modules() -> None:
"""Install minimal stub modules so ``agents.agent`` can be imported.
Unit tests only exercise pure-Python logic (cap hook counter, dict
injection). We stub out strands / ag_ui_strands symbols with plain
placeholder classes; tests that care about behavior monkey-patch the
relevant attributes.
"""
if "ag_ui_strands" not in sys.modules:
m = types.ModuleType("ag_ui_strands")
m.StrandsAgent = _Permissive # type: ignore[attr-defined]
m.StrandsAgentConfig = _Permissive # type: ignore[attr-defined]
m.ToolBehavior = _Permissive # type: ignore[attr-defined]
class _FakeFastAPI:
"""Accepts the decorators agent_server applies (``@app.get`` etc.)."""
def _decorator(self, *a, **k):
def _wrap(fn):
return fn
return _wrap
get = post = put = delete = patch = _decorator
# agent_server.py also calls ``app.add_middleware(...)`` to
# install HealthMiddleware; accept that on the stub too.
def add_middleware(self, *a, **k):
return None
m.create_strands_app = lambda *a, **k: _FakeFastAPI() # type: ignore[attr-defined]
sys.modules["ag_ui_strands"] = m
if "strands" not in sys.modules:
m = types.ModuleType("strands")
m.Agent = _Permissive # type: ignore[attr-defined]
def _tool_decorator(func=None, **_kwargs):
if callable(func):
return func
def _wrap(f):
return f
return _wrap
m.tool = _tool_decorator # type: ignore[attr-defined]
sys.modules["strands"] = m
if "strands.hooks" not in sys.modules:
m = types.ModuleType("strands.hooks")
for name in (
"AfterToolCallEvent",
"BeforeInvocationEvent",
"BeforeToolCallEvent",
"HookProvider",
"HookRegistry",
):
setattr(m, name, type(name, (), {}))
sys.modules["strands.hooks"] = m
if "strands.models" not in sys.modules:
sys.modules["strands.models"] = types.ModuleType("strands.models")
if "strands.models.openai" not in sys.modules:
m = types.ModuleType("strands.models.openai")
m.OpenAIModel = _Permissive # type: ignore[attr-defined]
sys.modules["strands.models.openai"] = m
if "uvicorn" not in sys.modules:
m = types.ModuleType("uvicorn")
m.run = lambda *a, **k: None # type: ignore[attr-defined]
sys.modules["uvicorn"] = m
if "dotenv" not in sys.modules:
m = types.ModuleType("dotenv")
m.load_dotenv = lambda *a, **k: None # type: ignore[attr-defined]
sys.modules["dotenv"] = m
_install_stub_modules()