apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-04-25 00:16:23 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Test fixes for Kafka (#834)

Browse source 
Test fixes for Kafka:
- Consumer: isinstance(max_concurrency, int) instead of is not None:
  MagicMock won't pass the check
- Kafka test: updated expected topic name to
  tg.request.prompt.default (colons replaced with dots)
This commit is contained in:
cybermaggedon 2026-04-18 23:06:01 +01:00 • committed by GitHub
parent b615150624
commit 48da6c5f8b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 3 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

4
tests/unit/test_pubsub/test_kafka_backend.py
View file

@ -57,9 +57,9 @@ class TestKafkaParseTopic:
backend._parse_topic('unknown:tg:topic') backend._parse_topic('unknown:tg:topic')
def test_topic_with_flow_suffix(self, backend): def test_topic_with_flow_suffix(self, backend):
"""Topic names with flow suffix (e.g. :default) are preserved.""" """Topic names with flow suffix (e.g. :default) have colons replaced with dots."""
name, cls, durable = backend._parse_topic('request:tg:prompt:default') name, cls, durable = backend._parse_topic('request:tg:prompt:default')
assert name == 'tg.request.prompt:default' assert name == 'tg.request.prompt.default'
class TestKafkaRetention: class TestKafkaRetention:

2
trustgraph-base/trustgraph/base/consumer.py
View file

@ -58,7 +58,7 @@ class Consumer:
# consumers in the same group causes rebalance storms where # consumers in the same group causes rebalance storms where
# no consumer can fetch. Cap to the backend's limit. # no consumer can fetch. Cap to the backend's limit.
max_concurrency = getattr(backend, 'max_consumer_concurrency', None) max_concurrency = getattr(backend, 'max_consumer_concurrency', None)
if max_concurrency is not None and concurrency > max_concurrency: if isinstance(max_concurrency, int) and concurrency > max_concurrency:
logger.info( logger.info(
f"Capping concurrency from {concurrency} to " f"Capping concurrency from {concurrency} to "
f"{max_concurrency} (backend limit)" f"{max_concurrency} (backend limit)"