feat: add in-process TestClient for fast unit testing#1344
Open
feat: add in-process TestClient for fast unit testing#1344
Conversation
Closes #507 Adds `robyn.testing.TestClient` that executes route handlers directly without starting a server. Replicates the full request pipeline from server.rs: before middlewares → handler → global response headers → after middlewares. No Rust changes needed — the wrapped handlers stored in FunctionInfo already do DI, pydantic validation, and response formatting via closures. Made-with: Cursor
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughWalkthroughA new in-process TestClient and TestResponse were added in Changes
Sequence DiagramsequenceDiagram
participant Client as TestClient
participant App as Robyn App
participant Router as Router/Registry
participant Middleware as Middlewares
participant Handler as Route Handler
participant TestResp as TestResponse
Client->>App: initialize(app)
Client->>Client: build route & middleware registries
Client->>Client: client.get/post(...)/request(...)
Client->>Client: construct Request (path, query, headers, body, files)
Client->>Middleware: run global BEFORE middlewares
Middleware-->>Client: continue or short-circuit Response
Client->>Router: match route (path, method) -> route, path params
Router->>Middleware: run route BEFORE middleware (with params)
Middleware-->>Client: continue or short-circuit Response
Client->>Handler: call handler(Request, **params)
Handler-->>Client: return Response or raw result
Client->>Client: normalize result -> Robyn Response
Client->>Middleware: run global AFTER middlewares
Middleware-->>Client: modify Response
Client->>Middleware: run route AFTER middleware
Middleware-->>Client: final Response
Client->>TestResp: convert Robyn Response -> TestResponse
Client-->>Caller: return TestResponse (status_code, headers, text, json(), content, ok)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
for more information, see https://pre-commit.ci
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
robyn/testing.py (2)
66-67: Unused regex constants.
_PARAM_REand_CATCHALL_REare defined but never used; the_compile_route_patternfunction usesstartswith(":")andstartswith("*")directly. Consider removing these dead constants.🧹 Proposed fix
-_PARAM_RE = re.compile(r":([^/]+)") -_CATCHALL_RE = re.compile(r"\*([^/]*)") - - def _compile_route_pattern(route: str) -> re.Pattern:🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@robyn/testing.py` around lines 66 - 67, _PARAM_RE and _CATCHALL_RE are defined but never used; update the code to remove these dead constants (or alternatively use them inside _compile_route_pattern) to eliminate unused definitions: locate the constants _PARAM_RE and _CATCHALL_RE and delete their declarations, and then confirm _compile_route_pattern still functions (it currently uses startswith(":")/startswith("*")); if you prefer to use the regexes instead, modify _compile_route_pattern to apply _PARAM_RE and _CATCHALL_RE where appropriate and remove the manual startswith checks, ensuring all tests pass.
309-331: Consider extracting duplicated JSON handling logic.The
json_datahandling pattern is repeated identically acrosspost,put,patch, anddelete. A small helper could reduce duplication.♻️ Optional refactor
+ def _prepare_json_kwargs(self, json_data: Any, kw: dict) -> dict: + if json_data is not None and "body" not in kw: + kw["body"] = json.dumps(json_data) + kw.setdefault("headers", {})["Content-Type"] = "application/json" + return kw + def post(self, path: str, json_data: Any = None, **kw) -> TestResponse: - if json_data is not None and "body" not in kw: - kw["body"] = json.dumps(json_data) - kw.setdefault("headers", {})["Content-Type"] = "application/json" + kw = self._prepare_json_kwargs(json_data, kw) return self._execute("POST", path, **kw)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@robyn/testing.py` around lines 309 - 331, Extract the repeated json_data handling into a small helper (e.g., _attach_json_body or _prepare_request_body) that takes json_data and the kw dict, sets kw["body"] = json.dumps(json_data) when body not present, and ensures kw.setdefault("headers", {})["Content-Type"] = "application/json"; then replace the duplicated blocks in post, put, patch, and delete to call that helper before returning self._execute(...). Keep the helper private and reuse it in the methods to remove duplication while preserving the existing behavior and signatures.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@robyn/testing.py`:
- Around line 224-238: The before-middleware handling assigns result to request
even when a middleware returns None, causing AttributeError later; update both
loops that call self._call (the self._global_before loop and the route-specific
block using self._mw_before.match/mw_fn) to treat a None result as a no-op
(i.e., leave request unchanged) while still short-circuiting and returning if
result is a Response (using _to_test_response); ensure you reference and modify
the branches around _call, the isinstance(result, Response) checks, and the
request = result assignments so None does not replace the request object.
- Around line 130-132: The constructor for the TestClient-like object currently
creates a dedicated loop with __init__ using asyncio.new_event_loop(), which
will conflict in environments with an existing running loop (pytest-asyncio);
change __init__ to obtain the loop via asyncio.get_event_loop() with a fallback
to new_event_loop only when no loop exists, and store a flag indicating
ownership. Update close() to only close the loop if this instance owns it (to
avoid closing a global running loop). In _call(), detect whether an event loop
is already running (asyncio.get_running_loop() or using the ownership flag) and
choose between self._loop.run_until_complete(...) when not running vs awaiting
the coroutine directly when inside a running loop. Ensure TestClient consumers
still work when used both synchronously and from async test functions.
- Around line 203-211: The current return uses "body_val or """, which converts
an explicit b"" into "" because b"" is falsy; change the call to pass the
computed body_val directly (or conditionally use "" only when the original body
was None) so that Request(...) receives the correct bytes type; inspect the
body/body_val variables and the Request(...) invocation to replace "body_val or
''" with simply "body_val" or an explicit None-check.
---
Nitpick comments:
In `@robyn/testing.py`:
- Around line 66-67: _PARAM_RE and _CATCHALL_RE are defined but never used;
update the code to remove these dead constants (or alternatively use them inside
_compile_route_pattern) to eliminate unused definitions: locate the constants
_PARAM_RE and _CATCHALL_RE and delete their declarations, and then confirm
_compile_route_pattern still functions (it currently uses
startswith(":")/startswith("*")); if you prefer to use the regexes instead,
modify _compile_route_pattern to apply _PARAM_RE and _CATCHALL_RE where
appropriate and remove the manual startswith checks, ensuring all tests pass.
- Around line 309-331: Extract the repeated json_data handling into a small
helper (e.g., _attach_json_body or _prepare_request_body) that takes json_data
and the kw dict, sets kw["body"] = json.dumps(json_data) when body not present,
and ensures kw.setdefault("headers", {})["Content-Type"] = "application/json";
then replace the duplicated blocks in post, put, patch, and delete to call that
helper before returning self._execute(...). Keep the helper private and reuse it
in the methods to remove duplication while preserving the existing behavior and
signatures.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 248685ad-5d08-4892-841b-d3bcdb2180b5
📒 Files selected for processing (2)
robyn/__init__.pyrobyn/testing.py
Comment on lines
+130
to
+132
| def __init__(self, app) -> None: | ||
| self.app = app | ||
| self._loop = asyncio.new_event_loop() |
There was a problem hiding this comment.
Event loop may conflict with async test frameworks.
Creating a dedicated event loop via asyncio.new_event_loop() will fail or behave unexpectedly when TestClient is instantiated or used within an already-running async context (e.g., pytest-asyncio tests). Consider using asyncio.get_event_loop() with fallback, or detecting the running loop scenario.
💡 Possible approach
def __init__(self, app) -> None:
self.app = app
- self._loop = asyncio.new_event_loop()
+ try:
+ self._loop = asyncio.get_running_loop()
+ self._owns_loop = False
+ except RuntimeError:
+ self._loop = asyncio.new_event_loop()
+ self._owns_loop = TrueThen in close():
def close(self) -> None:
- if not self._loop.is_closed():
+ if self._owns_loop and not self._loop.is_closed():
self._loop.close()And in _call(), check for running loop to decide between run_until_complete vs direct await.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@robyn/testing.py` around lines 130 - 132, The constructor for the
TestClient-like object currently creates a dedicated loop with __init__ using
asyncio.new_event_loop(), which will conflict in environments with an existing
running loop (pytest-asyncio); change __init__ to obtain the loop via
asyncio.get_event_loop() with a fallback to new_event_loop only when no loop
exists, and store a flag indicating ownership. Update close() to only close the
loop if this instance owns it (to avoid closing a global running loop). In
_call(), detect whether an event loop is already running
(asyncio.get_running_loop() or using the ownership flag) and choose between
self._loop.run_until_complete(...) when not running vs awaiting the coroutine
directly when inside a running loop. Ensure TestClient consumers still work when
used both synchronously and from async test functions.
Comment on lines
+203
to
+211
| body_val: Union[str, bytes] = "" | ||
| if body is not None: | ||
| body_val = body if isinstance(body, bytes) else body.encode("utf-8") | ||
|
|
||
| return Request( | ||
| qp, | ||
| h, | ||
| {}, | ||
| body_val or "", |
There was a problem hiding this comment.
Empty bytes body incorrectly coerced to empty string.
When body is explicitly set to b"", the expression body_val or "" evaluates to "" because b"" is falsy. This silently changes the body from bytes to string, potentially causing type inconsistencies.
🐛 Proposed fix
body_val: Union[str, bytes] = ""
if body is not None:
- body_val = body if isinstance(body, bytes) else body.encode("utf-8")
+ body_val = body if isinstance(body, bytes) else str(body).encode("utf-8")
return Request(
qp,
h,
{},
- body_val or "",
+ body_val if body_val else b"",
method,📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| body_val: Union[str, bytes] = "" | |
| if body is not None: | |
| body_val = body if isinstance(body, bytes) else body.encode("utf-8") | |
| return Request( | |
| qp, | |
| h, | |
| {}, | |
| body_val or "", | |
| body_val: Union[str, bytes] = "" | |
| if body is not None: | |
| body_val = body if isinstance(body, bytes) else str(body).encode("utf-8") | |
| return Request( | |
| qp, | |
| h, | |
| {}, | |
| body_val if body_val else b"", | |
| method, |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@robyn/testing.py` around lines 203 - 211, The current return uses "body_val
or """, which converts an explicit b"" into "" because b"" is falsy; change the
call to pass the computed body_val directly (or conditionally use "" only when
the original body was None) so that Request(...) receives the correct bytes
type; inspect the body/body_val variables and the Request(...) invocation to
replace "body_val or ''" with simply "body_val" or an explicit None-check.
There was a problem hiding this comment.
Actionable comments posted: 3
♻️ Duplicate comments (3)
robyn/testing.py (3)
223-227:⚠️ Potential issue | 🟠 MajorTreat
Nonefrom before middleware as a no-op.In-place before middlewares typically return nothing. Both branches currently assign that
Noneback torequest, so the next middleware or handler can crash onrequest.path_params.🐛 Proposed fix
for mw_fn in self._global_before: result = self._call(mw_fn, request) if isinstance(result, Response): return self._to_test_response(result) - request = result + if result is not None: + request = result ... if mw_fn is not None: request.path_params = mw_params result = self._call(mw_fn, request) if isinstance(result, Response): return self._to_test_response(result) - request = result + if result is not None: + request = resultAlso applies to: 231-236
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@robyn/testing.py` around lines 223 - 227, The before-middleware loop in testing.py assigns None results back to request causing downstream attribute errors; update the loop that iterates over self._global_before (and the similar loop at lines 231-236) so that after calling self._call(mw_fn, request) you only return if isinstance(result, Response) (using self._to_test_response), and only overwrite request when result is not None (i.e., if result is not None: request = result), treating a None return from the middleware as a no-op.
201-209:⚠️ Potential issue | 🟡 MinorPreserve explicit empty request bodies.
body_val or ""turns an explicitb""/""into"". That mutates the request body type/value before it reaches the handler. Only fall back to the default whenbody is None.🐛 Proposed fix
body_val: Union[str, bytes] = "" if body is not None: body_val = body if isinstance(body, bytes) else body.encode("utf-8") return Request( qp, h, {}, - body_val or "", + body_val if body is not None else "", method,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@robyn/testing.py` around lines 201 - 209, The code currently forces an explicit empty body to become "" via "body_val or """, losing an explicit b""/""; change the Request call to pass the actual encoded value when the original body was provided and only fall back to the default when body is None—e.g., replace the "body_val or """ usage with a conditional that uses body_val if body is not None (preserving empty bytes/strings) otherwise "" so the Request receives the true body value; refer to the body, body_val variables and the Request(...) call to locate where to update.
130-132:⚠️ Potential issue | 🟠 Major
TestClientstill breaks under async test runners.
client.get()eventually drives async handlers viaself._loop.run_until_complete(...). If the caller is already inside a running loop (pytest-asyncio,anyio, notebooks), this raisesRuntimeErrorand makes async handlers/middlewares unusable from async tests. This needs either a background loop/thread or async request methods instead of re-entering the loop from sync code.Also applies to: 177-181
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@robyn/testing.py` around lines 130 - 132, TestClient's synchronous use of self._loop.run_until_complete causes RuntimeError when the caller already has a running event loop; instead spin up a dedicated background event loop thread in TestClient.__init__ (create self._loop = asyncio.new_event_loop(), start a daemon thread that calls self._loop.run_forever) and replace all run_until_complete usages (e.g., in client.get() and the other request methods around the 177-181 block) with asyncio.run_coroutine_threadsafe(coro, self._loop).result() to execute coroutines on that background loop; also add proper cleanup to stop the loop and join the thread on client shutdown.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@robyn/testing.py`:
- Around line 259-266: The merge assumes response.headers supports .append() but
response.headers can be a dict; update the merge logic in the blocks handling
global response headers (where response.headers is accessed) to
normalize/convert response.headers to a Headers-like object before appending
(e.g., wrap dict into the same Headers implementation used elsewhere or build a
MutableHeaders-equivalent and assign back to response.headers), then iterate
self.app.response_headers.get_headers() and append values; ensure you handle
both dict and Headers types and apply the same change to the other occurrence
mentioned (around the block at lines ~296-299) and reference
self.app.response_headers, response.headers, excluded_response_headers_paths and
response_headers.get_headers when making the change.
- Around line 155-161: The route-specific middleware keys are built as
f"{_method_str(mw.route_type)}{mw.route}" (e.g., "GET/users/1") but
_compile_route_pattern() produces regexes with a leading "/" (e.g., "/users/1"),
so lookups never match; fix by normalizing the combined key to include the
leading slash (or strip the leading slash from patterns) when populating
_mw_before/_mw_after in the loop inside testing.py: change combined to
f"{_method_str(mw.route_type)}/{mw.route.lstrip('/')}" (or otherwise ensure the
same leading-slash convention used by _compile_route_pattern), leaving the rest
of the add(...) logic unchanged so route-specific before/after middlewares
resolve correctly.
- Around line 289-295: _in _to_test_response_, response.description is being
converted to bytes via str(), which serializes dicts/lists with Python repr
(single quotes) causing TestResponse.json() to fail; change the conversion to
detect JSON-serializable structures (e.g., isinstance(response.description,
(dict, list))) and call json.dumps(response.description).encode("utf-8") for
those cases (import json if missing), otherwise keep the existing string/bytes
handling so TestResponse.text/json match real server output.
---
Duplicate comments:
In `@robyn/testing.py`:
- Around line 223-227: The before-middleware loop in testing.py assigns None
results back to request causing downstream attribute errors; update the loop
that iterates over self._global_before (and the similar loop at lines 231-236)
so that after calling self._call(mw_fn, request) you only return if
isinstance(result, Response) (using self._to_test_response), and only overwrite
request when result is not None (i.e., if result is not None: request = result),
treating a None return from the middleware as a no-op.
- Around line 201-209: The code currently forces an explicit empty body to
become "" via "body_val or """, losing an explicit b""/""; change the Request
call to pass the actual encoded value when the original body was provided and
only fall back to the default when body is None—e.g., replace the "body_val or
""" usage with a conditional that uses body_val if body is not None (preserving
empty bytes/strings) otherwise "" so the Request receives the true body value;
refer to the body, body_val variables and the Request(...) call to locate where
to update.
- Around line 130-132: TestClient's synchronous use of
self._loop.run_until_complete causes RuntimeError when the caller already has a
running event loop; instead spin up a dedicated background event loop thread in
TestClient.__init__ (create self._loop = asyncio.new_event_loop(), start a
daemon thread that calls self._loop.run_forever) and replace all
run_until_complete usages (e.g., in client.get() and the other request methods
around the 177-181 block) with asyncio.run_coroutine_threadsafe(coro,
self._loop).result() to execute coroutines on that background loop; also add
proper cleanup to stop the loop and join the thread on client shutdown.
- Guard against before-middleware returning None (AttributeError) - Fix route-specific middleware key format so patterns actually match - Serialize dict/list response bodies with json.dumps instead of str() - Normalize response.headers to Headers before calling .append() Made-with: Cursor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
robyn/testing.py (2)
313-335: Consider extracting JSON body preparation to reduce duplication.The
json_datahandling logic is repeated identically inpost,put,patch, anddelete. A small helper would improve maintainability.♻️ Example refactor
+ def _prepare_json(self, json_data: Any, kw: dict) -> dict: + if json_data is not None and "body" not in kw: + kw["body"] = json.dumps(json_data) + kw.setdefault("headers", {})["Content-Type"] = "application/json" + return kw + def post(self, path: str, json_data: Any = None, **kw) -> TestResponse: - if json_data is not None and "body" not in kw: - kw["body"] = json.dumps(json_data) - kw.setdefault("headers", {})["Content-Type"] = "application/json" - return self._execute("POST", path, **kw) + return self._execute("POST", path, **self._prepare_json(json_data, kw))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@robyn/testing.py` around lines 313 - 335, Extract the repeated JSON body preparation into a private helper (e.g., _prepare_json_body or _attach_json_body) and call it from post, put, patch, and delete before invoking _execute; the helper should accept (kw, json_data), check if json_data is not None and "body" not in kw, set kw["body"] = json.dumps(json_data) and ensure kw.setdefault("headers", {})["Content-Type"] = "application/json", leaving _execute and the public methods otherwise unchanged.
66-67: Remove unused regex patterns.
_PARAM_REand_CATCHALL_REare defined but never used. The_compile_route_patternfunction usesstartswithchecks instead.🧹 Proposed fix
-_PARAM_RE = re.compile(r":([^/]+)") -_CATCHALL_RE = re.compile(r"\*([^/]*)") - - def _compile_route_pattern(route: str) -> re.Pattern:🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@robyn/testing.py` around lines 66 - 67, Remove the dead regex globals _PARAM_RE and _CATCHALL_RE from the module since they are defined but never used; update the module to delete these symbols and any references (none expected) and remove the re import if it becomes unused, leaving _compile_route_pattern as the canonical logic that uses startswith checks.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@robyn/testing.py`:
- Around line 261-270: The code only converts response.headers dict to a Headers
instance when merging global headers, causing TestResponse.headers to sometimes
remain a dict; always normalize response.headers to a Headers object before any
conditional logic. Change the block around response.headers so that you call
isinstance(response.headers, dict) and set response.headers =
Headers(response.headers) (or equivalent normalization) immediately after
obtaining response, then proceed to check
self.app.excluded_response_headers_paths, self.app.response_headers.is_empty(),
and perform appending via resp_headers.append; apply the same normalization fix
to the other similar block (the one around the TestResponse.headers assignment)
so TestResponse.headers always receives a Headers instance.
---
Nitpick comments:
In `@robyn/testing.py`:
- Around line 313-335: Extract the repeated JSON body preparation into a private
helper (e.g., _prepare_json_body or _attach_json_body) and call it from post,
put, patch, and delete before invoking _execute; the helper should accept (kw,
json_data), check if json_data is not None and "body" not in kw, set kw["body"]
= json.dumps(json_data) and ensure kw.setdefault("headers", {})["Content-Type"]
= "application/json", leaving _execute and the public methods otherwise
unchanged.
- Around line 66-67: Remove the dead regex globals _PARAM_RE and _CATCHALL_RE
from the module since they are defined but never used; update the module to
delete these symbols and any references (none expected) and remove the re import
if it becomes unused, leaving _compile_route_pattern as the canonical logic that
uses startswith checks.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
Comment on lines
+261
to
+270
| # ---- merge global response headers -------------------------------- | ||
| excluded = self.app.excluded_response_headers_paths or [] | ||
| if path not in excluded and not self.app.response_headers.is_empty(): | ||
| if isinstance(response.headers, dict): | ||
| response.headers = Headers(response.headers) | ||
| resp_headers = response.headers | ||
| global_dict = self.app.response_headers.get_headers() | ||
| for key, values in global_dict.items(): | ||
| for val in values: | ||
| resp_headers.append(key, val) |
There was a problem hiding this comment.
TestResponse.headers may receive a dict instead of Headers.
The normalization at lines 264-265 only runs when global headers are being merged. If the path is excluded or no global headers exist, response.headers remains a dict and flows directly to TestResponse.headers (line 304), violating the Headers type hint.
🐛 Proposed fix
Move normalization before the conditional:
+ if isinstance(response.headers, dict):
+ response.headers = Headers(response.headers)
+
# ---- merge global response headers --------------------------------
excluded = self.app.excluded_response_headers_paths or []
if path not in excluded and not self.app.response_headers.is_empty():
- if isinstance(response.headers, dict):
- response.headers = Headers(response.headers)
resp_headers = response.headers
global_dict = self.app.response_headers.get_headers()Also applies to: 302-306
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@robyn/testing.py` around lines 261 - 270, The code only converts
response.headers dict to a Headers instance when merging global headers, causing
TestResponse.headers to sometimes remain a dict; always normalize
response.headers to a Headers object before any conditional logic. Change the
block around response.headers so that you call isinstance(response.headers,
dict) and set response.headers = Headers(response.headers) (or equivalent
normalization) immediately after obtaining response, then proceed to check
self.app.excluded_response_headers_paths, self.app.response_headers.is_empty(),
and perform appending via resp_headers.append; apply the same normalization fix
to the other similar block (the one around the TestResponse.headers assignment)
so TestResponse.headers always receives a Headers instance.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #507
Summary
Adds
robyn.testing.TestClient— an in-process test client that executes route handlers directly without starting a server, subprocess, or socket.robyn/testing.py(~220 LOC) withTestClientandTestResponseTestClienttorobyn/__init__.pyand__all__FunctionInfoalready handle DI, pydantic validation, and response formatting via closuresHow it works
The
TestClientreplicates the full request pipeline fromsrc/server.rs::index():Requestfrom test call argsResponse)METHOD/path)fn_info.handler(request)— returns a fully formattedResponseTestResponseUsage
What's covered
:idpatterns)with TestClient(app) as client:)Explicitly deferred to follow-up PRs
serve_directory— Actix-only)startup_handler/shutdown_handlertriggersTest plan
integration_tests/base_routes.pyroutesMade with Cursor
Summary by CodeRabbit