test: add FastStream integration test#2070
Conversation
FastStream was the one documented integration with real span-emitting instrumentation but no test coverage. Exercise its RedisTelemetryMiddleware against the in-memory TestRedisBroker and assert the message spans and their OTel messaging attributes, mirroring the docs example. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
📝 WalkthroughWalkthroughAdded 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 3 files
Confidence score: 5/5
- In
tests/otel_integrations/test_faststream.py, thesorted({span['name'] for span in spans})set-based assertion can mask duplicate span names, so a regression that emits extra spans could still pass and reduce test sensitivity; this is low merge risk but worth tightening by asserting counts (or avoiding set deduplication) before/soon after merge.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="tests/otel_integrations/test_faststream.py">
<violation number="1" location="tests/otel_integrations/test_faststream.py:42">
P3: The test is well-constructed and matches FastStream's actual OTel middleware behavior. One subtle maintainability concern: the test uses `sorted({span['name'] for span in spans})` with a set comprehension (line 39), which deduplicates span names. If the middleware were to accidentally emit duplicate span names (e.g., two `test-channel process` spans), the assertion would silently pass since sets collapse duplicates. Using a list comprehension would surface unexpected duplicates and make the assertion stricter: `sorted([span['name'] for span in spans])`.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| spans = exporter.exported_spans_as_dict() | ||
|
|
||
| # A publish and a process span for each channel the message flows through (plus the initial create). | ||
| assert sorted({span['name'] for span in spans}) == snapshot( |
There was a problem hiding this comment.
P3: The test is well-constructed and matches FastStream's actual OTel middleware behavior. One subtle maintainability concern: the test uses sorted({span['name'] for span in spans}) with a set comprehension (line 39), which deduplicates span names. If the middleware were to accidentally emit duplicate span names (e.g., two test-channel process spans), the assertion would silently pass since sets collapse duplicates. Using a list comprehension would surface unexpected duplicates and make the assertion stricter: sorted([span['name'] for span in spans]).
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/otel_integrations/test_faststream.py, line 42:
<comment>The test is well-constructed and matches FastStream's actual OTel middleware behavior. One subtle maintainability concern: the test uses `sorted({span['name'] for span in spans})` with a set comprehension (line 39), which deduplicates span names. If the middleware were to accidentally emit duplicate span names (e.g., two `test-channel process` spans), the assertion would silently pass since sets collapse duplicates. Using a list comprehension would surface unexpected duplicates and make the assertion stricter: `sorted([span['name'] for span in spans])`.</comment>
<file context>
@@ -0,0 +1,56 @@
+ spans = exporter.exported_spans_as_dict()
+
+ # A publish and a process span for each channel the message flows through (plus the initial create).
+ assert sorted({span['name'] for span in spans}) == snapshot(
+ [
+ 'another-channel process',
</file context>
| assert sorted({span['name'] for span in spans}) == snapshot( | |
| assert sorted([span['name'] for span in spans]) == snapshot( |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pyproject.toml`:
- Line 165: Update the FastStream dependency entry in pyproject.toml to include
the otel extra alongside redis, ensuring
faststream.redis.opentelemetry.RedisTelemetryMiddleware is available when
dependencies are installed while preserving the existing minimum version.
In `@tests/otel_integrations/test_faststream.py`:
- Around line 42-50: Update the span-name assertion to sort the full list of
span names directly instead of constructing a set, preserving duplicate
publish/process spans while retaining the existing snapshot comparison.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 1e63e30e-08a0-42d7-bfd2-cd4143aa744e
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (2)
pyproject.tomltests/otel_integrations/test_faststream.py
🔗 Linked repositories identified
CodeRabbit considers these linked repositories for cross-repo context during reviews:
pydantic/logfire(manual)pydantic/pydantic-ai(manual)pydantic/pydantic(auto-detected)
| "sqlmodel>=0.0.15", | ||
| "mypy>=1.10.0", | ||
| "celery>=5.4.0", | ||
| "faststream[redis]>=0.5.0", |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== pyproject references =="
rg -n "faststream(\[.*\])?>=|RedisTelemetryMiddleware|opentelemetry|faststream.redis" pyproject.toml tests src docs || true
echo
echo "== files mentioning RedisTelemetryMiddleware =="
rg -n "RedisTelemetryMiddleware" . || true
echo
echo "== list candidate files =="
fd -a "faststream" . || true
fd -a "pyproject.toml" . || trueRepository: pydantic/logfire
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Focus on the files that can justify the dependency claim.
git ls-files | rg '(^|/)(pyproject\.toml|tests?/|docs?/).*' || true
echo
echo "== mentions of faststream in repo =="
rg -n "faststream" . || true
echo
echo "== mentions of RedisTelemetryMiddleware in repo =="
rg -n "RedisTelemetryMiddleware" . || trueRepository: pydantic/logfire
Length of output: 26920
🌐 Web query:
FastStream RedisTelemetryMiddleware first release opentelemetry 0.5.7
💡 Result:
The RedisTelemetryMiddleware component in FastStream is used for integrating OpenTelemetry tracing and metrics with the Redis broker [1][2][3]. While FastStream version 0.5.7 exists, it is not specifically identified as the first release of this middleware in available documentation [4]. The middleware is a standard part of the FastStream Redis integration and is used by passing an instance of it to the middlewares parameter when initializing a RedisBroker [2][3]. Key details regarding this component include: - Functionality: It provides telemetry for Redis operations, including support for custom tracer and meter providers [1][5]. - Usage: Developers typically initialize it within the RedisBroker configuration, for example: broker = RedisBroker(middlewares=(RedisTelemetryMiddleware(),)) [2][3]. - Evolution: FastStream has undergone significant changes in versions 0.6.x and 0.7.x, including updates to middleware handling and the removal of deprecated features (e.g., publisher/subscriber-level middlewares were removed in favor of broker-level or app-level usage in 0.7.0) [6][7][8]. If you are looking for the exact version where RedisTelemetryMiddleware was first introduced, it is recommended to review the pull requests and commit history on the official FastStream GitHub repository [9].
Citations:
- 1: https://faststream.ag2.ai/latest/api/faststream/redis/opentelemetry/RedisTelemetryMiddleware/
- 2: https://pydantic.dev/docs/logfire/integrations/event-streams/faststream/
- 3: https://faststream.ag2.ai/latest/getting-started/observability/opentelemetry/
- 4: https://pypi.org/project/faststream/0.5.7/
- 5: https://faststream.ag2.ai/latest/public_api/faststream/redis/opentelemetry/RedisTelemetryMiddleware/
- 6: https://faststream.ag2.ai/latest/release/
- 7: https://github.com/ag2ai/faststream/releases/tag/0.7.0
- 8: https://github.com/ag2ai/faststream/releases/tag/0.7.0rc0
- 9: https://github.com/ag2ai/faststream/releases
🌐 Web query:
site:dummy.faststream.airt.ai 0.5 release opentelemetry faststream.redis.opentelemetry
💡 Result:
In FastStream version 0.5, OpenTelemetry support is integrated as a core feature, allowing for the collection and exporting of telemetry data such as traces across your messaging infrastructure [1][2]. While there is no specific, isolated package named faststream.redis.opentelemetry, OpenTelemetry functionality is provided through the base FastStream OpenTelemetry middleware [3]. Users can implement telemetry for Redis brokers by utilizing the standard FastStream OpenTelemetry integration, which is designed to work with supported brokers including Redis [2][4]. The 0.5 release cycle significantly expanded these observability capabilities, with documentation explicitly noting that by version 0.5.28, the planned OpenTelemetry features were completed [1]. You can access these tools by importing them from the faststream.opentelemetry module and applying them to your broker configurations [2][3].
Citations:
- 1: https://dummy.faststream.airt.ai/0.5/release/
- 2: https://dummy.faststream.airt.ai/0.5/getting-started/opentelemetry/
- 3: https://dummy.faststream.airt.ai/0.5/api/faststream/opentelemetry/middleware/BaseTelemetryMiddleware/
- 4: https://dummy.faststream.airt.ai/0.5/public_api/faststream/redis/RedisBroker/
Add FastStream's otel extra. faststream[redis]>=0.5.0 doesn't cover faststream.redis.opentelemetry.RedisTelemetryMiddleware; the test needs the OpenTelemetry extra at install time.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pyproject.toml` at line 165, Update the FastStream dependency entry in
pyproject.toml to include the otel extra alongside redis, ensuring
faststream.redis.opentelemetry.RedisTelemetryMiddleware is available when
dependencies are installed while preserving the existing minimum version.
| assert sorted({span['name'] for span in spans}) == snapshot( | ||
| [ | ||
| 'another-channel process', | ||
| 'another-channel publish', | ||
| 'test-channel create', | ||
| 'test-channel process', | ||
| 'test-channel publish', | ||
| ] | ||
| ) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Preserve duplicate spans in the assertion.
The set comprehension discards multiplicity, so duplicate publish/process spans with the same name would still pass. Compare the sorted list directly.
Proposed fix
- assert sorted({span['name'] for span in spans}) == snapshot(
+ assert sorted(span['name'] for span in spans) == snapshot(📝 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.
| assert sorted({span['name'] for span in spans}) == snapshot( | |
| [ | |
| 'another-channel process', | |
| 'another-channel publish', | |
| 'test-channel create', | |
| 'test-channel process', | |
| 'test-channel publish', | |
| ] | |
| ) | |
| assert sorted(span['name'] for span in spans) == snapshot( | |
| [ | |
| 'another-channel process', | |
| 'another-channel publish', | |
| 'test-channel create', | |
| 'test-channel process', | |
| 'test-channel publish', | |
| ] | |
| ) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/otel_integrations/test_faststream.py` around lines 42 - 50, Update the
span-name assertion to sort the full list of span names directly instead of
constructing a set, preserving duplicate publish/process spans while retaining
the existing snapshot comparison.
1 participant
Summary
Adds an integration test for FastStream, the one documented integration that emits real spans but had no test coverage. It exercises FastStream's OpenTelemetry
RedisTelemetryMiddlewareagainst the in-memoryTestRedisBroker(no real Redis) and asserts the publish/process message spans plus their OpenTelemetry messaging semantic-convention attributes — mirroringdocs/integrations/event-streams/faststream.md.Context
Came out of a cross-reference of every integration guide against the SDK's test suite (checking each doc's
instrument_*call and span-name claims against what the tests actually assert). Result: the docs are well-rooted — of ~32 integrations with tests, all primary instrumentation methods and span-name claims match; the only genuine span-emitting integration with zero coverage was FastStream, which this fills.Test plan
tests/otel_integrations/test_faststream.pypasses deterministically (asserts span names + stable messaging attributes; skips the non-deterministic message IDs / timing metrics).faststream[redis]to the dev dependency group;test_redis.py/test_celery.pystill pass with it installed.Follow-ups