use "brewblox_history_" prefix for config

Author: steersbob

brewblox_history/models.py

@@ -33,7 +33,7 @@ def flatten(d, parent_key=''):
 class ServiceConfig(BaseSettings):
     model_config = SettingsConfigDict(
         env_file='.appenv',
-        env_prefix='brewblox_',
+        env_prefix='brewblox_history_',
         case_sensitive=False,
         json_schema_extra='ignore',
     )

brewblox_history/timeseries_api.py

@@ -199,10 +199,12 @@ async def timeseries_stream(ws: WebSocket):
                     'message': msg,
                 })
 
-    except WebSocketDisconnect:
+    except WebSocketDisconnect:  # pragma: no cover
         pass
 
     finally:
         # Coverage complains about next line -> exit not being covered
         for task in streams.values():  # pragma: no cover
             task.cancel()
+        await asyncio.gather(*streams.values(),
+                             return_exceptions=True)

brewblox_history/utils.py

@@ -123,7 +123,11 @@ def now() -> datetime:  # pragma: no cover
     return datetime.now(timezone.utc)
 
 
-def select_timeframe(start, duration, end, min_step) -> tuple[str, str, str]:
+def select_timeframe(start: DatetimeSrc_,
+                     duration: str,
+                     end: DatetimeSrc_,
+                     min_step: timedelta,
+                     ) -> tuple[str, str, str]:
     """Calculate start, end, and step for given start, duration, and end
 
     The returned `start` and `end` strings are either empty,

docker-compose.yml

@@ -12,15 +12,16 @@ services:
     image: victoriametrics/victoria-metrics:v1.88.0
     labels:
       - traefik.http.services.victoria.loadbalancer.server.port=8428
+    command: --envflag.enable=true --envflag.prefix=VM_
+    environment:
+      - VM_retentionPeriod=100y
+      - VM_influxMeasurementFieldSeparator=/
+      - VM_http_pathPrefix=/victoria
+      - VM_search_latencyOffset=10s
     volumes:
       - type: bind
         source: ./victoria
         target: /victoria-metrics-data
-    command: >-
-      --retentionPeriod=100y
-      --influxMeasurementFieldSeparator=/
-      --http.pathPrefix=/victoria
-      --search.latencyOffset=10s
 
   redis:
     image: redis:6.0
@@ -41,9 +42,12 @@ services:
       - type: bind
         source: ./parse_appenv.py
         target: /app/parse_appenv.py
+      - type: bind
+        source: ./entrypoint.sh
+        target: /app/entrypoint.sh
     environment:
-      - BREWBLOX_DEBUG=True
       - UVICORN_RELOAD=True
+      - BREWBLOX_HISTORY_DEBUG=True
     command: --unknown=flappy
     ports:
       - "5000:5000"

poetry.lock

@@ -33,6 +33,7 @@ pytest-docker = "*"
 pytest-asyncio = "*"
 pytest-httpx = "*"
 asgi-lifespan = "*"
+httpx-ws = "*"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

pyproject.toml

@@ -4,6 +4,7 @@
 """
 
 
+import asyncio
 import logging
 from collections.abc import Generator
 from pathlib import Path
@@ -12,9 +13,9 @@
 from asgi_lifespan import LifespanManager
 from fastapi import FastAPI
 from httpx import AsyncClient
+from httpx_ws.transport import ASGIWebSocketTransport
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource
 from pytest_docker.plugin import Services as DockerServices
-from starlette.testclient import TestClient
 
 from brewblox_history import app_factory, utils
 from brewblox_history.models import ServiceConfig
@@ -65,6 +66,23 @@ def config(monkeypatch: pytest.MonkeyPatch,
     yield cfg
 
 
+@pytest.fixture(autouse=True)
+def m_sleep(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest):
+    """
+    Allows keeping track of calls to asyncio sleep.
+    For tests, we want to reduce all sleep durations.
+    Set a breakpoint in the wrapper to track all calls.
+    """
+    real_func = asyncio.sleep
+
+    async def wrapper(delay: float, *args, **kwargs):
+        if delay > 0.1:
+            print(f'asyncio.sleep({delay}) in {request.node.name}')
+        return await real_func(delay, *args, **kwargs)
+    monkeypatch.setattr('asyncio.sleep', wrapper)
+    yield
+
+
 @pytest.fixture(autouse=True)
 def setup_logging(config):
     app_factory.setup_logging(True)
@@ -83,32 +101,28 @@ def app() -> FastAPI:
 
 
 @pytest.fixture
-async def client(app: FastAPI) -> Generator[AsyncClient, None, None]:
+async def manager(app: FastAPI) -> Generator[LifespanManager, None, None]:
     """
-    The default test client for making REST API calls.
-    Using this fixture will also guarantee that lifespan startup has happened.
+    AsyncClient does not automatically send ASGI lifespan events to the app
+    https://asgi.readthedocs.io/en/latest/specs/lifespan.html
 
-    Do not use `client` and `sync_client` at the same time.
+    For testing, this ensures that lifespan() functions are handled.
+    If you don't need to make HTTP requests, you can use the manager
+    without the `client` fixture.
     """
-    # AsyncClient does not automatically send ASGI lifespan events to the app
-    # https://asgi.readthedocs.io/en/latest/specs/lifespan.html
-    async with LifespanManager(app):
-        async with AsyncClient(app=app,
-                               base_url='http://test') as ac:
-            yield ac
+    async with LifespanManager(app) as mgr:
+        yield mgr
 
 
 @pytest.fixture
-def sync_client(app: FastAPI) -> Generator[TestClient, None, None]:
+async def client(app: FastAPI, manager: LifespanManager) -> Generator[AsyncClient, None, None]:
     """
-    The alternative test client for making REST API calls.
+    The default test client for making REST API calls.
     Using this fixture will also guarantee that lifespan startup has happened.
-
-    `sync_client` is provided because `client` cannot make websocket requests.
-
-    Do not use `client` and `sync_client` at the same time.
     """
-    # The Starlette TestClient does send lifespan events
-    # and does not need a separate LifespanManager
-    with TestClient(app=app, base_url='http://test') as c:
-        yield c
+    # AsyncClient does not automatically send ASGI lifespan events to the app
+    # https://asgi.readthedocs.io/en/latest/specs/lifespan.html
+    async with AsyncClient(app=app,
+                           base_url='http://test',
+                           transport=ASGIWebSocketTransport(app)) as ac:
+        yield ac

test/conftest.py

@@ -2,16 +2,17 @@
 Tests brewblox_history.timeseries_api
 """
 
+import asyncio
 from datetime import datetime, timezone
 from time import time_ns
 from unittest.mock import ANY, AsyncMock, Mock
 
 import pytest
 from fastapi import FastAPI
 from httpx import AsyncClient
+from httpx_ws import aconnect_ws
 from pytest import approx
 from pytest_mock import MockerFixture
-from starlette.testclient import TestClient, WebSocketTestSession
 
 from brewblox_history import app_factory, timeseries_api, utils
 from brewblox_history.models import (ServiceConfig, TimeSeriesCsvQuery,
@@ -166,7 +167,7 @@ async def csv_mock(args: TimeSeriesCsvQuery):
     assert resp.text == 'a,b,c\n'
 
 
-async def test_stream(sync_client: TestClient, config: ServiceConfig, m_victoria: Mock):
+async def test_stream(client: AsyncClient, config: ServiceConfig, m_victoria: Mock):
     config.ranges_interval = 0.001
     m_victoria.metrics.return_value = [
         TimeSeriesMetric(
@@ -200,18 +201,16 @@ async def test_stream(sync_client: TestClient, config: ServiceConfig, m_victoria
         ),
     ]
 
-    with sync_client.websocket_connect('/timeseries/stream') as ws:
-        ws: WebSocketTestSession
-
+    async with aconnect_ws('/timeseries/stream', client) as ws:
         # Metrics
-        ws.send_json({
+        await ws.send_json({
             'id': 'test-metrics',
             'command': 'metrics',
             'query': {
                 'fields': ['a', 'b', 'c'],
             },
         })
-        resp = ws.receive_json()
+        resp = await ws.receive_json()
         assert resp == {
             'id': 'test-metrics',
             'data': {
@@ -220,15 +219,15 @@ async def test_stream(sync_client: TestClient, config: ServiceConfig, m_victoria
         }
 
         # Ranges
-        ws.send_json({
+        await ws.send_json({
             'id': 'test-ranges-once',
             'command': 'ranges',
             'query': {
                 'fields': ['a', 'b', 'c'],
                 'end': '2021-07-15T14:29:30.000Z',
             },
         })
-        resp = ws.receive_json()
+        resp = await ws.receive_json()
         assert resp == {
             'id': 'test-ranges-once',
             'data': {
@@ -238,23 +237,23 @@ async def test_stream(sync_client: TestClient, config: ServiceConfig, m_victoria
         }
 
         # Live ranges
-        ws.send_json({
+        await ws.send_json({
             'id': 'test-ranges-live',
             'command': 'ranges',
             'query': {
                 'fields': ['a', 'b', 'c'],
                 'duration': '30m',
             },
         })
-        resp = ws.receive_json()
+        resp = await ws.receive_json()
         assert resp == {
             'id': 'test-ranges-live',
             'data': {
                 'initial': True,
                 'ranges': [ANY, ANY, ANY],
             },
         }
-        resp = ws.receive_json()
+        resp = await ws.receive_json()
         assert resp == {
             'id': 'test-ranges-live',
             'data': {
@@ -264,13 +263,19 @@ async def test_stream(sync_client: TestClient, config: ServiceConfig, m_victoria
         }
 
         # Stop live ranges
-        ws.send_json({
+        await ws.send_json({
             'id': 'test-ranges-live',
             'command': 'stop',
         })
 
+        # https://github.com/frankie567/httpx-ws/issues/49
+        await ws.close()
+        await asyncio.gather(ws._background_receive_task,
+                             ws._background_keepalive_ping_task,
+                             return_exceptions=True)
+
 
-async def test_stream_error(sync_client: TestClient, m_victoria: Mock):
+async def test_stream_error(client: AsyncClient, m_victoria: Mock):
     dt = datetime(2021, 7, 15, 19, tzinfo=timezone.utc)
     m_victoria.ranges.side_effect = RuntimeError
     m_victoria.metrics.return_value = [
@@ -281,16 +286,14 @@ async def test_stream_error(sync_client: TestClient, m_victoria: Mock):
         ),
     ]
 
-    with sync_client.websocket_connect('/timeseries/stream') as ws:
-        ws: WebSocketTestSession
-
+    async with aconnect_ws('/timeseries/stream', client) as ws:
         # Invalid request
-        ws.send_json({'empty': True})
-        resp = ws.receive_json()
+        await ws.send_json({'empty': True})
+        resp = await ws.receive_json()
         assert resp['error']
 
         # Backend raises error
-        ws.send_json({
+        await ws.send_json({
             'id': 'test-ranges-once',
             'command': 'ranges',
             'query': {
@@ -300,15 +303,15 @@ async def test_stream_error(sync_client: TestClient, m_victoria: Mock):
         })
 
         # Other command is OK
-        ws.send_json({
+        await ws.send_json({
             'id': 'test-metrics',
             'command': 'metrics',
             'query': {
                 'fields': ['a', 'b', 'c'],
             },
         })
 
-        resp = ws.receive_json()
+        resp = await ws.receive_json()
         assert resp == {
             'id': 'test-metrics',
             'data': {
@@ -319,3 +322,9 @@ async def test_stream_error(sync_client: TestClient, m_victoria: Mock):
                 }],
             },
         }
+
+        # https://github.com/frankie567/httpx-ws/issues/49
+        await ws.close()
+        await asyncio.gather(ws._background_receive_task,
+                             ws._background_keepalive_ping_task,
+                             return_exceptions=True)