"""Telemetry unit tests: app gauges, process metrics, log export and counts. The OpenTelemetry packages are an optional extra (``tavolo-app[otel]``), so the provider-dependent tests skip when they are not installed. """ from __future__ import annotations import logging import os import sys import unittest from types import SimpleNamespace from typing import Optional from unittest.mock import patch from kaya.session import Session from tavolo.telemetry import ( CountingInMemorySessionStore, CountingRedisSessionStore, TelemetryMixin, _NoOtelLogs, telemetry_from_settings, ) from tavolo.config import Settings from tests.helpers import async_test try: from opentelemetry.sdk._logs import LoggingHandler, LoggerProvider from opentelemetry.sdk._logs.export import ( InMemoryLogRecordExporter, SimpleLogRecordProcessor, ) from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import InMemoryMetricReader from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, ) HAVE_OTEL = True except ImportError: # pragma: no cover - installs without the otel extra HAVE_OTEL = False def _settings(env: dict) -> Settings: with patch.dict(os.environ, env, clear=True): return Settings.from_env() class _FakeSessions: def __init__(self, count: int = 0) -> None: self.value: Optional[int] = count self.error: Optional[Exception] = None async def count(self) -> int: if self.error is not None: raise self.error assert self.value is not None return self.value class _FakeGames: def __init__(self, count: int = 0) -> None: self.value: Optional[int] = count async def count_ongoing(self) -> int: assert self.value is not None return self.value class _FakeRedis: """Minimal async Redis double covering the session-count scan.""" def __init__(self, keys: list) -> None: self.keys = keys async def scan_iter(self, match: str = "", count: int = 0): prefix = match.rstrip("*") for key in self.keys: if key.startswith(prefix): yield key class TelemetrySetupTest(unittest.TestCase): def test_disabled_returns_none(self) -> None: self.assertIsNone( telemetry_from_settings( _settings({}), _FakeSessions(), _FakeGames() ) ) def test_missing_otel_dependencies_raise(self) -> None: settings = _settings({"OTEL_ENABLED": "true"}) missing = "opentelemetry.exporter.otlp.proto.http._log_exporter" with patch.dict(sys.modules, {missing: None}): with self.assertRaises(RuntimeError): telemetry_from_settings(settings, _FakeSessions(), _FakeGames()) class SessionCountTest(unittest.TestCase): @async_test async def test_in_memory_counts_live_sessions(self) -> None: now = [1000.0] store = CountingInMemorySessionStore(clock=lambda: now[0]) await store.save("a", Session("a", {"user": "alice"}), max_age=10) await store.save("b", Session("b", {"user": "bob"}), max_age=10) await store.save("c", Session("c", {"user": "carol"})) # no expiry self.assertEqual(3, await store.count()) now[0] += 11 # a and b have expired self.assertEqual(1, await store.count()) await store.delete("c") self.assertEqual(0, await store.count()) @async_test async def test_redis_counts_only_session_keys(self) -> None: store = CountingRedisSessionStore( _FakeRedis(["kaya:session:a", "kaya:session:b", "tavolo:code:X"]) ) self.assertEqual(2, await store.count()) class NoOtelLogsFilterTest(unittest.TestCase): def test_filters_opentelemetry_records_only(self) -> None: filter_ = _NoOtelLogs() self.assertFalse( filter_.filter(logging.makeLogRecord({"name": "opentelemetry.sdk"})) ) self.assertTrue( filter_.filter(logging.makeLogRecord({"name": "tavolo.app"})) ) self.assertTrue( filter_.filter(logging.makeLogRecord({"name": "granian.server"})) ) def _metric_points(reader) -> dict: """Map metric name -> list of (attributes, value) data points.""" data = reader.get_metrics_data() assert data is not None points = {} for resource_metrics in data.resource_metrics: for scope_metrics in resource_metrics.scope_metrics: for metric in scope_metrics.metrics: points[metric.name] = [ (dict(point.attributes), point.value) for point in metric.data.data_points ] return points @unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed") class MetricExportTest(unittest.TestCase): # The instrumentation packages are process-wide singletons whose # callbacks cannot be re-registered, so the providers and mixin are # built once for the whole class (as the application does). sessions = _FakeSessions(0) games = _FakeGames(0) @classmethod def setUpClass(cls) -> None: resource = Resource.create({"service.name": "test"}) cls.reader = InMemoryMetricReader() cls.provider = MeterProvider(resource=resource, metric_readers=[cls.reader]) cls.span_exporter = InMemorySpanExporter() cls.tracer_provider = TracerProvider(resource=resource) cls.tracer_provider.add_span_processor( SimpleSpanProcessor(cls.span_exporter) ) cls.mixin = TelemetryMixin( meter_provider=cls.provider, tracer_provider=cls.tracer_provider, logger_provider=None, log_handler=None, session_store=cls.sessions, game_store=cls.games, poll_interval_seconds=1, ) cls.mixin.apply(None) # type: ignore[arg-type] @classmethod def tearDownClass(cls) -> None: cls.mixin.shutdown(None) # type: ignore[arg-type] @async_test async def test_app_gauges_report_polled_counts(self) -> None: self.sessions.value = 3 self.sessions.error = None self.games.value = 2 await self.mixin.refresh_counts() points = _metric_points(self.reader) self.assertEqual([({}, 3)], points["tavolo.active_users"]) self.assertEqual([({}, 2)], points["tavolo.ongoing_games"]) @async_test async def test_failed_refresh_keeps_last_value(self) -> None: self.sessions.value = 5 self.sessions.error = None await self.mixin.refresh_counts() self.sessions.error = RuntimeError("redis down") await self.mixin.refresh_counts() points = _metric_points(self.reader) self.assertEqual([({}, 5)], points["tavolo.active_users"]) @async_test async def test_process_and_gc_metrics_are_exported(self) -> None: await self.mixin.refresh_counts() points = _metric_points(self.reader) for name in ( "process.cpu.time", "process.cpu.utilization", "process.memory.usage", "process.memory.virtual", "cpython.gc.collections", "cpython.gc.collected_objects", "cpython.gc.uncollectable_objects", ): self.assertIn(name, points) self.assertGreater(points["process.memory.usage"][0][1], 0) modes = {attrs["type"] for attrs, _ in points["process.cpu.time"]} self.assertEqual({"user", "system"}, modes) generations = { attrs["cpython.gc.generation"] for attrs, _ in points["cpython.gc.collections"] } self.assertEqual({0, 1, 2}, generations) # Host-level metrics are deliberately not configured. self.assertNotIn("system.cpu.time", points) @async_test async def test_database_clients_are_instrumented(self) -> None: import asyncpg.connection import redis.asyncio self.assertTrue(hasattr(redis.asyncio.Redis.execute_command, "__wrapped__")) self.assertTrue(hasattr(asyncpg.connection.Connection.fetch, "__wrapped__")) # Exercise each wrapper through stubs: the downstream call fails (no # server), but the span is recorded before it does. client = object.__new__(redis.asyncio.Redis) client.connection_pool = SimpleNamespace( connection_kwargs={"host": "127.0.0.1", "port": 6379, "db": 0} ) with self.assertRaises(AttributeError): await client.execute_command("GET", "key") connection = object.__new__(asyncpg.connection.Connection) connection._aborted = True # keep __del__ quiet on the stub with self.assertRaises(Exception): await connection.fetch("SELECT 1") spans = { span.name: span for span in self.span_exporter.get_finished_spans() } self.assertEqual("GET ?", spans["GET"].attributes["db.statement"]) self.assertEqual("redis", spans["GET"].attributes["db.system"]) self.assertEqual("SELECT 1", spans["SELECT"].attributes["db.statement"]) self.assertEqual("postgresql", spans["SELECT"].attributes["db.system"]) @unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed") class LogExportTest(unittest.TestCase): def test_logs_are_exported_and_otel_records_filtered(self) -> None: exporter = InMemoryLogRecordExporter() resource = Resource.create({"service.name": "test"}) provider = LoggerProvider(resource=resource) provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) handler = LoggingHandler(level=logging.NOTSET, logger_provider=provider) handler.addFilter(_NoOtelLogs()) root = logging.getLogger() previous_level = root.level root.addHandler(handler) root.setLevel(logging.DEBUG) try: logging.getLogger("tavolo.test").info("exported message") logging.getLogger("opentelemetry.test").info("filtered message") bodies = [r.log_record.body for r in exporter.get_finished_logs()] self.assertIn("exported message", bodies) self.assertNotIn("filtered message", bodies) records = exporter.get_finished_logs() self.assertEqual( "test", records[0].resource.attributes["service.name"] ) finally: root.removeHandler(handler) handler.close() root.setLevel(previous_level) provider.shutdown() if __name__ == "__main__": unittest.main()