wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
21 lines
541 B
Python
21 lines
541 B
Python
"""Tests for traffic control behavior."""
|
|
|
|
import pytest
|
|
|
|
from deeptutor.services.llm.traffic_control import TrafficController
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_traffic_control_context() -> None:
|
|
"""Traffic control context manager should acquire and release slots."""
|
|
controller = TrafficController(
|
|
provider_name="test",
|
|
max_concurrency=1,
|
|
requests_per_minute=60,
|
|
)
|
|
|
|
async with controller:
|
|
assert controller._semaphore.locked() is True
|
|
|
|
assert controller._semaphore.locked() is False
|