"""Integration: host-metrics read paths (DISTINCT ON latest + SQL date_bin history). Validates the two host_agent query helpers that back the slow host views, against a live Postgres: the latest-per-(resource,metric) lookup and the bucket-averaged history (which now aggregates in SQL instead of shipping raw rows to Python). Requires STEWARD_DATABASE_URL. """ from __future__ import annotations import asyncio import os import uuid from datetime import datetime, timedelta, timezone import pytest pytestmark = pytest.mark.integration _NEEDS_DB = pytest.mark.skipif( not os.environ.get("STEWARD_DATABASE_URL"), reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)", ) @pytest.fixture def app(): if not os.environ.get("STEWARD_DATABASE_URL"): pytest.skip("needs Postgres") from steward.app import create_app return create_app(testing=False) @_NEEDS_DB def test_latest_distinct_on_and_sql_bucketed_history(app): from sqlalchemy import text from steward.models.metrics import PluginMetric from plugins.host_agent.metrics_query import ( SOURCE_MODULE, _history_for_host, _latest_metrics_for_host, ) now = datetime.now(timezone.utc) hostname = "metrics-host-" + uuid.uuid4().hex[:8] async def _go(): async with app.db_sessionmaker() as s: async with s.begin(): await s.execute( text("DELETE FROM plugin_metrics WHERE resource_name LIKE :p"), {"p": hostname + "%"}, ) rows = [] # Host-level CPU samples, oldest → newest (10, 20, 30). for i, val in enumerate([10.0, 20.0, 30.0]): rows.append(PluginMetric( source_module=SOURCE_MODULE, resource_name=hostname, metric_name="cpu_pct", value=val, recorded_at=now - timedelta(minutes=30 - i * 10), )) # A sub-resource (root mount) to exercise the host:% match. rows.append(PluginMetric( source_module=SOURCE_MODULE, resource_name=hostname + ":/", metric_name="disk_used_pct", value=80.0, recorded_at=now - timedelta(minutes=5), )) s.add_all(rows) latest = await _latest_metrics_for_host(s, hostname) hist = await _history_for_host(s, hostname, now - timedelta(hours=1)) return latest, hist latest, hist = asyncio.run(_go()) # DISTINCT ON returns the newest sample per (resource, metric). assert latest[hostname]["cpu_pct"] == 30.0 assert latest[hostname + ":/"]["disk_used_pct"] == 80.0 # SQL date_bin aggregation returns cpu_pct buckets; averages stay in range. cpu = hist["cpu_pct"] assert cpu, "expected bucketed cpu_pct history" assert all(10.0 <= v <= 30.0 for _, v in cpu) @_NEEDS_DB def test_rollup_aggregates_old_raw_and_prunes(app): from sqlalchemy import text from steward.core.metrics_retention import rollup_plugin_metrics from steward.models.metrics import PluginMetric now = datetime.now(timezone.utc) hostname = "rollup-host-" + uuid.uuid4().hex[:8] async def _go(): async with app.db_sessionmaker() as s: async with s.begin(): for tbl in ("plugin_metrics", "plugin_metrics_hourly"): await s.execute( text(f"DELETE FROM {tbl} WHERE resource_name LIKE :p"), {"p": hostname + "%"}, ) old = (now - timedelta(days=10)).replace(minute=5, second=0, microsecond=0) s.add_all([ PluginMetric(source_module="host_agent", resource_name=hostname, metric_name="cpu_pct", value=10.0, recorded_at=old), PluginMetric(source_module="host_agent", resource_name=hostname, metric_name="cpu_pct", value=20.0, recorded_at=old.replace(minute=35)), # same hour bucket PluginMetric(source_module="host_agent", resource_name=hostname, metric_name="cpu_pct", value=99.0, recorded_at=now - timedelta(hours=1)), # recent → kept raw ]) async with s.begin(): counts = await rollup_plugin_metrics(s, raw_days=7, rollup_days=90, now=now) hrly = (await s.execute(text( "SELECT value_avg, sample_count FROM plugin_metrics_hourly " "WHERE resource_name = :h AND metric_name = 'cpu_pct'"), {"h": hostname})).all() raw_left = (await s.execute(text( "SELECT count(*) FROM plugin_metrics WHERE resource_name = :h"), {"h": hostname})).scalar() return counts, hrly, raw_left counts, hrly, raw_left = asyncio.run(_go()) assert counts["buckets_rolled"] == 1 assert counts["raw_rows_rolled"] == 2 assert len(hrly) == 1 assert abs(float(hrly[0][0]) - 15.0) < 0.001 # avg(10, 20) assert hrly[0][1] == 2 assert raw_left == 1 # only the recent sample remains @_NEEDS_DB def test_history_merges_hourly_and_raw(app): from sqlalchemy import text from steward.models.metrics import PluginMetric, PluginMetricHourly from plugins.host_agent.metrics_query import _history_for_host now = datetime.now(timezone.utc) hostname = "merge-host-" + uuid.uuid4().hex[:8] async def _go(): async with app.db_sessionmaker() as s: async with s.begin(): for tbl in ("plugin_metrics", "plugin_metrics_hourly"): await s.execute( text(f"DELETE FROM {tbl} WHERE resource_name LIKE :p"), {"p": hostname + "%"}, ) old_bucket = (now - timedelta(days=10)).replace(minute=0, second=0, microsecond=0) s.add(PluginMetricHourly( source_module="host_agent", resource_name=hostname, metric_name="cpu_pct", bucket=old_bucket, value_avg=42.0, value_max=50.0, sample_count=120)) s.add(PluginMetric( source_module="host_agent", resource_name=hostname, metric_name="cpu_pct", value=77.0, recorded_at=now - timedelta(hours=1))) # 30d range with a 7d raw window → spans the rollup boundary. return await _history_for_host(s, hostname, now - timedelta(days=30), raw_days=7) hist = asyncio.run(_go()) cpu = hist["cpu_pct"] vals = [v for _, v in cpu] assert 42.0 in vals, "expected the rolled-up hourly point" assert 77.0 in vals, "expected the recent raw point" assert cpu == sorted(cpu, key=lambda p: p[0]), "series must be time-ordered"