diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index db4af91..6427c4d 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -35,6 +35,11 @@ jobs:
- name: Install dependencies
run: uv sync --locked --extra dev
+ - name: Set up matplotlib cache
+ run: |
+ echo "MPLCONFIGDIR=$RUNNER_TEMP/matplotlib" >> "$GITHUB_ENV"
+ mkdir -p "$RUNNER_TEMP/matplotlib"
+
- name: Run tests with coverage
run: |
uv run pytest \
diff --git a/src/meshmon/reports.py b/src/meshmon/reports.py
index faacd4f..1455aea 100644
--- a/src/meshmon/reports.py
+++ b/src/meshmon/reports.py
@@ -507,12 +507,18 @@ def aggregate_yearly(role: str, year: int) -> YearlyAggregate:
"""
agg = YearlyAggregate(year=year, role=role)
metrics = get_metrics_for_role(role)
+ today = date.today()
- # Process month by month to limit memory usage
- for month in range(1, 13):
- # Don't aggregate future months
- if date(year, month, 1) > date.today():
- break
+ periods = get_available_periods(role)
+ months_with_data = sorted({month for y, month in periods if y == year})
+
+ if year > today.year:
+ months_with_data = []
+ elif year == today.year:
+ months_with_data = [month for month in months_with_data if month <= today.month]
+
+ # Process only months that have data to avoid unnecessary daily scans.
+ for month in months_with_data:
monthly = aggregate_monthly(role, year, month)
if monthly.daily: # Has data
agg.monthly.append(monthly)
diff --git a/tests/charts/test_chart_io.py b/tests/charts/test_chart_io.py
index b6e6086..4a1e269 100644
--- a/tests/charts/test_chart_io.py
+++ b/tests/charts/test_chart_io.py
@@ -116,10 +116,9 @@ class TestLoadChartStats:
def test_returns_empty_on_invalid_json(self, configured_env):
"""Returns empty dict on invalid JSON."""
- cfg = __import__("meshmon.env", fromlist=["get_config"]).get_config()
- stats_path = cfg.out_dir / "assets" / "repeater" / "chart_stats.json"
+ stats_path = configured_env["out_dir"] / "assets" / "repeater" / "chart_stats.json"
stats_path.parent.mkdir(parents=True, exist_ok=True)
- stats_path.write_text("not valid json {{{")
+ stats_path.write_text("not valid json {{{", encoding="utf-8")
loaded = load_chart_stats("repeater")
diff --git a/tests/charts/test_chart_render.py b/tests/charts/test_chart_render.py
index aecdcc3..0a4bc89 100644
--- a/tests/charts/test_chart_render.py
+++ b/tests/charts/test_chart_render.py
@@ -3,6 +3,7 @@
import os
from datetime import datetime, timedelta
from pathlib import Path
+from xml.etree import ElementTree as ET
import pytest
@@ -16,6 +17,14 @@ from meshmon.charts import (
from .conftest import extract_svg_data_attributes, normalize_svg_for_snapshot
+def _svg_viewbox_dims(svg: str) -> tuple[float, float]:
+ root = ET.fromstring(svg)
+ viewbox = root.attrib.get("viewBox")
+ assert viewbox is not None
+ _, _, width, height = viewbox.split()
+ return float(width), float(height)
+
+
class TestRenderChartSvg:
"""Tests for render_chart_svg function."""
@@ -34,10 +43,14 @@ class TestRenderChartSvg:
def test_respects_width_height(self, sample_timeseries, light_theme):
"""SVG respects specified dimensions."""
- svg = render_chart_svg(sample_timeseries, light_theme, width=600, height=200)
+ svg_default = render_chart_svg(sample_timeseries, light_theme)
+ svg_small = render_chart_svg(sample_timeseries, light_theme, width=600, height=200)
- # Check viewBox or dimensions in SVG attributes
- assert "600" in svg or "viewBox" in svg
+ default_w, default_h = _svg_viewbox_dims(svg_default)
+ small_w, small_h = _svg_viewbox_dims(svg_small)
+
+ assert small_w < default_w
+ assert small_h < default_h
def test_uses_theme_colors(self, sample_timeseries, light_theme, dark_theme):
"""Different themes produce different colors."""
@@ -97,6 +110,8 @@ class TestDataPointsInjection:
for point in data["points"]:
assert "ts" in point
assert "v" in point
+ assert isinstance(point["ts"], int)
+ assert isinstance(point["v"], (int, float))
def test_includes_metadata_attributes(self, sample_timeseries, light_theme):
"""SVG includes metric, period, theme attributes."""
diff --git a/tests/charts/test_statistics.py b/tests/charts/test_statistics.py
index a2b367d..aaeed22 100644
--- a/tests/charts/test_statistics.py
+++ b/tests/charts/test_statistics.py
@@ -11,6 +11,8 @@ from meshmon.charts import (
calculate_statistics,
)
+BASE_TIME = datetime(2024, 1, 1, 0, 0, 0)
+
class TestCalculateStatistics:
"""Tests for calculate_statistics function."""
@@ -113,7 +115,7 @@ class TestStatisticsWithVariousData:
def test_constant_values(self):
"""All same values gives min=avg=max."""
- now = datetime.now()
+ now = BASE_TIME
points = [DataPoint(timestamp=now + timedelta(hours=i), value=5.0) for i in range(10)]
ts = TimeSeries(metric="test", role="companion", period="day", points=points)
@@ -125,7 +127,7 @@ class TestStatisticsWithVariousData:
def test_increasing_values(self):
"""Increasing values have correct stats."""
- now = datetime.now()
+ now = BASE_TIME
points = [DataPoint(timestamp=now + timedelta(hours=i), value=float(i)) for i in range(10)]
ts = TimeSeries(metric="test", role="companion", period="day", points=points)
@@ -138,7 +140,7 @@ class TestStatisticsWithVariousData:
def test_negative_values(self):
"""Handles negative values correctly."""
- now = datetime.now()
+ now = BASE_TIME
points = [
DataPoint(timestamp=now, value=-10.0),
DataPoint(timestamp=now + timedelta(hours=1), value=-5.0),
@@ -154,7 +156,7 @@ class TestStatisticsWithVariousData:
def test_large_values(self):
"""Handles large values correctly."""
- now = datetime.now()
+ now = BASE_TIME
points = [
DataPoint(timestamp=now, value=1e10),
DataPoint(timestamp=now + timedelta(hours=1), value=1e11),
@@ -168,7 +170,7 @@ class TestStatisticsWithVariousData:
def test_small_decimal_values(self):
"""Handles small decimal values correctly."""
- now = datetime.now()
+ now = BASE_TIME
points = [
DataPoint(timestamp=now, value=0.001),
DataPoint(timestamp=now + timedelta(hours=1), value=0.002),
diff --git a/tests/charts/test_timeseries.py b/tests/charts/test_timeseries.py
index 7018058..e32a708 100644
--- a/tests/charts/test_timeseries.py
+++ b/tests/charts/test_timeseries.py
@@ -11,13 +11,15 @@ from meshmon.charts import (
)
from meshmon.db import insert_metrics
+BASE_TIME = datetime(2024, 1, 1, 0, 0, 0)
+
class TestDataPoint:
"""Tests for DataPoint dataclass."""
def test_stores_timestamp_and_value(self):
"""Stores timestamp and value."""
- ts = datetime.now()
+ ts = BASE_TIME
dp = DataPoint(timestamp=ts, value=3.85)
assert dp.timestamp == ts
@@ -25,13 +27,13 @@ class TestDataPoint:
def test_value_types(self):
"""Accepts float and int values."""
- ts = datetime.now()
+ ts = BASE_TIME
dp_float = DataPoint(timestamp=ts, value=3.85)
assert dp_float.value == 3.85
- dp_int = DataPoint(timestamp=ts, value=100.0)
- assert dp_int.value == 100.0
+ dp_int = DataPoint(timestamp=ts, value=100)
+ assert dp_int.value == 100
class TestTimeSeries:
@@ -123,7 +125,7 @@ class TestLoadTimeseriesFromDb:
ts = load_timeseries_from_db(
role="repeater",
metric="bat",
- end_time=datetime.now(),
+ end_time=BASE_TIME,
lookback=timedelta(hours=1),
period="week",
)
@@ -157,7 +159,7 @@ class TestLoadTimeseriesFromDb:
ts = load_timeseries_from_db(
role="repeater",
metric="nonexistent_metric",
- end_time=datetime.now(),
+ end_time=BASE_TIME,
lookback=timedelta(hours=1),
period="day",
)
diff --git a/tests/charts/test_transforms.py b/tests/charts/test_transforms.py
index 616f09f..4b03f50 100644
--- a/tests/charts/test_transforms.py
+++ b/tests/charts/test_transforms.py
@@ -10,6 +10,8 @@ from meshmon.charts import (
)
from meshmon.db import insert_metrics
+BASE_TIME = datetime(2024, 1, 1, 0, 0, 0)
+
class TestCounterToRateConversion:
"""Tests for counter metric rate conversion."""
@@ -35,8 +37,9 @@ class TestCounterToRateConversion:
assert len(ts.points) == 4
# All rates should be positive (counter increasing)
+ expected_rate = (100.0 / 900.0) * 60.0
for p in ts.points:
- assert p.value >= 0
+ assert p.value == pytest.approx(expected_rate)
def test_handles_counter_reset(self, initialized_db, configured_env):
"""Counter resets (negative delta) are skipped."""
@@ -58,6 +61,11 @@ class TestCounterToRateConversion:
# Reset point should be skipped, so fewer points
assert len(ts.points) == 2 # Only valid deltas
+ expected_rate = (100.0 / 900.0) * 60.0
+ assert ts.points[0].timestamp == datetime.fromtimestamp(base_ts + 900)
+ assert ts.points[1].timestamp == datetime.fromtimestamp(base_ts + 2700)
+ assert ts.points[0].value == pytest.approx(expected_rate)
+ assert ts.points[1].value == pytest.approx(expected_rate)
def test_applies_scale_factor(self, initialized_db, configured_env):
"""Counter rate is scaled (typically x60 for per-minute)."""
@@ -136,19 +144,19 @@ class TestGaugeValueTransform:
class TestTimeBinning:
"""Tests for time series aggregation/binning."""
- def test_no_binning_for_day(self, initialized_db, configured_env):
+ def test_no_binning_for_day(self):
"""Day period uses raw data (no binning)."""
assert PERIOD_CONFIG["day"].bin_seconds is None
- def test_30_min_bins_for_week(self, initialized_db, configured_env):
+ def test_30_min_bins_for_week(self):
"""Week period uses 30-minute bins."""
assert PERIOD_CONFIG["week"].bin_seconds == 1800
- def test_2_hour_bins_for_month(self, initialized_db, configured_env):
+ def test_2_hour_bins_for_month(self):
"""Month period uses 2-hour bins."""
assert PERIOD_CONFIG["month"].bin_seconds == 7200
- def test_1_day_bins_for_year(self, initialized_db, configured_env):
+ def test_1_day_bins_for_year(self):
"""Year period uses 1-day bins."""
assert PERIOD_CONFIG["year"].bin_seconds == 86400
@@ -181,7 +189,7 @@ class TestEmptyData:
ts = load_timeseries_from_db(
role="repeater",
metric="nonexistent",
- end_time=datetime.now(),
+ end_time=BASE_TIME,
lookback=timedelta(days=1),
period="day",
)
@@ -199,7 +207,7 @@ class TestEmptyData:
ts = load_timeseries_from_db(
role="repeater",
metric="bat",
- end_time=datetime.now(),
+ end_time=BASE_TIME,
lookback=timedelta(hours=1),
period="day",
)
diff --git a/tests/client/test_connect.py b/tests/client/test_connect.py
index d8c273e..c899f07 100644
--- a/tests/client/test_connect.py
+++ b/tests/client/test_connect.py
@@ -1,6 +1,6 @@
"""Tests for MeshCore connection functions."""
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock
import pytest
@@ -12,6 +12,13 @@ from meshmon.meshcore_client import (
)
+def _reset_config():
+ import meshmon.env
+
+ meshmon.env._config = None
+ return meshmon.env.get_config()
+
+
class TestAutoDetectSerialPort:
"""Tests for auto_detect_serial_port function."""
@@ -63,12 +70,20 @@ class TestAutoDetectSerialPort:
assert result is None
- def test_handles_import_error(self):
+ def test_handles_import_error(self, monkeypatch):
"""Returns None when pyserial not installed."""
- with patch.dict("sys.modules", {"serial": None, "serial.tools": None, "serial.tools.list_ports": None}):
- # Force re-import to test import error handling
- # This test may need adjustment based on actual import structure
- pass
+ import builtins
+
+ real_import = builtins.__import__
+
+ def mock_import(name, *args, **kwargs):
+ if name in {"serial", "serial.tools.list_ports"}:
+ raise ImportError("No module named 'serial'")
+ return real_import(name, *args, **kwargs)
+
+ monkeypatch.setattr(builtins, "__import__", mock_import)
+
+ assert auto_detect_serial_port() is None
class TestConnectFromEnv:
@@ -89,19 +104,22 @@ class TestConnectFromEnv:
monkeypatch.setattr("meshmon.meshcore_client.MESHCORE_AVAILABLE", True)
monkeypatch.setenv("MESH_TRANSPORT", "serial")
monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
+ monkeypatch.setenv("MESH_SERIAL_BAUD", "57600")
+ monkeypatch.setenv("MESH_DEBUG", "1")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
- mock_create = AsyncMock(return_value=MagicMock())
+ mock_client = MagicMock()
+ mock_create = AsyncMock(return_value=mock_client)
mock_meshcore = MagicMock()
mock_meshcore.create_serial = mock_create
monkeypatch.setattr("meshmon.meshcore_client.MeshCore", mock_meshcore)
- await connect_from_env()
+ result = await connect_from_env()
- mock_create.assert_called_once()
+ assert result is mock_client
+ mock_create.assert_called_once_with("/dev/ttyACM0", 57600, debug=True)
@pytest.mark.asyncio
async def test_tcp_connection(self, configured_env, monkeypatch):
@@ -111,18 +129,19 @@ class TestConnectFromEnv:
monkeypatch.setenv("MESH_TCP_HOST", "localhost")
monkeypatch.setenv("MESH_TCP_PORT", "4403")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
- mock_create = AsyncMock(return_value=MagicMock())
+ mock_client = MagicMock()
+ mock_create = AsyncMock(return_value=mock_client)
mock_meshcore = MagicMock()
mock_meshcore.create_tcp = mock_create
monkeypatch.setattr("meshmon.meshcore_client.MeshCore", mock_meshcore)
- await connect_from_env()
+ result = await connect_from_env()
- mock_create.assert_called_once()
+ assert result is mock_client
+ mock_create.assert_called_once_with("localhost", 4403)
@pytest.mark.asyncio
async def test_unknown_transport(self, configured_env, monkeypatch):
@@ -130,8 +149,7 @@ class TestConnectFromEnv:
monkeypatch.setattr("meshmon.meshcore_client.MESHCORE_AVAILABLE", True)
monkeypatch.setenv("MESH_TRANSPORT", "unknown")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
result = await connect_from_env()
@@ -144,8 +162,7 @@ class TestConnectFromEnv:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
mock_create = AsyncMock(side_effect=Exception("Connection failed"))
mock_meshcore = MagicMock()
@@ -156,6 +173,7 @@ class TestConnectFromEnv:
result = await connect_from_env()
assert result is None
+ mock_create.assert_called_once()
@pytest.mark.asyncio
async def test_ble_connection(self, configured_env, monkeypatch):
@@ -165,18 +183,19 @@ class TestConnectFromEnv:
monkeypatch.setenv("MESH_BLE_ADDR", "AA:BB:CC:DD:EE:FF")
monkeypatch.setenv("MESH_BLE_PIN", "123456")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
- mock_create = AsyncMock(return_value=MagicMock())
+ mock_client = MagicMock()
+ mock_create = AsyncMock(return_value=mock_client)
mock_meshcore = MagicMock()
mock_meshcore.create_ble = mock_create
monkeypatch.setattr("meshmon.meshcore_client.MeshCore", mock_meshcore)
- await connect_from_env()
+ result = await connect_from_env()
- mock_create.assert_called_once()
+ assert result is mock_client
+ mock_create.assert_called_once_with("AA:BB:CC:DD:EE:FF", pin="123456")
@pytest.mark.asyncio
async def test_ble_missing_address(self, configured_env, monkeypatch):
@@ -185,8 +204,7 @@ class TestConnectFromEnv:
monkeypatch.setenv("MESH_TRANSPORT", "ble")
# Don't set MESH_BLE_ADDR
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
result = await connect_from_env()
@@ -199,23 +217,24 @@ class TestConnectFromEnv:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
# Don't set MESH_SERIAL_PORT to trigger auto-detection
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
# Set up mock port detection
mock_port = MagicMock()
mock_port.device = "/dev/ttyACM0"
mock_serial_port.tools.list_ports.comports.return_value = [mock_port]
- mock_create = AsyncMock(return_value=MagicMock())
+ mock_client = MagicMock()
+ mock_create = AsyncMock(return_value=mock_client)
mock_meshcore = MagicMock()
mock_meshcore.create_serial = mock_create
monkeypatch.setattr("meshmon.meshcore_client.MeshCore", mock_meshcore)
- await connect_from_env()
+ result = await connect_from_env()
- mock_create.assert_called_once()
+ assert result is mock_client
+ mock_create.assert_called_once_with("/dev/ttyACM0", 115200, debug=False)
@pytest.mark.asyncio
async def test_serial_auto_detect_fails(self, configured_env, monkeypatch, mock_serial_port):
@@ -224,8 +243,7 @@ class TestConnectFromEnv:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
# Don't set MESH_SERIAL_PORT to trigger auto-detection
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
# No ports available
mock_serial_port.tools.list_ports.comports.return_value = []
@@ -245,8 +263,7 @@ class TestConnectWithLock:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
mock_client = MagicMock()
mock_client.disconnect = AsyncMock()
@@ -269,8 +286,7 @@ class TestConnectWithLock:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
mock_create = AsyncMock(side_effect=Exception("Connection failed"))
mock_meshcore = MagicMock()
@@ -288,9 +304,7 @@ class TestConnectWithLock:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
- import meshmon.env
- meshmon.env._config = None
- cfg = meshmon.env.get_config()
+ cfg = _reset_config()
mock_client = MagicMock()
mock_client.disconnect = AsyncMock()
@@ -313,9 +327,7 @@ class TestConnectWithLock:
monkeypatch.setenv("MESH_TCP_HOST", "localhost")
monkeypatch.setenv("MESH_TCP_PORT", "4403")
- import meshmon.env
- meshmon.env._config = None
- cfg = meshmon.env.get_config()
+ cfg = _reset_config()
mock_client = MagicMock()
mock_client.disconnect = AsyncMock()
@@ -338,8 +350,7 @@ class TestConnectWithLock:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
- import meshmon.env
- meshmon.env._config = None
+ _reset_config()
mock_client = MagicMock()
mock_client.disconnect = AsyncMock(side_effect=Exception("Disconnect error"))
@@ -363,8 +374,7 @@ class TestConnectWithLock:
monkeypatch.setenv("MESH_TRANSPORT", "serial")
monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
- import meshmon.env
- meshmon.env._config = None
+ cfg = _reset_config()
mock_create = AsyncMock(side_effect=Exception("Connection failed"))
mock_meshcore = MagicMock()
@@ -377,7 +387,6 @@ class TestConnectWithLock:
# Lock should be released after exiting context
# We can verify by acquiring it again without timeout
- cfg = meshmon.env.get_config()
lock_path = cfg.state_dir / "serial.lock"
if lock_path.exists():
import fcntl
diff --git a/tests/client/test_contacts.py b/tests/client/test_contacts.py
index d3c96cb..d43d9d3 100644
--- a/tests/client/test_contacts.py
+++ b/tests/client/test_contacts.py
@@ -1,5 +1,6 @@
"""Tests for contact lookup functions."""
+from types import SimpleNamespace
from unittest.mock import MagicMock
@@ -28,6 +29,7 @@ class TestGetContactByName:
result = get_contact_by_name(mock_meshcore_client, "NonExistent")
assert result is None
+ mock_meshcore_client.get_contact_by_name.assert_called_once_with("NonExistent")
def test_returns_none_when_method_not_available(self):
"""Returns None when get_contact_by_name method not available."""
@@ -48,6 +50,7 @@ class TestGetContactByName:
result = get_contact_by_name(mock_meshcore_client, "TestNode")
assert result is None
+ mock_meshcore_client.get_contact_by_name.assert_called_once_with("TestNode")
class TestGetContactByKeyPrefix:
@@ -75,6 +78,7 @@ class TestGetContactByKeyPrefix:
result = get_contact_by_key_prefix(mock_meshcore_client, "xyz789")
assert result is None
+ mock_meshcore_client.get_contact_by_key_prefix.assert_called_once_with("xyz789")
def test_returns_none_when_method_not_available(self):
"""Returns None when get_contact_by_key_prefix method not available."""
@@ -95,6 +99,7 @@ class TestGetContactByKeyPrefix:
result = get_contact_by_key_prefix(mock_meshcore_client, "abc123")
assert result is None
+ mock_meshcore_client.get_contact_by_key_prefix.assert_called_once_with("abc123")
class TestExtractContactInfo:
@@ -126,13 +131,14 @@ class TestExtractContactInfo:
"""Extracts info from object-based contact."""
from meshmon.meshcore_client import extract_contact_info
- contact = MagicMock()
- contact.adv_name = "TestNode"
- contact.name = "test"
- contact.pubkey_prefix = "abc123"
- contact.public_key = "abc123def456"
- contact.type = 1
- contact.flags = 0
+ contact = SimpleNamespace(
+ adv_name="TestNode",
+ name="test",
+ pubkey_prefix="abc123",
+ public_key="abc123def456",
+ type=1,
+ flags=0,
+ )
result = extract_contact_info(contact)
@@ -158,14 +164,10 @@ class TestExtractContactInfo:
"""Converts bytes values from object attributes to hex."""
from meshmon.meshcore_client import extract_contact_info
- contact = MagicMock()
- contact.adv_name = "TestNode"
- contact.public_key = bytes.fromhex("deadbeef")
- # Make sure other attributes return AttributeError when accessed
- del contact.name
- del contact.pubkey_prefix
- del contact.type
- del contact.flags
+ contact = SimpleNamespace(
+ adv_name="TestNode",
+ public_key=bytes.fromhex("deadbeef"),
+ )
result = extract_contact_info(contact)
@@ -231,13 +233,7 @@ class TestListContactsSummary:
"""Handles mix of dict and object contacts."""
from meshmon.meshcore_client import list_contacts_summary
- obj_contact = MagicMock()
- obj_contact.adv_name = "ObjectNode"
- del obj_contact.name
- del obj_contact.pubkey_prefix
- del obj_contact.public_key
- del obj_contact.type
- del obj_contact.flags
+ obj_contact = SimpleNamespace(adv_name="ObjectNode")
contacts = [
{"adv_name": "DictNode"},
diff --git a/tests/client/test_meshcore_available.py b/tests/client/test_meshcore_available.py
index fbcd64f..4cfc418 100644
--- a/tests/client/test_meshcore_available.py
+++ b/tests/client/test_meshcore_available.py
@@ -1,6 +1,6 @@
"""Tests for MESHCORE_AVAILABLE flag handling."""
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock
import pytest
@@ -28,6 +28,7 @@ class TestMeshcoreAvailableTrue:
assert success is True
assert event_type == "SELF_INFO"
+ assert payload == {"bat": 3850}
assert error is None
@pytest.mark.asyncio
@@ -55,7 +56,7 @@ class TestMeshcoreAvailableTrue:
result = await connect_from_env()
assert result == mock_mc
- mock_meshcore.create_serial.assert_called_once()
+ mock_meshcore.create_serial.assert_called_once_with("/dev/ttyACM0", 115200, debug=False)
class TestMeshcoreAvailableFalse:
@@ -109,28 +110,54 @@ class TestMeshcoreAvailableFalse:
class TestMeshcoreImportFallback:
"""Tests for import fallback behavior."""
- def test_meshcore_none_when_import_fails(self):
+ def test_meshcore_none_when_import_fails(self, monkeypatch):
"""MeshCore is None when import fails."""
- # This tests the behavior defined in the module
- # When meshcore library is not installed, MESHCORE_AVAILABLE should be False
- # and MeshCore/EventType should be None
+ import builtins
+ import importlib
- # We can verify the fallback behavior by checking module-level attributes
- # after patching the import
- with patch.dict("sys.modules", {"meshcore": None}):
- # Importing with meshcore unavailable
- # Note: This test verifies the pattern used in the module
- # The actual import fallback is tested implicitly by the other tests
- pass
+ import meshmon.meshcore_client as module
- def test_event_type_check_handles_none(self, monkeypatch):
+ real_import = builtins.__import__
+
+ def mock_import(name, *args, **kwargs):
+ if name == "meshcore":
+ raise ImportError("No module named 'meshcore'")
+ return real_import(name, *args, **kwargs)
+
+ monkeypatch.setattr(builtins, "__import__", mock_import)
+
+ module = importlib.reload(module)
+
+ assert module.MESHCORE_AVAILABLE is False
+ assert module.MeshCore is None
+ assert module.EventType is None
+
+ monkeypatch.setattr(builtins, "__import__", real_import)
+ importlib.reload(module)
+
+ @pytest.mark.asyncio
+ async def test_event_type_check_handles_none(self, monkeypatch):
"""EventType checks handle None gracefully."""
monkeypatch.setattr("meshmon.meshcore_client.MESHCORE_AVAILABLE", True)
monkeypatch.setattr("meshmon.meshcore_client.EventType", None)
+ from meshmon.meshcore_client import run_command
- # When EventType is None, the error check should be skipped
- # This is tested by the run_command success tests which mock EventType
+ from .conftest import make_mock_event
+
+ event = make_mock_event("SELF_INFO", {"bat": 3850})
+
+ async def cmd():
+ return event
+
+ success, event_type, payload, error = await run_command(
+ MagicMock(), cmd(), "test"
+ )
+
+ assert success is True
+ assert event_type == "SELF_INFO"
+ assert payload == {"bat": 3850}
+ assert error is None
class TestContactFunctionsWithUnavailableMeshcore:
diff --git a/tests/client/test_run_command.py b/tests/client/test_run_command.py
index 345adc0..1f2b8f7 100644
--- a/tests/client/test_run_command.py
+++ b/tests/client/test_run_command.py
@@ -70,7 +70,7 @@ class TestRunCommandSuccess:
mock_meshcore_client, cmd(), "test"
)
- assert "voltage" in payload
+ assert payload == {"voltage": 3.85}
@pytest.mark.asyncio
async def test_converts_namedtuple_payload(self, mock_meshcore_client, monkeypatch):
@@ -91,8 +91,7 @@ class TestRunCommandSuccess:
mock_meshcore_client, cmd(), "test"
)
- assert payload["voltage"] == 3.85
- assert payload["uptime"] == 86400
+ assert payload == {"voltage": 3.85, "uptime": 86400}
class TestRunCommandFailure:
@@ -118,7 +117,9 @@ class TestRunCommandFailure:
cmd_coro.close()
assert success is False
- assert "not available" in error
+ assert event_type is None
+ assert payload is None
+ assert error == "meshcore not available"
@pytest.mark.asyncio
async def test_returns_failure_on_none_event(self, mock_meshcore_client, monkeypatch):
@@ -133,7 +134,7 @@ class TestRunCommandFailure:
)
assert success is False
- assert "No response" in error
+ assert error == "No response received"
@pytest.mark.asyncio
async def test_returns_failure_on_error_event(self, mock_meshcore_client, monkeypatch):
@@ -157,7 +158,9 @@ class TestRunCommandFailure:
)
assert success is False
- assert error is not None
+ assert event_type == "ERROR"
+ assert payload is None
+ assert error == "Command failed"
@pytest.mark.asyncio
async def test_returns_failure_on_timeout(self, mock_meshcore_client, monkeypatch):
@@ -172,7 +175,7 @@ class TestRunCommandFailure:
)
assert success is False
- assert "Timeout" in error
+ assert error == "Timeout"
@pytest.mark.asyncio
async def test_returns_failure_on_exception(self, mock_meshcore_client, monkeypatch):
@@ -187,7 +190,7 @@ class TestRunCommandFailure:
)
assert success is False
- assert "Connection lost" in error
+ assert error == "Connection lost"
class TestRunCommandEventTypeParsing:
@@ -203,11 +206,14 @@ class TestRunCommandEventTypeParsing:
async def cmd():
return event
- _, event_type, _, _ = await run_command(
+ success, event_type, payload, error = await run_command(
mock_meshcore_client, cmd(), "test"
)
+ assert success is True
assert event_type == "CUSTOM_EVENT"
+ assert payload == {}
+ assert error is None
@pytest.mark.asyncio
async def test_falls_back_to_str_type(self, mock_meshcore_client, monkeypatch):
@@ -221,8 +227,11 @@ class TestRunCommandEventTypeParsing:
async def cmd():
return event
- _, event_type, _, _ = await run_command(
+ success, event_type, payload, error = await run_command(
mock_meshcore_client, cmd(), "test"
)
+ assert success is True
assert event_type == "STRING_TYPE"
+ assert payload == {}
+ assert error is None
diff --git a/tests/config/test_config_file.py b/tests/config/test_config_file.py
index 1c5cc85..5c79062 100644
--- a/tests/config/test_config_file.py
+++ b/tests/config/test_config_file.py
@@ -1,10 +1,24 @@
"""Tests for meshcore.conf file parsing."""
-from unittest.mock import patch
+import os
from meshmon.env import _parse_config_value
+def _load_config_from_content(tmp_path, monkeypatch, content: str | None) -> None:
+ import meshmon.env as env
+
+ config_path = tmp_path / "meshcore.conf"
+ if content is not None:
+ config_path.write_text(content)
+
+ fake_env_path = tmp_path / "src" / "meshmon" / "env.py"
+ fake_env_path.parent.mkdir(parents=True, exist_ok=True)
+ fake_env_path.write_text("")
+
+ monkeypatch.setattr(env, "__file__", str(fake_env_path))
+ env._load_config_file()
+
class TestParseConfigValueDetailed:
"""Detailed tests for _parse_config_value."""
@@ -112,15 +126,11 @@ class TestParseConfigValueDetailed:
class TestLoadConfigFileBehavior:
"""Tests for _load_config_file behavior."""
- def test_nonexistent_file_no_error(self, tmp_path, monkeypatch):
+ def test_nonexistent_file_no_error(self, tmp_path, monkeypatch, isolate_config_loading):
"""Missing config file doesn't raise error."""
- # Point to non-existent path
- fake_module_path = tmp_path / "src" / "meshmon" / "env.py"
- fake_module_path.parent.mkdir(parents=True)
- fake_module_path.write_text("")
+ _load_config_from_content(tmp_path, monkeypatch, content=None)
- # No exception should be raised
- # The function checks for existence first
+ assert "MESH_TRANSPORT" not in os.environ
def test_skips_empty_lines(self, tmp_path, monkeypatch, isolate_config_loading):
"""Empty lines are skipped."""
@@ -130,41 +140,38 @@ MESH_TRANSPORT=tcp
MESH_DEBUG=1
"""
- config_path = tmp_path / "meshcore.conf"
- config_path.write_text(config_content)
+ _load_config_from_content(tmp_path, monkeypatch, config_content)
- # Mock the config path location
- with patch("meshmon.env.Path") as mock_path:
- mock_path.return_value.resolve.return_value.parent.parent.parent.__truediv__.return_value = config_path
- mock_path.return_value.resolve.return_value.parent.parent.parent / "meshcore.conf"
- # _load_config_file() would need to be called manually or tested via Config
+ assert os.environ["MESH_TRANSPORT"] == "tcp"
+ assert os.environ["MESH_DEBUG"] == "1"
- def test_skips_comment_lines(self, tmp_path):
+ def test_skips_comment_lines(self, tmp_path, monkeypatch, isolate_config_loading):
"""Lines starting with # are skipped."""
config_content = """# This is a comment
MESH_TRANSPORT=tcp
# Another comment
"""
- config_path = tmp_path / "meshcore.conf"
- config_path.write_text(config_content)
- # The parsing logic skips lines starting with #
+ _load_config_from_content(tmp_path, monkeypatch, config_content)
- def test_handles_export_prefix(self, tmp_path):
+ assert os.environ["MESH_TRANSPORT"] == "tcp"
+
+ def test_handles_export_prefix(self, tmp_path, monkeypatch, isolate_config_loading):
"""Lines with 'export ' prefix are handled."""
config_content = "export MESH_TRANSPORT=tcp\n"
- config_path = tmp_path / "meshcore.conf"
- config_path.write_text(config_content)
- # The parsing logic removes 'export ' prefix
+ _load_config_from_content(tmp_path, monkeypatch, config_content)
- def test_skips_lines_without_equals(self, tmp_path):
+ assert os.environ["MESH_TRANSPORT"] == "tcp"
+
+ def test_skips_lines_without_equals(self, tmp_path, monkeypatch, isolate_config_loading):
"""Lines without = are skipped."""
config_content = """MESH_TRANSPORT=tcp
this line has no equals
MESH_DEBUG=1
"""
- config_path = tmp_path / "meshcore.conf"
- config_path.write_text(config_content)
- # Invalid lines are skipped
+ _load_config_from_content(tmp_path, monkeypatch, config_content)
+
+ assert os.environ["MESH_TRANSPORT"] == "tcp"
+ assert os.environ["MESH_DEBUG"] == "1"
def test_env_vars_take_precedence(self, tmp_path, monkeypatch, isolate_config_loading):
"""Environment variables override config file values."""
@@ -173,11 +180,9 @@ MESH_DEBUG=1
# Config file has different value
config_content = "MESH_TRANSPORT=serial\n"
- config_path = tmp_path / "meshcore.conf"
- config_path.write_text(config_content)
+ _load_config_from_content(tmp_path, monkeypatch, config_content)
# After loading, env var should still be "ble"
- import os
assert os.environ.get("MESH_TRANSPORT") == "ble"
diff --git a/tests/config/test_env.py b/tests/config/test_env.py
index 1516cc5..f49915e 100644
--- a/tests/config/test_env.py
+++ b/tests/config/test_env.py
@@ -35,7 +35,7 @@ class TestGetIntEdgeCases:
assert get_int("TEST_INT", 0) == 42
def test_whitespace_around_number(self, monkeypatch):
- """Whitespace around number causes fallback to default."""
+ """Whitespace around number is tolerated by int()."""
monkeypatch.setenv("TEST_INT", " 42 ")
# Python's int() handles whitespace
assert get_int("TEST_INT", 0) == 42
@@ -50,7 +50,7 @@ class TestGetBoolEdgeCases:
assert get_bool("TEST_BOOL") is True
def test_with_spaces(self, monkeypatch):
- """Spaces are ignored after lowering."""
+ """Whitespace causes a non-match since get_bool does not strip."""
monkeypatch.setenv("TEST_BOOL", " yes ")
# .lower() doesn't strip, so " yes " != "yes"
# This will return False
diff --git a/tests/database/test_db_insert.py b/tests/database/test_db_insert.py
index 8f34022..4012c75 100644
--- a/tests/database/test_db_insert.py
+++ b/tests/database/test_db_insert.py
@@ -1,7 +1,5 @@
"""Tests for database insert functions."""
-import time
-
import pytest
from meshmon.db import (
@@ -10,13 +8,15 @@ from meshmon.db import (
insert_metrics,
)
+BASE_TS = 1704067200
+
class TestInsertMetric:
"""Tests for insert_metric function."""
def test_inserts_single_metric(self, initialized_db):
"""Inserts a single metric successfully."""
- ts = int(time.time())
+ ts = BASE_TS
result = insert_metric(ts, "companion", "battery_mv", 3850.0, initialized_db)
@@ -33,7 +33,7 @@ class TestInsertMetric:
def test_returns_false_on_duplicate(self, initialized_db):
"""Returns False for duplicate (ts, role, metric) tuple."""
- ts = int(time.time())
+ ts = BASE_TS
# First insert succeeds
assert insert_metric(ts, "companion", "test", 1.0, initialized_db) is True
@@ -43,28 +43,28 @@ class TestInsertMetric:
def test_different_roles_not_duplicate(self, initialized_db):
"""Same ts/metric with different roles are not duplicates."""
- ts = int(time.time())
+ ts = BASE_TS
assert insert_metric(ts, "companion", "test", 1.0, initialized_db) is True
assert insert_metric(ts, "repeater", "test", 2.0, initialized_db) is True
def test_different_metrics_not_duplicate(self, initialized_db):
"""Same ts/role with different metrics are not duplicates."""
- ts = int(time.time())
+ ts = BASE_TS
assert insert_metric(ts, "companion", "test1", 1.0, initialized_db) is True
assert insert_metric(ts, "companion", "test2", 2.0, initialized_db) is True
def test_invalid_role_raises(self, initialized_db):
"""Invalid role raises ValueError."""
- ts = int(time.time())
+ ts = BASE_TS
with pytest.raises(ValueError, match="Invalid role"):
insert_metric(ts, "invalid", "test", 1.0, initialized_db)
def test_sql_injection_blocked(self, initialized_db):
"""SQL injection attempt raises ValueError."""
- ts = int(time.time())
+ ts = BASE_TS
with pytest.raises(ValueError, match="Invalid role"):
insert_metric(ts, "'; DROP TABLE metrics; --", "test", 1.0, initialized_db)
@@ -75,7 +75,7 @@ class TestInsertMetrics:
def test_inserts_multiple_metrics(self, initialized_db):
"""Inserts multiple metrics from dict."""
- ts = int(time.time())
+ ts = BASE_TS
metrics = {
"battery_mv": 3850.0,
"contacts": 5,
@@ -95,7 +95,7 @@ class TestInsertMetrics:
def test_returns_insert_count(self, initialized_db):
"""Returns correct count of inserted metrics."""
- ts = int(time.time())
+ ts = BASE_TS
metrics = {"a": 1.0, "b": 2.0, "c": 3.0}
count = insert_metrics(ts, "companion", metrics, initialized_db)
@@ -104,7 +104,7 @@ class TestInsertMetrics:
def test_skips_non_numeric_values(self, initialized_db):
"""Non-numeric values are silently skipped."""
- ts = int(time.time())
+ ts = BASE_TS
metrics = {
"battery_mv": 3850.0, # Numeric - inserted
"name": "test", # String - skipped
@@ -119,7 +119,7 @@ class TestInsertMetrics:
def test_handles_int_and_float(self, initialized_db):
"""Both int and float values are inserted."""
- ts = int(time.time())
+ ts = BASE_TS
metrics = {
"int_value": 42,
"float_value": 3.14,
@@ -131,7 +131,7 @@ class TestInsertMetrics:
def test_converts_int_to_float(self, initialized_db):
"""Integer values are stored as float."""
- ts = int(time.time())
+ ts = BASE_TS
metrics = {"contacts": 5}
insert_metrics(ts, "companion", metrics, initialized_db)
@@ -146,7 +146,7 @@ class TestInsertMetrics:
def test_empty_dict_returns_zero(self, initialized_db):
"""Empty dict returns 0."""
- ts = int(time.time())
+ ts = BASE_TS
count = insert_metrics(ts, "companion", {}, initialized_db)
@@ -154,7 +154,7 @@ class TestInsertMetrics:
def test_skips_duplicates_silently(self, initialized_db):
"""Duplicate metrics are skipped without error."""
- ts = int(time.time())
+ ts = BASE_TS
metrics = {"test": 1.0}
# First insert
@@ -167,7 +167,7 @@ class TestInsertMetrics:
def test_partial_duplicates(self, initialized_db):
"""Partial duplicates: some inserted, some skipped."""
- ts = int(time.time())
+ ts = BASE_TS
# First insert
insert_metrics(ts, "companion", {"existing": 1.0}, initialized_db)
@@ -183,25 +183,25 @@ class TestInsertMetrics:
def test_invalid_role_raises(self, initialized_db):
"""Invalid role raises ValueError."""
- ts = int(time.time())
+ ts = BASE_TS
with pytest.raises(ValueError, match="Invalid role"):
insert_metrics(ts, "invalid", {"test": 1.0}, initialized_db)
def test_companion_metrics(self, initialized_db, sample_companion_metrics):
"""Inserts companion metrics dict."""
- ts = int(time.time())
+ ts = BASE_TS
count = insert_metrics(ts, "companion", sample_companion_metrics, initialized_db)
# Should insert all numeric fields
- assert count >= 4 # At least battery_mv, uptime_secs, contacts, recv, sent
+ assert count == len(sample_companion_metrics)
def test_repeater_metrics(self, initialized_db, sample_repeater_metrics):
"""Inserts repeater metrics dict."""
- ts = int(time.time())
+ ts = BASE_TS
count = insert_metrics(ts, "repeater", sample_repeater_metrics, initialized_db)
# Should insert all numeric fields
- assert count >= 10 # Many metrics
+ assert count == len(sample_repeater_metrics)
diff --git a/tests/database/test_db_maintenance.py b/tests/database/test_db_maintenance.py
index d012de6..fb68d2d 100644
--- a/tests/database/test_db_maintenance.py
+++ b/tests/database/test_db_maintenance.py
@@ -26,16 +26,24 @@ class TestVacuumDb:
# Should not raise
vacuum_db(initialized_db)
- def test_runs_analyze(self, initialized_db, capfd):
+ def test_runs_analyze(self, initialized_db):
"""ANALYZE should be run after VACUUM."""
+ conn = sqlite3.connect(initialized_db)
+ conn.execute(
+ "INSERT INTO metrics (ts, role, metric, value) VALUES (1, 'companion', 'test', 1.0)"
+ )
+ conn.commit()
+ conn.close()
+
# Vacuum includes ANALYZE
vacuum_db(initialized_db)
# Check that database stats were updated
conn = sqlite3.connect(initialized_db)
- conn.execute("SELECT * FROM sqlite_stat1")
- # After ANALYZE, sqlite_stat1 should have entries if tables have data
+ cursor = conn.execute("SELECT COUNT(*) FROM sqlite_stat1")
+ count = cursor.fetchone()[0]
conn.close()
+ assert count > 0
def test_uses_default_path_when_none(self, configured_env, monkeypatch):
"""Uses get_db_path() when no path provided."""
diff --git a/tests/database/test_db_queries.py b/tests/database/test_db_queries.py
index 832c79f..bf38fca 100644
--- a/tests/database/test_db_queries.py
+++ b/tests/database/test_db_queries.py
@@ -1,7 +1,5 @@
"""Tests for database query functions."""
-import time
-
import pytest
from meshmon.db import (
@@ -13,13 +11,15 @@ from meshmon.db import (
insert_metrics,
)
+BASE_TS = 1704067200
+
class TestGetMetricsForPeriod:
"""Tests for get_metrics_for_period function."""
def test_returns_dict_by_metric(self, initialized_db):
"""Returns dict with metric names as keys."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {
"battery_mv": 3850.0,
"contacts": 5,
@@ -35,7 +35,7 @@ class TestGetMetricsForPeriod:
def test_returns_timestamp_value_tuples(self, initialized_db):
"""Each metric has list of (ts, value) tuples."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
result = get_metrics_for_period(
@@ -47,7 +47,7 @@ class TestGetMetricsForPeriod:
def test_sorted_by_timestamp(self, initialized_db):
"""Results are sorted by timestamp ascending."""
- base_ts = int(time.time())
+ base_ts = BASE_TS
# Insert out of order
insert_metrics(base_ts + 200, "companion", {"test": 3.0}, initialized_db)
@@ -63,7 +63,7 @@ class TestGetMetricsForPeriod:
def test_respects_time_range(self, initialized_db):
"""Only returns data within specified time range."""
- base_ts = int(time.time())
+ base_ts = BASE_TS
insert_metrics(base_ts - 200, "companion", {"test": 1.0}, initialized_db) # Outside
insert_metrics(base_ts, "companion", {"test": 2.0}, initialized_db) # Inside
@@ -78,7 +78,7 @@ class TestGetMetricsForPeriod:
def test_filters_by_role(self, initialized_db):
"""Only returns data for specified role."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
insert_metrics(ts, "repeater", {"test": 2.0}, initialized_db)
@@ -90,7 +90,7 @@ class TestGetMetricsForPeriod:
def test_computes_bat_pct(self, initialized_db):
"""Computes bat_pct from battery voltage."""
- ts = int(time.time())
+ ts = BASE_TS
# 4200 mV = 4.2V = 100%
insert_metrics(ts, "companion", {"battery_mv": 4200.0}, initialized_db)
@@ -103,7 +103,7 @@ class TestGetMetricsForPeriod:
def test_bat_pct_for_repeater(self, initialized_db):
"""Computes bat_pct for repeater using 'bat' field."""
- ts = int(time.time())
+ ts = BASE_TS
# 3000 mV = 3.0V = 0%
insert_metrics(ts, "repeater", {"bat": 3000.0}, initialized_db)
@@ -133,7 +133,7 @@ class TestGetLatestMetrics:
def test_returns_most_recent(self, initialized_db):
"""Returns metrics at most recent timestamp."""
- base_ts = int(time.time())
+ base_ts = BASE_TS
insert_metrics(base_ts, "companion", {"test": 1.0}, initialized_db)
insert_metrics(base_ts + 100, "companion", {"test": 2.0}, initialized_db)
@@ -145,7 +145,7 @@ class TestGetLatestMetrics:
def test_includes_ts(self, initialized_db):
"""Result includes 'ts' key with timestamp."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
result = get_latest_metrics("companion", initialized_db)
@@ -155,7 +155,7 @@ class TestGetLatestMetrics:
def test_includes_all_metrics(self, initialized_db):
"""Result includes all metrics at that timestamp."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {
"battery_mv": 3850.0,
"contacts": 5,
@@ -170,7 +170,7 @@ class TestGetLatestMetrics:
def test_computes_bat_pct(self, initialized_db):
"""Computes bat_pct from battery voltage."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"battery_mv": 3820.0}, initialized_db)
result = get_latest_metrics("companion", initialized_db)
@@ -186,7 +186,7 @@ class TestGetLatestMetrics:
def test_filters_by_role(self, initialized_db):
"""Only returns data for specified role."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
insert_metrics(ts + 100, "repeater", {"test": 2.0}, initialized_db)
@@ -206,7 +206,7 @@ class TestGetMetricCount:
def test_counts_rows(self, initialized_db):
"""Counts total metric rows for role."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"a": 1.0, "b": 2.0, "c": 3.0}, initialized_db)
count = get_metric_count("companion", initialized_db)
@@ -215,7 +215,7 @@ class TestGetMetricCount:
def test_filters_by_role(self, initialized_db):
"""Only counts rows for specified role."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"a": 1.0}, initialized_db)
insert_metrics(ts, "repeater", {"b": 2.0, "c": 3.0}, initialized_db)
@@ -238,7 +238,7 @@ class TestGetDistinctTimestamps:
def test_counts_unique_timestamps(self, initialized_db):
"""Counts distinct timestamps."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"a": 1.0, "b": 2.0}, initialized_db) # 1 ts
insert_metrics(ts + 100, "companion", {"a": 3.0}, initialized_db) # 2nd ts
@@ -248,7 +248,7 @@ class TestGetDistinctTimestamps:
def test_filters_by_role(self, initialized_db):
"""Only counts timestamps for specified role."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"a": 1.0}, initialized_db)
insert_metrics(ts + 100, "companion", {"a": 2.0}, initialized_db)
insert_metrics(ts, "repeater", {"a": 3.0}, initialized_db)
@@ -267,7 +267,7 @@ class TestGetAvailableMetrics:
def test_returns_metric_names(self, initialized_db):
"""Returns list of distinct metric names."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {
"battery_mv": 3850.0,
"contacts": 5,
@@ -282,7 +282,7 @@ class TestGetAvailableMetrics:
def test_sorted_alphabetically(self, initialized_db):
"""Metrics are sorted alphabetically."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {
"zebra": 1.0,
"apple": 2.0,
@@ -295,7 +295,7 @@ class TestGetAvailableMetrics:
def test_filters_by_role(self, initialized_db):
"""Only returns metrics for specified role."""
- ts = int(time.time())
+ ts = BASE_TS
insert_metrics(ts, "companion", {"companion_metric": 1.0}, initialized_db)
insert_metrics(ts, "repeater", {"repeater_metric": 2.0}, initialized_db)
diff --git a/tests/html/test_jinja_env.py b/tests/html/test_jinja_env.py
index ff61ac6..7559707 100644
--- a/tests/html/test_jinja_env.py
+++ b/tests/html/test_jinja_env.py
@@ -1,5 +1,7 @@
"""Tests for Jinja2 environment and custom filters."""
+import re
+
import pytest
from jinja2 import Environment
@@ -32,9 +34,7 @@ class TestGetJinjaEnv:
"""Returns the same environment instance (cached)."""
env1 = get_jinja_env()
env2 = get_jinja_env()
- # Implementation may or may not cache - just verify both work
- assert env1 is not None
- assert env2 is not None
+ assert env1 is env2
class TestJinjaFilters:
@@ -54,16 +54,14 @@ class TestJinjaFilters:
template = env.from_string("{{ value|format_number }}")
result = template.render(value=1234567)
- # Should have some separator
- assert "1234567" not in result or len(result) > 7
+ assert result == "1,234,567"
def test_format_number_handles_none(self, env):
"""format_number handles None gracefully."""
template = env.from_string("{{ value|format_number }}")
result = template.render(value=None)
- # Should return dash or empty string for None
- assert result in ["-", "N/A", "None", "", " - "]
+ assert result == "N/A"
def test_format_time_filter_exists(self, env):
"""format_time filter is registered."""
@@ -73,19 +71,16 @@ class TestJinjaFilters:
"""format_time formats Unix timestamp."""
template = env.from_string("{{ value|format_time }}")
- # Use a recent timestamp
- import time
- ts = int(time.time()) - 3600
+ ts = 1704067200
result = template.render(value=ts)
- # Should produce some formatted time string
- assert len(result) > 0
+ assert re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", result)
def test_format_time_handles_none(self, env):
"""format_time handles None gracefully."""
template = env.from_string("{{ value|format_time }}")
result = template.render(value=None)
- assert result in ["-", "N/A", "None", "", " - "]
+ assert result == "N/A"
def test_format_uptime_filter_exists(self, env):
"""format_uptime filter is registered."""
@@ -97,8 +92,7 @@ class TestJinjaFilters:
# 1 day, 2 hours, 30 minutes = 95400 seconds
result = template.render(value=95400)
- # Should produce some duration output
- assert len(result) > 0
+ assert result == "1d 2h 30m"
def test_format_duration_filter_exists(self, env):
"""format_duration filter is registered."""
diff --git a/tests/html/test_metrics_builders.py b/tests/html/test_metrics_builders.py
index e356979..01d9f3e 100644
--- a/tests/html/test_metrics_builders.py
+++ b/tests/html/test_metrics_builders.py
@@ -96,8 +96,8 @@ class TestBuildNodeDetails:
result = build_node_details("repeater")
# Should have hardware in one of the items
- labels = [item.get("label", "").lower() for item in result]
- assert "hardware" in labels
+ hardware = next(item for item in result if item.get("label") == "Hardware")
+ assert hardware["value"] == "Test LoRa Device"
def test_different_roles(self, configured_env):
"""Different roles return details."""
@@ -132,8 +132,8 @@ class TestBuildRadioConfig:
result = build_radio_config()
- values = [item.get("value", "") for item in result]
- assert any("869" in str(v) for v in values)
+ freq = next(item for item in result if item.get("label") == "Frequency")
+ assert freq["value"] == "869.618 MHz"
def test_handles_missing_config(self, configured_env):
"""Returns list even with default config."""
@@ -167,6 +167,9 @@ class TestBuildTrafficTableRows:
assert "label" in row
assert "rx" in row
assert "tx" in row
+ assert "rx_raw" in row
+ assert "tx_raw" in row
+ assert "unit" in row
def test_handles_empty_list(self):
"""Handles empty traffic metrics list."""
diff --git a/tests/html/test_page_context.py b/tests/html/test_page_context.py
index 9863d06..212b0ca 100644
--- a/tests/html/test_page_context.py
+++ b/tests/html/test_page_context.py
@@ -1,6 +1,6 @@
"""Tests for page context building."""
-import time
+from datetime import datetime, timedelta
import pytest
@@ -9,41 +9,54 @@ from meshmon.html import (
get_status,
)
+FIXED_NOW = datetime(2024, 1, 1, 12, 0, 0)
+
+
+@pytest.fixture
+def fixed_now(monkeypatch):
+ class FixedDatetime(datetime):
+ @classmethod
+ def now(cls):
+ return FIXED_NOW
+
+ monkeypatch.setattr("meshmon.html.datetime", FixedDatetime)
+ return FIXED_NOW
+
class TestGetStatus:
"""Tests for get_status function."""
- def test_online_for_recent_data(self):
+ def test_online_for_recent_data(self, fixed_now):
"""Returns 'online' for data less than 30 minutes old."""
# 10 minutes ago
- recent_ts = int(time.time()) - 600
+ recent_ts = int(fixed_now.timestamp()) - 600
status_class, status_label = get_status(recent_ts)
assert status_class == "online"
- def test_stale_for_medium_age_data(self):
+ def test_stale_for_medium_age_data(self, fixed_now):
"""Returns 'stale' for data 30 minutes to 2 hours old."""
# 1 hour ago
- medium_ts = int(time.time()) - 3600
+ medium_ts = int(fixed_now.timestamp()) - 3600
status_class, status_label = get_status(medium_ts)
assert status_class == "stale"
- def test_offline_for_old_data(self):
+ def test_offline_for_old_data(self, fixed_now):
"""Returns 'offline' for data more than 2 hours old."""
# 3 hours ago
- old_ts = int(time.time()) - 10800
+ old_ts = int(fixed_now.timestamp()) - 10800
status_class, status_label = get_status(old_ts)
assert status_class == "offline"
- def test_offline_for_very_old_data(self):
+ def test_offline_for_very_old_data(self, fixed_now):
"""Returns 'offline' for very old data."""
# 7 days ago
- very_old_ts = int(time.time()) - 604800
+ very_old_ts = int(fixed_now.timestamp()) - int(timedelta(days=7).total_seconds())
status_class, status_label = get_status(very_old_ts)
@@ -61,41 +74,39 @@ class TestGetStatus:
assert status_class == "offline"
- def test_online_for_current_time(self):
+ def test_online_for_current_time(self, fixed_now):
"""Returns 'online' for current timestamp."""
- now_ts = int(time.time())
+ now_ts = int(fixed_now.timestamp())
status_class, status_label = get_status(now_ts)
assert status_class == "online"
- def test_boundary_30_minutes(self):
+ def test_boundary_30_minutes(self, fixed_now):
"""Tests boundary at exactly 30 minutes."""
# Exactly 30 minutes ago
- boundary_ts = int(time.time()) - 1800
+ boundary_ts = int(fixed_now.timestamp()) - 1800
status_class, _ = get_status(boundary_ts)
- # At boundary, could be either online or stale depending on implementation
- assert status_class in ["online", "stale"]
+ assert status_class == "stale"
- def test_boundary_2_hours(self):
+ def test_boundary_2_hours(self, fixed_now):
"""Tests boundary at exactly 2 hours."""
# Exactly 2 hours ago
- boundary_ts = int(time.time()) - 7200
+ boundary_ts = int(fixed_now.timestamp()) - 7200
status_class, _ = get_status(boundary_ts)
- # At boundary, could be either stale or offline
- assert status_class in ["stale", "offline"]
+ assert status_class == "offline"
- def test_returns_tuple(self):
+ def test_returns_tuple(self, fixed_now):
"""Returns tuple of (status_class, status_label)."""
- status = get_status(int(time.time()))
+ status = get_status(int(fixed_now.timestamp()))
assert isinstance(status, tuple)
assert len(status) == 2
- def test_status_label_is_string(self):
+ def test_status_label_is_string(self, fixed_now):
"""Status label is a string."""
- _, status_label = get_status(int(time.time()))
+ _, status_label = get_status(int(fixed_now.timestamp()))
assert isinstance(status_label, str)
@@ -103,10 +114,10 @@ class TestBuildPageContext:
"""Tests for build_page_context function."""
@pytest.fixture
- def sample_row(self, sample_repeater_metrics):
+ def sample_row(self, sample_repeater_metrics, fixed_now):
"""Create a sample row with timestamp."""
row = sample_repeater_metrics.copy()
- row["ts"] = int(time.time()) - 300 # 5 minutes ago
+ row["ts"] = int(fixed_now.timestamp()) - 300 # 5 minutes ago
return row
def test_returns_dict(self, configured_env, sample_row):
@@ -141,8 +152,7 @@ class TestBuildPageContext:
at_root=True,
)
- assert "status_class" in context
- assert context["status_class"] in ["online", "stale", "offline"]
+ assert context["status_class"] == "online"
def test_handles_none_row(self, configured_env):
"""Handles None row gracefully."""
@@ -183,10 +193,10 @@ class TestBuildPageContext:
assert "period" in context
assert context["period"] == "day"
- def test_different_roles(self, configured_env, sample_row, sample_companion_metrics):
+ def test_different_roles(self, configured_env, sample_row, sample_companion_metrics, fixed_now):
"""Context varies by role."""
companion_row = sample_companion_metrics.copy()
- companion_row["ts"] = int(time.time()) - 300
+ companion_row["ts"] = int(fixed_now.timestamp()) - 300
repeater_context = build_page_context(
role="repeater",
@@ -219,6 +229,5 @@ class TestBuildPageContext:
at_root=False,
)
- # Non-root pages need relative path to CSS
- assert "css_path" in root_context or "at_root" in root_context
- assert "css_path" in non_root_context or "at_root" in non_root_context
+ assert root_context["css_path"] == "/"
+ assert non_root_context["css_path"] == "../"
diff --git a/tests/html/test_reports_index.py b/tests/html/test_reports_index.py
index d471ede..1615e3f 100644
--- a/tests/html/test_reports_index.py
+++ b/tests/html/test_reports_index.py
@@ -56,14 +56,13 @@ class TestRenderReportsIndex:
"""Index page includes title."""
html = render_reports_index(sample_report_sections)
- assert "" in html
- assert "Report" in html or "report" in html
+ assert "Reports Archive" in html
def test_includes_year(self, configured_env, sample_report_sections):
"""Lists available years."""
html = render_reports_index(sample_report_sections)
- assert "2024" in html
+ assert "/reports/repeater/2024/" in html
def test_handles_empty_sections(self, configured_env):
"""Handles empty report sections."""
@@ -76,8 +75,21 @@ class TestRenderReportsIndex:
"""Includes role names in output."""
html = render_reports_index(sample_report_sections)
- # Should mention both roles
- assert "repeater" in html.lower() or "Repeater" in html
+ assert "Repeater" in html
+ assert "Companion" in html
+
+ def test_includes_descriptions(self, configured_env, sample_report_sections, monkeypatch):
+ """Includes role descriptions from config."""
+ monkeypatch.setenv("REPEATER_DISPLAY_NAME", "Alpha Repeater")
+ monkeypatch.setenv("COMPANION_DISPLAY_NAME", "Beta Node")
+ monkeypatch.setenv("REPORT_LOCATION_SHORT", "Test Ridge")
+ import meshmon.env
+ meshmon.env._config = None
+
+ html = render_reports_index(sample_report_sections)
+
+ assert "Alpha Repeater — Remote node in Test Ridge" in html
+ assert "Beta Node — Local USB-connected node" in html
def test_includes_css_reference(self, configured_env, sample_report_sections):
"""Includes reference to stylesheet."""
@@ -95,4 +107,4 @@ class TestRenderReportsIndex:
html = render_reports_index(sections)
assert isinstance(html, str)
- assert "</html>" in html
+ assert "No reports available yet." in html
diff --git a/tests/html/test_write_site.py b/tests/html/test_write_site.py
index 9879df6..b4533a1 100644
--- a/tests/html/test_write_site.py
+++ b/tests/html/test_write_site.py
@@ -1,6 +1,5 @@
"""Tests for write_site and related output functions."""
-
import pytest
from meshmon.db import get_latest_metrics
@@ -9,9 +8,96 @@ from meshmon.html import (
write_site,
)
+BASE_TS = 1704067200
+
+
+def _sample_companion_metrics() -> dict[str, float]:
+ return {
+ "battery_mv": 3850.0,
+ "uptime_secs": 86400.0,
+ "contacts": 5.0,
+ "recv": 1234.0,
+ "sent": 567.0,
+ "errors": 0.0,
+ }
+
+
+def _sample_repeater_metrics() -> dict[str, float]:
+ return {
+ "bat": 3920.0,
+ "uptime": 172800.0,
+ "last_rssi": -85.0,
+ "last_snr": 7.5,
+ "noise_floor": -115.0,
+ "tx_queue_len": 0.0,
+ "nb_recv": 5678.0,
+ "nb_sent": 2345.0,
+ "airtime": 3600.0,
+ "rx_airtime": 7200.0,
+ "flood_dups": 12.0,
+ "direct_dups": 5.0,
+ "sent_flood": 100.0,
+ "recv_flood": 200.0,
+ "sent_direct": 50.0,
+ "recv_direct": 75.0,
+ }
+
+
+@pytest.fixture(scope="module")
+def html_db_cache(tmp_path_factory):
+ """Create and populate a shared DB once for HTML write_site tests."""
+ from meshmon.db import init_db, insert_metrics
+
+ root_dir = tmp_path_factory.mktemp("html-db")
+ state_dir = root_dir / "state"
+ state_dir.mkdir()
+
+ db_path = state_dir / "metrics.db"
+ init_db(db_path=db_path)
+
+ now = BASE_TS
+ day_seconds = 86400
+
+ sample_companion_metrics = _sample_companion_metrics()
+ sample_repeater_metrics = _sample_repeater_metrics()
+
+ # Insert 7 days of companion data (every hour)
+ for day in range(7):
+ for hour in range(24):
+ ts = now - (day * day_seconds) - (hour * 3600)
+ metrics = sample_companion_metrics.copy()
+ metrics["battery_mv"] = 3700 + (hour * 10) + (day * 5)
+ metrics["recv"] = 100 * (day + 1) + hour
+ metrics["sent"] = 50 * (day + 1) + hour
+ insert_metrics(ts, "companion", metrics, db_path=db_path)
+
+ # Insert 7 days of repeater data (every 15 minutes)
+ for day in range(7):
+ for interval in range(96): # 24 * 4
+ ts = now - (day * day_seconds) - (interval * 900)
+ metrics = sample_repeater_metrics.copy()
+ metrics["bat"] = 3700 + (interval * 2) + (day * 5)
+ metrics["nb_recv"] = 1000 * (day + 1) + interval * 10
+ metrics["nb_sent"] = 500 * (day + 1) + interval * 5
+ insert_metrics(ts, "repeater", metrics, db_path=db_path)
+
+ return {"state_dir": state_dir, "db_path": db_path}
+
@pytest.fixture
-def metrics_rows(populated_db):
+def html_env(html_db_cache, tmp_out_dir, monkeypatch):
+ """Env with shared DB and per-test output directory."""
+ monkeypatch.setenv("STATE_DIR", str(html_db_cache["state_dir"]))
+ monkeypatch.setenv("OUT_DIR", str(tmp_out_dir))
+
+ import meshmon.env
+ meshmon.env._config = None
+
+ return {"state_dir": html_db_cache["state_dir"], "out_dir": tmp_out_dir}
+
+
+@pytest.fixture
+def metrics_rows(html_env):
"""Get latest metrics rows for both roles."""
companion_row = get_latest_metrics("companion")
repeater_row = get_latest_metrics("repeater")
@@ -21,17 +107,17 @@ def metrics_rows(populated_db):
class TestWriteSite:
"""Tests for write_site function."""
- def test_creates_output_directory(self, configured_env, metrics_rows):
+ def test_creates_output_directory(self, html_env, metrics_rows):
"""Creates output directory if it doesn't exist."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
assert out_dir.exists()
- def test_generates_repeater_pages(self, configured_env, metrics_rows):
+ def test_generates_repeater_pages(self, html_env, metrics_rows):
"""Generates repeater HTML pages at root."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
@@ -39,9 +125,9 @@ class TestWriteSite:
for period in ["day", "week", "month", "year"]:
assert (out_dir / f"{period}.html").exists()
- def test_generates_companion_pages(self, configured_env, metrics_rows):
+ def test_generates_companion_pages(self, html_env, metrics_rows):
"""Generates companion HTML pages in subdirectory."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
@@ -50,9 +136,9 @@ class TestWriteSite:
for period in ["day", "week", "month", "year"]:
assert (companion_dir / f"{period}.html").exists()
- def test_html_files_are_valid(self, configured_env, metrics_rows):
+ def test_html_files_are_valid(self, html_env, metrics_rows):
"""Generated HTML files have valid structure."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
@@ -76,9 +162,9 @@ class TestWriteSite:
class TestCopyStaticAssets:
"""Tests for copy_static_assets function."""
- def test_copies_css(self, configured_env):
+ def test_copies_css(self, html_env):
"""Copies CSS stylesheet."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
out_dir.mkdir(parents=True, exist_ok=True)
copy_static_assets()
@@ -86,9 +172,9 @@ class TestCopyStaticAssets:
css_file = out_dir / "styles.css"
assert css_file.exists()
- def test_copies_javascript(self, configured_env):
+ def test_copies_javascript(self, html_env):
"""Copies JavaScript files."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
out_dir.mkdir(parents=True, exist_ok=True)
copy_static_assets()
@@ -96,9 +182,9 @@ class TestCopyStaticAssets:
js_file = out_dir / "chart-tooltip.js"
assert js_file.exists()
- def test_css_is_valid(self, configured_env):
+ def test_css_is_valid(self, html_env):
"""Copied CSS has expected content."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
out_dir.mkdir(parents=True, exist_ok=True)
copy_static_assets()
@@ -106,12 +192,11 @@ class TestCopyStaticAssets:
css_file = out_dir / "styles.css"
content = css_file.read_text()
- # Should have CSS variables
- assert "--" in content or "{" in content
+ assert "--bg-primary" in content
- def test_requires_output_directory(self, configured_env):
+ def test_requires_output_directory(self, html_env):
"""Requires output directory to exist."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
# Ensure out_dir exists
out_dir.mkdir(parents=True, exist_ok=True)
@@ -120,9 +205,9 @@ class TestCopyStaticAssets:
assert (out_dir / "styles.css").exists()
- def test_overwrites_existing(self, configured_env):
+ def test_overwrites_existing(self, html_env):
"""Overwrites existing static files."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
out_dir.mkdir(parents=True, exist_ok=True)
# Create a fake CSS file
@@ -133,15 +218,15 @@ class TestCopyStaticAssets:
# Should be overwritten with real content
content = css_file.read_text()
- assert "/* fake */" not in content or len(content) > 20
-
+ assert content != "/* fake */"
+
class TestHtmlOutput:
"""Tests for HTML output structure."""
- def test_pages_include_navigation(self, configured_env, metrics_rows):
+ def test_pages_include_navigation(self, html_env, metrics_rows):
"""HTML pages include navigation."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
@@ -151,9 +236,9 @@ class TestHtmlOutput:
assert "week" in content.lower()
assert "month" in content.lower()
- def test_pages_include_meta_tags(self, configured_env, metrics_rows):
+ def test_pages_include_meta_tags(self, html_env, metrics_rows):
"""HTML pages include meta tags."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
@@ -162,9 +247,9 @@ class TestHtmlOutput:
assert "<meta" in content
assert "charset" in content.lower() or "utf-8" in content.lower()
- def test_pages_include_title(self, configured_env, metrics_rows):
+ def test_pages_include_title(self, html_env, metrics_rows):
"""HTML pages include title tag."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
@@ -173,9 +258,9 @@ class TestHtmlOutput:
assert "<title>" in content
assert "" in content
- def test_pages_reference_css(self, configured_env, metrics_rows):
+ def test_pages_reference_css(self, html_env, metrics_rows):
"""HTML pages reference stylesheet."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
@@ -183,9 +268,9 @@ class TestHtmlOutput:
assert "styles.css" in content
- def test_companion_pages_relative_css(self, configured_env, metrics_rows):
+ def test_companion_pages_relative_css(self, html_env, metrics_rows):
"""Companion pages use relative path to CSS."""
- out_dir = configured_env["out_dir"]
+ out_dir = html_env["out_dir"]
write_site(metrics_rows["companion"], metrics_rows["repeater"])
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index b3836ea..863e31e 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -1,46 +1,222 @@
"""Integration test fixtures."""
+import os
import time
from unittest.mock import AsyncMock, MagicMock
import pytest
+_INTEGRATION_ENV = {
+ "REPORT_LOCATION_NAME": "Test Location",
+ "REPORT_LOCATION_SHORT": "Test",
+ "REPEATER_DISPLAY_NAME": "Test Repeater",
+ "COMPANION_DISPLAY_NAME": "Test Companion",
+ "MESH_TRANSPORT": "serial",
+ "MESH_SERIAL_PORT": "/dev/ttyACM0",
+}
+RENDERED_CHART_METRICS = {
+ "companion": ["battery_mv", "recv", "contacts"],
+ "repeater": ["bat", "nb_recv", "last_rssi"],
+}
-@pytest.fixture
-def populated_db_with_history(initialized_db, sample_companion_metrics, sample_repeater_metrics):
- """Database populated with 30 days of historical data for integration tests."""
+
+def _sample_companion_metrics() -> dict[str, float]:
+ return {
+ "battery_mv": 3850.0,
+ "uptime_secs": 86400.0,
+ "contacts": 5.0,
+ "recv": 1234.0,
+ "sent": 567.0,
+ "errors": 0.0,
+ }
+
+
+def _sample_repeater_metrics() -> dict[str, float]:
+ return {
+ "bat": 3920.0,
+ "uptime": 172800.0,
+ "last_rssi": -85.0,
+ "last_snr": 7.5,
+ "noise_floor": -115.0,
+ "tx_queue_len": 0.0,
+ "nb_recv": 5678.0,
+ "nb_sent": 2345.0,
+ "airtime": 3600.0,
+ "rx_airtime": 7200.0,
+ "flood_dups": 12.0,
+ "direct_dups": 5.0,
+ "sent_flood": 100.0,
+ "recv_flood": 200.0,
+ "sent_direct": 50.0,
+ "recv_direct": 75.0,
+ }
+
+
+def _populate_db_with_history(
+ db_path,
+ sample_companion_metrics: dict[str, float],
+ sample_repeater_metrics: dict[str, float],
+ days: int = 30,
+ companion_step_seconds: int = 3600,
+ repeater_step_seconds: int = 900,
+) -> None:
from meshmon.db import insert_metrics
now = int(time.time())
day_seconds = 86400
+ companion_steps = day_seconds // companion_step_seconds
+ repeater_steps = day_seconds // repeater_step_seconds
- # Insert 30 days of companion data (every hour)
- for day in range(30):
- for hour in range(24):
- ts = now - (day * day_seconds) - (hour * 3600)
+ # Insert companion data (default: 30 days, hourly)
+ for day in range(days):
+ for step in range(companion_steps):
+ ts = now - (day * day_seconds) - (step * companion_step_seconds)
metrics = sample_companion_metrics.copy()
# Vary values to create realistic patterns
- metrics["battery_mv"] = 3700 + (hour * 5) + (day % 7) * 10
- metrics["recv"] = 100 + day * 10 + hour
- metrics["sent"] = 50 + day * 5 + hour
- metrics["uptime_secs"] = (30 - day) * day_seconds + hour * 3600
- insert_metrics(ts, "companion", metrics, initialized_db)
+ metrics["battery_mv"] = 3700 + (step * 5) + (day % 7) * 10
+ metrics["recv"] = 100 + day * 10 + step
+ metrics["sent"] = 50 + day * 5 + step
+ metrics["uptime_secs"] = (days - day) * day_seconds + step * companion_step_seconds
+ insert_metrics(ts, "companion", metrics, db_path=db_path)
- # Insert 30 days of repeater data (every 15 minutes)
- for day in range(30):
- for interval in range(96): # 24 * 4 = 96 intervals per day
- ts = now - (day * day_seconds) - (interval * 900)
+ # Insert repeater data (default: 30 days, every 15 minutes)
+ for day in range(days):
+ for interval in range(repeater_steps):
+ ts = now - (day * day_seconds) - (interval * repeater_step_seconds)
metrics = sample_repeater_metrics.copy()
# Vary values to create realistic patterns
metrics["bat"] = 3800 + (interval % 24) * 5 + (day % 7) * 10
metrics["nb_recv"] = 1000 + day * 100 + interval
metrics["nb_sent"] = 500 + day * 50 + interval
- metrics["uptime"] = (30 - day) * day_seconds + interval * 900
+ metrics["uptime"] = (days - day) * day_seconds + interval * repeater_step_seconds
metrics["last_rssi"] = -90 + (interval % 20)
metrics["last_snr"] = 5 + (interval % 10) * 0.5
- insert_metrics(ts, "repeater", metrics, initialized_db)
+ insert_metrics(ts, "repeater", metrics, db_path=db_path)
- return initialized_db
+
+@pytest.fixture
+def reports_env(reports_db_cache, tmp_out_dir, monkeypatch):
+ """Integration env wired to the shared reports DB and per-test output."""
+ monkeypatch.setenv("STATE_DIR", str(reports_db_cache["state_dir"]))
+ monkeypatch.setenv("OUT_DIR", str(tmp_out_dir))
+ for key, value in _INTEGRATION_ENV.items():
+ monkeypatch.setenv(key, value)
+
+ import meshmon.env
+ meshmon.env._config = None
+
+ return {
+ "state_dir": reports_db_cache["state_dir"],
+ "out_dir": tmp_out_dir,
+ }
+
+
+@pytest.fixture(scope="session")
+def rendered_chart_metrics():
+ """Minimal chart set to keep integration rendering tests fast."""
+ return RENDERED_CHART_METRICS
+
+
+@pytest.fixture
+def populated_db_with_history(reports_db_cache, reports_env):
+ """Shared database populated with a fixed history window for integration tests."""
+ return reports_db_cache["db_path"]
+
+
+@pytest.fixture(scope="module")
+def reports_db_cache(tmp_path_factory):
+ """Create and populate a shared reports DB once per module."""
+ from meshmon.db import init_db
+
+ root_dir = tmp_path_factory.mktemp("reports-db")
+ state_dir = root_dir / "state"
+ state_dir.mkdir()
+
+ db_path = state_dir / "metrics.db"
+ init_db(db_path=db_path)
+ _populate_db_with_history(
+ db_path,
+ _sample_companion_metrics(),
+ _sample_repeater_metrics(),
+ days=14,
+ companion_step_seconds=7200,
+ repeater_step_seconds=7200,
+ )
+
+ return {
+ "state_dir": state_dir,
+ "db_path": db_path,
+ }
+
+
+@pytest.fixture(scope="module")
+def rendered_charts_cache(tmp_path_factory):
+ """Cache rendered charts once per module to speed up integration tests."""
+ from meshmon.charts import render_all_charts, save_chart_stats
+ from meshmon.db import init_db
+
+ root_dir = tmp_path_factory.mktemp("rendered-charts")
+ state_dir = root_dir / "state"
+ out_dir = root_dir / "out"
+ state_dir.mkdir()
+ out_dir.mkdir()
+
+ env_keys = ["STATE_DIR", "OUT_DIR", *_INTEGRATION_ENV.keys()]
+ previous_env = {key: os.environ.get(key) for key in env_keys}
+
+ os.environ["STATE_DIR"] = str(state_dir)
+ os.environ["OUT_DIR"] = str(out_dir)
+ for key, value in _INTEGRATION_ENV.items():
+ os.environ[key] = value
+
+ import meshmon.env
+ meshmon.env._config = None
+
+ db_path = state_dir / "metrics.db"
+ init_db(db_path=db_path)
+ _populate_db_with_history(
+ db_path,
+ _sample_companion_metrics(),
+ _sample_repeater_metrics(),
+ days=7,
+ companion_step_seconds=3600,
+ repeater_step_seconds=3600,
+ )
+
+ for role in ["companion", "repeater"]:
+ charts, stats = render_all_charts(role, metrics=RENDERED_CHART_METRICS[role])
+ save_chart_stats(role, stats)
+
+ yield {
+ "state_dir": state_dir,
+ "out_dir": out_dir,
+ "db_path": db_path,
+ }
+
+ for key, value in previous_env.items():
+ if value is None:
+ os.environ.pop(key, None)
+ else:
+ os.environ[key] = value
+
+ meshmon.env._config = None
+
+
+@pytest.fixture
+def rendered_charts(rendered_charts_cache, monkeypatch):
+ """Expose cached charts with env wired for per-test access."""
+ state_dir = rendered_charts_cache["state_dir"]
+ out_dir = rendered_charts_cache["out_dir"]
+
+ monkeypatch.setenv("STATE_DIR", str(state_dir))
+ monkeypatch.setenv("OUT_DIR", str(out_dir))
+ for key, value in _INTEGRATION_ENV.items():
+ monkeypatch.setenv(key, value)
+
+ import meshmon.env
+ meshmon.env._config = None
+
+ return rendered_charts_cache
@pytest.fixture
@@ -89,18 +265,9 @@ def mock_meshcore_successful_collection(sample_companion_metrics):
@pytest.fixture
def full_integration_env(configured_env, monkeypatch):
- """Full integration environment with all directories set up.
-
- Builds on top of configured_env from root conftest.py to ensure
- consistent directory paths when used with other fixtures like
- initialized_db and populated_db_with_history.
- """
- monkeypatch.setenv("REPORT_LOCATION_NAME", "Test Location")
- monkeypatch.setenv("REPORT_LOCATION_SHORT", "Test")
- monkeypatch.setenv("REPEATER_DISPLAY_NAME", "Test Repeater")
- monkeypatch.setenv("COMPANION_DISPLAY_NAME", "Test Companion")
- monkeypatch.setenv("MESH_TRANSPORT", "serial")
- monkeypatch.setenv("MESH_SERIAL_PORT", "/dev/ttyACM0")
+ """Full integration environment with per-test directories."""
+ for key, value in _INTEGRATION_ENV.items():
+ monkeypatch.setenv(key, value)
import meshmon.env
meshmon.env._config = None
diff --git a/tests/integration/test_collection_pipeline.py b/tests/integration/test_collection_pipeline.py
index 83432a6..d1b5f25 100644
--- a/tests/integration/test_collection_pipeline.py
+++ b/tests/integration/test_collection_pipeline.py
@@ -5,6 +5,10 @@ from unittest.mock import patch
import pytest
+from tests.scripts.conftest import load_script_module
+
+BASE_TS = 1704067200
+
@pytest.mark.integration
class TestCompanionCollectionPipeline:
@@ -36,12 +40,10 @@ class TestCompanionCollectionPipeline:
# Import and run collection (inline to avoid import issues)
# Note: We import the function directly rather than the script
- import time
-
from meshmon.db import insert_metrics
# Simulate collection logic
- ts = int(time.time())
+ ts = BASE_TS
metrics = {}
async with mock_connect_with_lock() as mc:
@@ -115,22 +117,31 @@ class TestCollectionWithCircuitBreaker:
self, full_integration_env, monkeypatch
):
"""Collection should be skipped when circuit breaker is open."""
- import time
-
from meshmon.retry import CircuitBreaker
# Create an open circuit breaker
state_dir = full_integration_env["state_dir"]
cb = CircuitBreaker(state_dir / "repeater_circuit.json")
- cb.consecutive_failures = 10
- cb.cooldown_until = time.time() + 3600 # 1 hour from now
- cb._save() # Use private method
+ cb.record_failure(max_failures=1, cooldown_s=3600)
# Verify circuit is open
assert cb.is_open() is True
- # Collection should check circuit breaker and skip
- # This tests the pattern used in collect_repeater.py
+ module = load_script_module("collect_repeater.py")
+ connect_called = False
+
+ @asynccontextmanager
+ async def mock_connect_with_lock(*args, **kwargs):
+ nonlocal connect_called
+ connect_called = True
+ yield None
+
+ monkeypatch.setattr(module, "connect_with_lock", mock_connect_with_lock)
+
+ result = await module.collect_repeater()
+
+ assert result == 0
+ assert connect_called is False
@pytest.mark.asyncio
async def test_circuit_breaker_records_failure(self, full_integration_env, monkeypatch):
@@ -157,8 +168,6 @@ class TestCollectionWithCircuitBreaker:
@pytest.mark.asyncio
async def test_circuit_breaker_state_persists(self, full_integration_env):
"""Circuit breaker state should persist to disk."""
- import time
-
from meshmon.retry import CircuitBreaker
state_dir = full_integration_env["state_dir"]
@@ -166,12 +175,10 @@ class TestCollectionWithCircuitBreaker:
# Create and configure circuit breaker
cb1 = CircuitBreaker(state_file)
- cb1.consecutive_failures = 5
- cb1.cooldown_until = time.time() + 1800
- cb1._save() # Use private method
+ cb1.record_failure(max_failures=1, cooldown_s=1800)
# Load in new instance
cb2 = CircuitBreaker(state_file)
- assert cb2.consecutive_failures == 5
+ assert cb2.consecutive_failures == 1
assert cb2.cooldown_until == cb1.cooldown_until
diff --git a/tests/integration/test_rendering_pipeline.py b/tests/integration/test_rendering_pipeline.py
index 3916b8b..3ed3cdd 100644
--- a/tests/integration/test_rendering_pipeline.py
+++ b/tests/integration/test_rendering_pipeline.py
@@ -9,42 +9,21 @@ import pytest
class TestChartRenderingPipeline:
"""Test chart rendering end-to-end."""
- def test_renders_all_chart_periods(self, populated_db_with_history, full_integration_env):
+ def test_renders_all_chart_periods(self, rendered_charts):
"""Should render charts for all periods (day/week/month/year)."""
- from meshmon.charts import render_all_charts, save_chart_stats
- from meshmon.db import get_metric_count
+ out_dir = rendered_charts["out_dir"]
- # Verify data exists
- companion_count = get_metric_count("companion")
- repeater_count = get_metric_count("repeater")
+ for role in ["companion", "repeater"]:
+ assets_dir = out_dir / "assets" / role
+ assert assets_dir.exists()
- assert companion_count > 0
- assert repeater_count > 0
+ for period in ["day", "week", "month", "year"]:
+ period_svgs = list(assets_dir.glob(f"*_{period}_*.svg"))
+ assert period_svgs, f"No {period} charts found for {role}"
- # Render companion charts
- charts, stats = render_all_charts("companion")
- save_chart_stats("companion", stats)
-
- # Should have charts for multiple metrics and periods
- assert len(charts) > 0
- assert len(stats) > 0
-
- # Render repeater charts
- charts, stats = render_all_charts("repeater")
- save_chart_stats("repeater", stats)
-
- assert len(charts) > 0
- assert len(stats) > 0
-
- def test_chart_files_created(self, populated_db_with_history, full_integration_env):
+ def test_chart_files_created(self, rendered_charts):
"""Should create SVG chart files in output directory."""
- from meshmon.charts import render_all_charts, save_chart_stats
-
- out_dir = full_integration_env["out_dir"]
-
- # Render charts
- charts, stats = render_all_charts("repeater")
- save_chart_stats("repeater", stats)
+ out_dir = rendered_charts["out_dir"]
# Check SVG files exist
assets_dir = out_dir / "assets" / "repeater"
@@ -58,13 +37,9 @@ class TestChartRenderingPipeline:
stats_file = assets_dir / "chart_stats.json"
assert stats_file.exists()
- def test_chart_statistics_calculated(self, populated_db_with_history, full_integration_env):
+ def test_chart_statistics_calculated(self, rendered_charts):
"""Should calculate correct statistics for charts."""
- from meshmon.charts import load_chart_stats, render_all_charts, save_chart_stats
-
- # Render charts
- charts, stats = render_all_charts("repeater")
- save_chart_stats("repeater", stats)
+ from meshmon.charts import load_chart_stats
# Load and verify stats
loaded_stats = load_chart_stats("repeater")
@@ -81,24 +56,19 @@ class TestChartRenderingPipeline:
assert "min" in period_stats
assert "max" in period_stats
assert "avg" in period_stats
+ assert "current" in period_stats
@pytest.mark.integration
class TestHtmlRenderingPipeline:
"""Test HTML site rendering end-to-end."""
- def test_renders_site_pages(self, populated_db_with_history, full_integration_env):
+ def test_renders_site_pages(self, rendered_charts):
"""Should render all HTML site pages."""
- from meshmon.charts import render_all_charts, save_chart_stats
from meshmon.db import get_latest_metrics
from meshmon.html import write_site
- out_dir = full_integration_env["out_dir"]
-
- # First render charts (needed for site)
- for role in ["repeater", "companion"]:
- charts, stats = render_all_charts(role)
- save_chart_stats(role, stats)
+ out_dir = rendered_charts["out_dir"]
# Get latest metrics for write_site
companion_row = get_latest_metrics("companion")
@@ -116,6 +86,8 @@ class TestHtmlRenderingPipeline:
# Check companion pages exist
assert (out_dir / "companion" / "day.html").exists()
assert (out_dir / "companion" / "week.html").exists()
+ assert (out_dir / "companion" / "month.html").exists()
+ assert (out_dir / "companion" / "year.html").exists()
def test_copies_static_assets(self, full_integration_env):
"""Should copy static assets (CSS, JS)."""
@@ -129,18 +101,12 @@ class TestHtmlRenderingPipeline:
assert (out_dir / "styles.css").exists()
assert (out_dir / "chart-tooltip.js").exists()
- def test_html_contains_chart_data(self, populated_db_with_history, full_integration_env):
+ def test_html_contains_chart_data(self, rendered_charts):
"""HTML should contain embedded chart SVGs."""
- from meshmon.charts import render_all_charts, save_chart_stats
from meshmon.db import get_latest_metrics
from meshmon.html import write_site
- out_dir = full_integration_env["out_dir"]
-
- # Render charts first
- for role in ["repeater", "companion"]:
- charts, stats = render_all_charts(role)
- save_chart_stats(role, stats)
+ out_dir = rendered_charts["out_dir"]
# Get latest metrics for write_site
companion_row = get_latest_metrics("companion")
@@ -155,22 +121,17 @@ class TestHtmlRenderingPipeline:
# Should contain SVG elements
assert " charts -> HTML."""
def test_full_chain_from_database_to_html(
- self, populated_db_with_history, full_integration_env
+ self, rendered_charts
):
"""Complete chain: database metrics -> charts -> HTML site."""
- from meshmon.charts import render_all_charts, save_chart_stats
from meshmon.db import get_latest_metrics, get_metric_count
from meshmon.html import copy_static_assets, write_site
- out_dir = full_integration_env["out_dir"]
+ out_dir = rendered_charts["out_dir"]
# 1. Verify database has data
assert get_metric_count("repeater") > 0
assert get_metric_count("companion") > 0
- # 2. Render charts for both roles
- total_charts = 0
+ # 2. Verify rendered charts exist for both roles
for role in ["repeater", "companion"]:
- charts, stats = render_all_charts(role)
- save_chart_stats(role, stats)
- total_charts += len(charts)
-
- assert total_charts > 0
+ assets_dir = out_dir / "assets" / role
+ svg_files = list(assets_dir.glob("*.svg"))
+ assert svg_files, f"No charts found for {role}"
# 3. Copy static assets
copy_static_assets()
@@ -234,7 +191,11 @@ class TestFullRenderingChain:
assert "" in html_content or "" in html_content.lower()
assert "