adam/meshcore-stats
1
0
Fork 0
mirror of https://github.com/jorijn/meshcore-stats.git synced 2026-08-07 09:12:47 +02:00
Code Issues Packages Projects Releases Wiki Activity

test: add comprehensive pytest test suite with 95% coverage (#29)

Browse Source 
* test: add comprehensive pytest test suite with 95% coverage

Add full unit and integration test coverage for the meshcore-stats project:

- 1020 tests covering all modules (db, charts, html, reports, client, etc.)
- 95.95% code coverage with pytest-cov (95% threshold enforced)
- GitHub Actions CI workflow for automated testing on push/PR
- Proper mocking of external dependencies (meshcore, serial, filesystem)
- SVG snapshot infrastructure for chart regression testing
- Integration tests for collection and rendering pipelines

Test organization:
- tests/charts/: Chart rendering and statistics
- tests/client/: MeshCore client and connection handling
- tests/config/: Environment and configuration parsing
- tests/database/: SQLite operations and migrations
- tests/html/: HTML generation and Jinja templates
- tests/reports/: Report generation and formatting
- tests/retry/: Circuit breaker and retry logic
- tests/unit/: Pure unit tests for utilities
- tests/integration/: End-to-end pipeline tests

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* chore: add test-engineer agent configuration

Add project-local test-engineer agent for pytest test development,
coverage analysis, and test review tasks.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* docs: comprehensive test suite review with 956 tests analyzed

Conducted thorough review of all 956 test cases across 47 test files:

- Unit Tests: 338 tests (battery, metrics, log, telemetry, env, charts, html, reports, formatters)
- Config Tests: 53 tests (env loading, config file parsing)
- Database Tests: 115 tests (init, insert, queries, migrations, maintenance, validation)
- Retry Tests: 59 tests (circuit breaker, async retries, factory)
- Charts Tests: 76 tests (transforms, statistics, timeseries, rendering, I/O)
- HTML Tests: 81 tests (site generation, Jinja2, metrics builders, reports index)
- Reports Tests: 149 tests (location, JSON/TXT formatting, aggregation, counter totals)
- Client Tests: 63 tests (contacts, connection, meshcore availability, commands)
- Integration Tests: 22 tests (reports, collection, rendering pipelines)

Results:
- Overall Pass Rate: 99.7% (953/956)
- 3 tests marked for improvement (empty test bodies in client tests)
- 0 tests requiring fixes

Key findings documented in test_review/tests.md including quality
observations, F.I.R.S.T. principle adherence, and recommendations.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* test: implement snapshot testing for charts and reports

Add comprehensive snapshot testing infrastructure:

SVG Chart Snapshots:
- Deterministic fixtures with fixed timestamps (2024-01-15 12:00:00)
- Tests for gauge/counter metrics in light/dark themes
- Empty chart and single-point edge cases
- Extended normalize_svg_for_snapshot_full() for reproducible comparisons

TXT Report Snapshots:
- Monthly/yearly report snapshots for repeater and companion
- Empty report handling tests
- Tests in tests/reports/test_snapshots.py

Infrastructure:
- tests/snapshots/conftest.py with shared fixtures
- UPDATE_SNAPSHOTS=1 environment variable for regeneration
- scripts/generate_snapshots.py for batch snapshot generation

Run `UPDATE_SNAPSHOTS=1 pytest tests/charts/test_chart_render.py::TestSvgSnapshots`
to generate initial snapshots.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* test: fix SVG normalization and generate initial snapshots

Fix normalize_svg_for_snapshot() to handle:
- clipPath IDs like id="p47c77a2a6e"
- url(#p...) references
- xlink:href="#p..." references
- <dc:date> timestamps

Generated initial snapshot files:
- 7 SVG chart snapshots (gauge, counter, empty, single-point in light/dark)
- 6 TXT report snapshots (monthly/yearly for repeater/companion + empty)

All 13 snapshot tests now pass.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* test: fix SVG normalization to preserve axis rendering

The SVG normalization was replacing all matplotlib-generated IDs with
the same value, causing duplicate IDs that broke SVG rendering:
- Font glyphs, clipPaths, and tick marks all got id="normalized"
- References couldn't resolve to the correct elements
- X and Y axes failed to render in normalized snapshots

Fix uses type-specific prefixes with sequential numbering:
- glyph_N for font glyphs (DejaVuSans-XX patterns)
- clip_N for clipPath definitions (p[0-9a-f]{8,} patterns)
- tick_N for tick marks (m[0-9a-f]{8,} patterns)

This ensures all IDs remain unique while still being deterministic
for snapshot comparison.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* chore: add coverage and pytest artifacts to gitignore

Add .coverage, .coverage.*, htmlcov/, and .pytest_cache/ to prevent
test artifacts from being committed.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* style: fix all ruff lint errors across codebase

- Sort and organize imports (I001)
- Use modern type annotations (X | Y instead of Union, collections.abc)
- Remove unused imports (F401)
- Combine nested if statements (SIM102)
- Use ternary operators where appropriate (SIM108)
- Combine nested with statements (SIM117)
- Use contextlib.suppress instead of try-except-pass (SIM105)
- Add noqa comments for intentional SIM115 violations (file locks)
- Add TYPE_CHECKING import for forward references
- Fix exception chaining (B904)

All 1033 tests pass.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* docs: add TDD workflow and pre-commit requirements to CLAUDE.md

- Add mandatory test-driven development workflow (write tests first)
- Add pre-commit requirements (must run lint and tests before committing)
- Document test organization and running commands
- Document 95% coverage requirement

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: resolve mypy type checking errors with proper structural fixes

- charts.py: Create PeriodConfig dataclass for type-safe period configuration,
  use mdates.date2num() for matplotlib datetime handling, fix x-axis limits
  for single-point charts
- db.py: Add explicit int() conversion with None handling for SQLite returns
- env.py: Add class-level type annotations to Config class
- html.py: Add MetricDisplay TypedDict, fix import order, add proper type
  annotations for table data functions
- meshcore_client.py: Add return type annotation

Update tests to use new dataclass attribute access and regenerate SVG
snapshots. Add mypy step to CLAUDE.md pre-commit requirements.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: cast Jinja2 template.render() to str for mypy

Jinja2's type stubs declare render() as returning Any, but it actually
returns str. Wrap with str() to satisfy mypy's no-any-return check.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* ci: improve workflow security and reliability

- test.yml: Pin all actions by SHA, add concurrency control to cancel
  in-progress runs on rapid pushes
- release-please.yml: Pin action by SHA, add 10-minute timeout
- conftest.py: Fix snapshot_base_time to use explicit UTC timezone for
  consistent behavior across CI and local environments

Regenerate SVG snapshots with UTC-aware timestamps.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix: add mypy command to permissions in settings.local.json

* test: add comprehensive script tests with coroutine warning fixes

- Add tests/scripts/ with tests for collect_companion, collect_repeater,
  and render scripts (1135 tests total, 96% coverage)
- Fix unawaited coroutine warnings by using AsyncMock properly for async
  functions and async_context_manager_factory fixture for context managers
- Add --cov=scripts to CI workflow and pyproject.toml coverage config
- Omit scripts/generate_snapshots.py from coverage (dev utility)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* docs: migrate claude setup to codex skills

* feat: migrate dependencies to uv (#31)

* fix: run tests through uv

* test: fix ruff lint issues in tests

Consolidate patch context managers and clean unused imports/variables

Use datetime.UTC in snapshot fixtures

* test: avoid unawaited async mocks in entrypoint tests

* ci: replace codecov with github coverage artifacts

Add junit XML output and coverage summary in job output

Upload HTML and XML coverage artifacts (3.12 only) on every run

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Jorijn Schrijvershof
2026-01-08 17:16:53 +01:00
committed by GitHub
parent 45bdf5d6d4
commit a9f6926104
125 changed files with 28005 additions and 448 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/database/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Database tests."""
tests/database/conftest.py
+59
View File
@@ -0,0 +1,59 @@
"""Fixtures for database tests."""
import time
from pathlib import Path
import pytest
@pytest.fixture
def db_path(tmp_state_dir):
"""Database path in temp state directory."""
return tmp_state_dir / "metrics.db"
@pytest.fixture
def migrations_dir():
"""Path to actual migrations directory."""
return Path(__file__).parent.parent.parent / "src" / "meshmon" / "migrations"
@pytest.fixture
def initialized_db(db_path, configured_env):
"""Fresh database with migrations applied."""
from meshmon.db import init_db
init_db(db_path)
return db_path
@pytest.fixture
def populated_db(initialized_db, sample_companion_metrics, sample_repeater_metrics):
"""Database with 7 days of sample data."""
from meshmon.db import insert_metrics
now = int(time.time())
day_seconds = 86400
# 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()
# Vary values slightly
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, initialized_db)
# 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()
# Vary values slightly
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, initialized_db)
return initialized_db
tests/database/test_db_init.py
+179
View File
@@ -0,0 +1,179 @@
"""Tests for database initialization and migrations."""
import sqlite3
import pytest
from meshmon.db import (
_get_schema_version,
get_connection,
init_db,
)
class TestInitDb:
"""Tests for init_db function."""
def test_creates_database_file(self, db_path, configured_env):
"""Creates database file if it doesn't exist."""
assert not db_path.exists()
init_db(db_path)
assert db_path.exists()
def test_creates_parent_directories(self, tmp_path, configured_env):
"""Creates parent directories if needed."""
nested_path = tmp_path / "deep" / "nested" / "metrics.db"
assert not nested_path.parent.exists()
init_db(nested_path)
assert nested_path.exists()
def test_applies_migrations(self, db_path, configured_env):
"""Applies schema migrations."""
init_db(db_path)
with get_connection(db_path, readonly=True) as conn:
version = _get_schema_version(conn)
assert version >= 1
def test_safe_to_call_multiple_times(self, db_path, configured_env):
"""Can be called multiple times without error."""
init_db(db_path)
init_db(db_path) # Should not raise
init_db(db_path) # Should not raise
with get_connection(db_path, readonly=True) as conn:
version = _get_schema_version(conn)
assert version >= 1
def test_enables_wal_mode(self, db_path, configured_env):
"""Enables WAL journal mode."""
init_db(db_path)
conn = sqlite3.connect(db_path)
try:
cursor = conn.execute("PRAGMA journal_mode")
mode = cursor.fetchone()[0]
assert mode.lower() == "wal"
finally:
conn.close()
def test_creates_metrics_table(self, db_path, configured_env):
"""Creates metrics table with correct schema."""
init_db(db_path)
with get_connection(db_path, readonly=True) as conn:
# Check table exists
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='metrics'"
)
assert cursor.fetchone() is not None
# Check columns
cursor = conn.execute("PRAGMA table_info(metrics)")
columns = {row["name"]: row for row in cursor}
assert "ts" in columns
assert "role" in columns
assert "metric" in columns
assert "value" in columns
def test_creates_db_meta_table(self, db_path, configured_env):
"""Creates db_meta table for schema versioning."""
init_db(db_path)
with get_connection(db_path, readonly=True) as conn:
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='db_meta'"
)
assert cursor.fetchone() is not None
class TestGetConnection:
"""Tests for get_connection context manager."""
def test_returns_connection(self, initialized_db):
"""Returns a working connection."""
with get_connection(initialized_db) as conn:
assert conn is not None
cursor = conn.execute("SELECT 1")
assert cursor.fetchone()[0] == 1
def test_row_factory_enabled(self, initialized_db):
"""Row factory is set to sqlite3.Row."""
with get_connection(initialized_db) as conn:
conn.execute(
"INSERT INTO metrics (ts, role, metric, value) VALUES (1, 'companion', 'test', 1.0)"
)
with get_connection(initialized_db, readonly=True) as conn:
cursor = conn.execute("SELECT * FROM metrics WHERE metric = 'test'")
row = cursor.fetchone()
# sqlite3.Row supports dict-like access
assert row["metric"] == "test"
assert row["value"] == 1.0
def test_commits_on_success(self, initialized_db):
"""Commits transaction on normal exit."""
with get_connection(initialized_db) as conn:
conn.execute(
"INSERT INTO metrics (ts, role, metric, value) VALUES (1, 'companion', 'test', 1.0)"
)
# Check data persisted
with get_connection(initialized_db, readonly=True) as conn:
cursor = conn.execute("SELECT COUNT(*) FROM metrics WHERE metric = 'test'")
assert cursor.fetchone()[0] == 1
def test_rollback_on_exception(self, initialized_db):
"""Rolls back transaction on exception."""
try:
with get_connection(initialized_db) as conn:
conn.execute(
"INSERT INTO metrics (ts, role, metric, value) VALUES (2, 'companion', 'test2', 1.0)"
)
raise ValueError("Test error")
except ValueError:
pass
# Check data was rolled back
with get_connection(initialized_db, readonly=True) as conn:
cursor = conn.execute("SELECT COUNT(*) FROM metrics WHERE metric = 'test2'")
assert cursor.fetchone()[0] == 0
def test_readonly_mode(self, initialized_db):
"""Read-only mode prevents writes."""
with (
get_connection(initialized_db, readonly=True) as conn,
pytest.raises(sqlite3.OperationalError),
):
conn.execute(
"INSERT INTO metrics (ts, role, metric, value) VALUES (1, 'companion', 'test', 1.0)"
)
class TestMigrationsDirectory:
"""Tests for migrations directory and files."""
def test_migrations_dir_exists(self, migrations_dir):
"""Migrations directory exists."""
assert migrations_dir.exists()
assert migrations_dir.is_dir()
def test_has_initial_migration(self, migrations_dir):
"""Has at least the initial schema migration."""
sql_files = list(migrations_dir.glob("*.sql"))
assert len(sql_files) >= 1
# Check for 001 prefixed file
initial = [f for f in sql_files if f.stem.startswith("001")]
assert len(initial) == 1
def test_migrations_are_numbered(self, migrations_dir):
"""Migration files follow NNN_description.sql pattern."""
import re
pattern = re.compile(r"^\d{3}_.*\.sql$")
for sql_file in migrations_dir.glob("*.sql"):
assert pattern.match(sql_file.name), f"{sql_file.name} doesn't match pattern"
tests/database/test_db_insert.py
+207
View File
@@ -0,0 +1,207 @@
"""Tests for database insert functions."""
import time
import pytest
from meshmon.db import (
get_connection,
insert_metric,
insert_metrics,
)
class TestInsertMetric:
"""Tests for insert_metric function."""
def test_inserts_single_metric(self, initialized_db):
"""Inserts a single metric successfully."""
ts = int(time.time())
result = insert_metric(ts, "companion", "battery_mv", 3850.0, initialized_db)
assert result is True
with get_connection(initialized_db, readonly=True) as conn:
cursor = conn.execute(
"SELECT value FROM metrics WHERE ts = ? AND role = ? AND metric = ?",
(ts, "companion", "battery_mv")
)
row = cursor.fetchone()
assert row is not None
assert row["value"] == 3850.0
def test_returns_false_on_duplicate(self, initialized_db):
"""Returns False for duplicate (ts, role, metric) tuple."""
ts = int(time.time())
# First insert succeeds
assert insert_metric(ts, "companion", "test", 1.0, initialized_db) is True
# Second insert with same key returns False
assert insert_metric(ts, "companion", "test", 2.0, initialized_db) is False
def test_different_roles_not_duplicate(self, initialized_db):
"""Same ts/metric with different roles are not duplicates."""
ts = int(time.time())
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())
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())
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())
with pytest.raises(ValueError, match="Invalid role"):
insert_metric(ts, "'; DROP TABLE metrics; --", "test", 1.0, initialized_db)
class TestInsertMetrics:
"""Tests for insert_metrics function (bulk insert)."""
def test_inserts_multiple_metrics(self, initialized_db):
"""Inserts multiple metrics from dict."""
ts = int(time.time())
metrics = {
"battery_mv": 3850.0,
"contacts": 5,
"uptime_secs": 86400,
}
count = insert_metrics(ts, "companion", metrics, initialized_db)
assert count == 3
with get_connection(initialized_db, readonly=True) as conn:
cursor = conn.execute(
"SELECT COUNT(*) FROM metrics WHERE ts = ?",
(ts,)
)
assert cursor.fetchone()[0] == 3
def test_returns_insert_count(self, initialized_db):
"""Returns correct count of inserted metrics."""
ts = int(time.time())
metrics = {"a": 1.0, "b": 2.0, "c": 3.0}
count = insert_metrics(ts, "companion", metrics, initialized_db)
assert count == 3
def test_skips_non_numeric_values(self, initialized_db):
"""Non-numeric values are silently skipped."""
ts = int(time.time())
metrics = {
"battery_mv": 3850.0, # Numeric - inserted
"name": "test", # String - skipped
"status": None, # None - skipped
"flags": [1, 2, 3], # List - skipped
"nested": {"a": 1}, # Dict - skipped
}
count = insert_metrics(ts, "companion", metrics, initialized_db)
assert count == 1 # Only battery_mv
def test_handles_int_and_float(self, initialized_db):
"""Both int and float values are inserted."""
ts = int(time.time())
metrics = {
"int_value": 42,
"float_value": 3.14,
}
count = insert_metrics(ts, "companion", metrics, initialized_db)
assert count == 2
def test_converts_int_to_float(self, initialized_db):
"""Integer values are stored as float."""
ts = int(time.time())
metrics = {"contacts": 5}
insert_metrics(ts, "companion", metrics, initialized_db)
with get_connection(initialized_db, readonly=True) as conn:
cursor = conn.execute(
"SELECT value FROM metrics WHERE metric = 'contacts'"
)
row = cursor.fetchone()
assert row["value"] == 5.0
assert isinstance(row["value"], float)
def test_empty_dict_returns_zero(self, initialized_db):
"""Empty dict returns 0."""
ts = int(time.time())
count = insert_metrics(ts, "companion", {}, initialized_db)
assert count == 0
def test_skips_duplicates_silently(self, initialized_db):
"""Duplicate metrics are skipped without error."""
ts = int(time.time())
metrics = {"test": 1.0}
# First insert
count1 = insert_metrics(ts, "companion", metrics, initialized_db)
assert count1 == 1
# Second insert - same key
count2 = insert_metrics(ts, "companion", metrics, initialized_db)
assert count2 == 0 # Duplicate skipped
def test_partial_duplicates(self, initialized_db):
"""Partial duplicates: some inserted, some skipped."""
ts = int(time.time())
# First insert
insert_metrics(ts, "companion", {"existing": 1.0}, initialized_db)
# Second insert with mix
metrics = {
"existing": 2.0, # Duplicate - skipped
"new": 3.0, # New - inserted
}
count = insert_metrics(ts, "companion", metrics, initialized_db)
assert count == 1 # Only "new" inserted
def test_invalid_role_raises(self, initialized_db):
"""Invalid role raises ValueError."""
ts = int(time.time())
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())
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
def test_repeater_metrics(self, initialized_db, sample_repeater_metrics):
"""Inserts repeater metrics dict."""
ts = int(time.time())
count = insert_metrics(ts, "repeater", sample_repeater_metrics, initialized_db)
# Should insert all numeric fields
assert count >= 10 # Many metrics
tests/database/test_db_maintenance.py
+195
View File
@@ -0,0 +1,195 @@
"""Tests for database maintenance functions."""
import os
import sqlite3
from meshmon.db import (
get_db_path,
init_db,
vacuum_db,
)
class TestVacuumDb:
"""Tests for vacuum_db function."""
def test_vacuums_existing_db(self, initialized_db):
"""Vacuum should run without error on initialized database."""
# Add some data then 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()
# Should not raise
vacuum_db(initialized_db)
def test_runs_analyze(self, initialized_db, capfd):
"""ANALYZE should be run after VACUUM."""
# 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
conn.close()
def test_uses_default_path_when_none(self, configured_env, monkeypatch):
"""Uses get_db_path() when no path provided."""
# Initialize db at default location
init_db()
# vacuum_db with None should use default path
vacuum_db(None)
def test_can_vacuum_empty_db(self, initialized_db):
"""Can vacuum an empty database."""
vacuum_db(initialized_db)
def test_reclaims_space_after_delete(self, initialized_db):
"""Vacuum should reclaim space after deleting rows."""
conn = sqlite3.connect(initialized_db)
# Insert many rows
for i in range(1000):
conn.execute(
"INSERT INTO metrics (ts, role, metric, value) VALUES (?, 'companion', 'test', 1.0)",
(i,)
)
conn.commit()
# Get size before delete
conn.close()
size_before = os.path.getsize(initialized_db)
# Delete all rows
conn = sqlite3.connect(initialized_db)
conn.execute("DELETE FROM metrics")
conn.commit()
conn.close()
# Vacuum
vacuum_db(initialized_db)
# Size should be smaller (or at least not larger)
size_after = os.path.getsize(initialized_db)
# Note: Due to WAL mode, this might not always shrink dramatically
# but vacuum should at least complete without error
assert size_after <= size_before + 4096 # Allow for some overhead
class TestGetDbPath:
"""Tests for get_db_path function."""
def test_returns_path_in_state_dir(self, configured_env):
"""Path should be in the configured state directory."""
path = get_db_path()
assert path.name == "metrics.db"
assert str(configured_env["state_dir"]) in str(path)
def test_returns_path_object(self, configured_env):
"""Should return a Path object."""
from pathlib import Path
path = get_db_path()
assert isinstance(path, Path)
class TestDatabaseIntegrity:
"""Tests for database integrity after operations."""
def test_wal_mode_enabled(self, initialized_db):
"""Database should be in WAL mode."""
conn = sqlite3.connect(initialized_db)
cursor = conn.execute("PRAGMA journal_mode")
mode = cursor.fetchone()[0]
conn.close()
assert mode.lower() == "wal"
def test_foreign_keys_disabled_by_default(self, initialized_db):
"""Foreign keys should be disabled (SQLite default)."""
conn = sqlite3.connect(initialized_db)
cursor = conn.execute("PRAGMA foreign_keys")
enabled = cursor.fetchone()[0]
conn.close()
# Default is off, and we don't explicitly enable them
assert enabled == 0
def test_metrics_table_exists(self, initialized_db):
"""Metrics table should exist after init."""
conn = sqlite3.connect(initialized_db)
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='metrics'"
)
result = cursor.fetchone()
conn.close()
assert result is not None
assert result[0] == "metrics"
def test_db_meta_table_exists(self, initialized_db):
"""db_meta table should exist after init."""
conn = sqlite3.connect(initialized_db)
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='db_meta'"
)
result = cursor.fetchone()
conn.close()
assert result is not None
def test_metrics_index_exists(self, initialized_db):
"""Index on metrics(role, ts) should exist."""
conn = sqlite3.connect(initialized_db)
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='index' AND name='idx_metrics_role_ts'"
)
result = cursor.fetchone()
conn.close()
assert result is not None
def test_vacuum_preserves_data(self, initialized_db):
"""Vacuum should not lose any data."""
conn = sqlite3.connect(initialized_db)
for i in range(100):
conn.execute(
"INSERT INTO metrics (ts, role, metric, value) VALUES (?, 'companion', 'test', ?)",
(i, float(i))
)
conn.commit()
conn.close()
# Vacuum
vacuum_db(initialized_db)
# Check data is still there
conn = sqlite3.connect(initialized_db)
cursor = conn.execute("SELECT COUNT(*) FROM metrics")
count = cursor.fetchone()[0]
conn.close()
assert count == 100
def test_vacuum_preserves_schema_version(self, initialized_db):
"""Vacuum should not change schema version."""
from meshmon.db import _get_schema_version
conn = sqlite3.connect(initialized_db)
version_before = _get_schema_version(conn)
conn.close()
vacuum_db(initialized_db)
conn = sqlite3.connect(initialized_db)
version_after = _get_schema_version(conn)
conn.close()
assert version_before == version_after
tests/database/test_db_migrations.py
+331
View File
@@ -0,0 +1,331 @@
"""Tests for database migration system."""
import sqlite3
from pathlib import Path
import pytest
from meshmon.db import (
_apply_migrations,
_get_migration_files,
_get_schema_version,
_set_schema_version,
get_schema_version,
)
class TestGetMigrationFiles:
"""Tests for _get_migration_files function."""
def test_finds_migration_files(self):
"""Should find actual migration files in MIGRATIONS_DIR."""
migrations = _get_migration_files()
assert len(migrations) >= 2
# Should include 001 and 002
versions = [v for v, _ in migrations]
assert 1 in versions
assert 2 in versions
def test_returns_sorted_by_version(self):
"""Migrations should be sorted by version number."""
migrations = _get_migration_files()
versions = [v for v, _ in migrations]
assert versions == sorted(versions)
def test_returns_path_objects(self):
"""Each migration should have a Path object."""
migrations = _get_migration_files()
for _version, path in migrations:
assert isinstance(path, Path)
assert path.exists()
assert path.suffix == ".sql"
def test_extracts_version_from_filename(self):
"""Version number extracted from filename prefix."""
migrations = _get_migration_files()
for version, path in migrations:
filename_version = int(path.stem.split("_")[0])
assert version == filename_version
def test_empty_when_no_migrations_dir(self, tmp_path, monkeypatch):
"""Returns empty list when migrations dir doesn't exist."""
fake_dir = tmp_path / "nonexistent"
monkeypatch.setattr("meshmon.db.MIGRATIONS_DIR", fake_dir)
migrations = _get_migration_files()
assert migrations == []
def test_skips_invalid_filenames(self, tmp_path, monkeypatch):
"""Skips files without valid version prefix."""
migrations_dir = tmp_path / "migrations"
migrations_dir.mkdir()
# Create valid migration
(migrations_dir / "001_valid.sql").write_text("-- valid")
# Create invalid migrations
(migrations_dir / "invalid_name.sql").write_text("-- invalid")
(migrations_dir / "abc_noversion.sql").write_text("-- no version")
monkeypatch.setattr("meshmon.db.MIGRATIONS_DIR", migrations_dir)
migrations = _get_migration_files()
assert len(migrations) == 1
assert migrations[0][0] == 1
class TestGetSchemaVersion:
"""Tests for _get_schema_version internal function."""
def test_returns_zero_for_fresh_db(self, tmp_path):
"""Fresh database with no db_meta returns 0."""
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
version = _get_schema_version(conn)
assert version == 0
conn.close()
def test_returns_stored_version(self, tmp_path):
"""Returns version from db_meta table."""
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE db_meta (
key TEXT PRIMARY KEY NOT NULL,
value TEXT NOT NULL
)
""")
conn.execute(
"INSERT INTO db_meta (key, value) VALUES ('schema_version', '5')"
)
conn.commit()
version = _get_schema_version(conn)
assert version == 5
conn.close()
def test_returns_zero_when_key_missing(self, tmp_path):
"""Returns 0 if db_meta exists but schema_version key is missing."""
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE db_meta (
key TEXT PRIMARY KEY NOT NULL,
value TEXT NOT NULL
)
""")
conn.execute(
"INSERT INTO db_meta (key, value) VALUES ('other_key', 'value')"
)
conn.commit()
version = _get_schema_version(conn)
assert version == 0
conn.close()
class TestSetSchemaVersion:
"""Tests for _set_schema_version internal function."""
def test_inserts_new_version(self, tmp_path):
"""Can insert schema version into fresh db_meta."""
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE db_meta (
key TEXT PRIMARY KEY NOT NULL,
value TEXT NOT NULL
)
""")
_set_schema_version(conn, 3)
conn.commit()
cursor = conn.execute(
"SELECT value FROM db_meta WHERE key = 'schema_version'"
)
assert cursor.fetchone()[0] == "3"
conn.close()
def test_updates_existing_version(self, tmp_path):
"""Can update existing schema version."""
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE db_meta (
key TEXT PRIMARY KEY NOT NULL,
value TEXT NOT NULL
)
""")
conn.execute(
"INSERT INTO db_meta (key, value) VALUES ('schema_version', '1')"
)
conn.commit()
_set_schema_version(conn, 5)
conn.commit()
cursor = conn.execute(
"SELECT value FROM db_meta WHERE key = 'schema_version'"
)
assert cursor.fetchone()[0] == "5"
conn.close()
class TestApplyMigrations:
"""Tests for _apply_migrations function."""
def test_applies_all_migrations_to_fresh_db(self, tmp_path, monkeypatch):
"""Applies all migrations to a fresh database."""
# Create mock migrations
migrations_dir = tmp_path / "migrations"
migrations_dir.mkdir()
(migrations_dir / "001_initial.sql").write_text("""
CREATE TABLE IF NOT EXISTS db_meta (
key TEXT PRIMARY KEY NOT NULL,
value TEXT NOT NULL
);
CREATE TABLE test1 (id INTEGER);
""")
(migrations_dir / "002_second.sql").write_text("""
CREATE TABLE test2 (id INTEGER);
""")
monkeypatch.setattr("meshmon.db.MIGRATIONS_DIR", migrations_dir)
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
_apply_migrations(conn)
# Check both tables exist
cursor = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
)
tables = [row[0] for row in cursor]
assert "test1" in tables
assert "test2" in tables
assert "db_meta" in tables
# Check version is updated
assert _get_schema_version(conn) == 2
conn.close()
def test_skips_already_applied_migrations(self, tmp_path, monkeypatch):
"""Skips migrations that have already been applied."""
migrations_dir = tmp_path / "migrations"
migrations_dir.mkdir()
(migrations_dir / "001_initial.sql").write_text("""
CREATE TABLE IF NOT EXISTS db_meta (
key TEXT PRIMARY KEY NOT NULL,
value TEXT NOT NULL
);
CREATE TABLE test1 (id INTEGER);
""")
(migrations_dir / "002_second.sql").write_text("""
CREATE TABLE test2 (id INTEGER);
""")
monkeypatch.setattr("meshmon.db.MIGRATIONS_DIR", migrations_dir)
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
# Apply first time
_apply_migrations(conn)
# Apply second time - should not fail
_apply_migrations(conn)
assert _get_schema_version(conn) == 2
conn.close()
def test_raises_when_no_migrations(self, tmp_path, monkeypatch):
"""Raises error when no migration files exist."""
empty_dir = tmp_path / "empty_migrations"
empty_dir.mkdir()
monkeypatch.setattr("meshmon.db.MIGRATIONS_DIR", empty_dir)
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
with pytest.raises(RuntimeError, match="No migration files found"):
_apply_migrations(conn)
conn.close()
def test_rolls_back_failed_migration(self, tmp_path, monkeypatch):
"""Rolls back if a migration fails."""
migrations_dir = tmp_path / "migrations"
migrations_dir.mkdir()
(migrations_dir / "001_initial.sql").write_text("""
CREATE TABLE IF NOT EXISTS db_meta (
key TEXT PRIMARY KEY NOT NULL,
value TEXT NOT NULL
);
CREATE TABLE test1 (id INTEGER);
""")
(migrations_dir / "002_broken.sql").write_text("""
THIS IS NOT VALID SQL;
""")
monkeypatch.setattr("meshmon.db.MIGRATIONS_DIR", migrations_dir)
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
with pytest.raises(RuntimeError, match="Migration.*failed"):
_apply_migrations(conn)
# Version should still be 1 (first migration applied)
assert _get_schema_version(conn) == 1
conn.close()
class TestPublicGetSchemaVersion:
"""Tests for public get_schema_version function."""
def test_returns_zero_when_db_missing(self, configured_env):
"""Returns 0 when database file doesn't exist."""
version = get_schema_version()
assert version == 0
def test_returns_version_from_existing_db(self, initialized_db):
"""Returns schema version from initialized database."""
version = get_schema_version()
# Should be at least version 2 (we have 2 migrations)
assert version >= 2
def test_uses_readonly_connection(self, initialized_db, monkeypatch):
"""Opens database in readonly mode."""
calls = []
original_get_connection = __import__(
"meshmon.db", fromlist=["get_connection"]
).get_connection
from contextlib import contextmanager
@contextmanager
def mock_get_connection(*args, **kwargs):
calls.append(kwargs)
with original_get_connection(*args, **kwargs) as conn:
yield conn
monkeypatch.setattr("meshmon.db.get_connection", mock_get_connection)
get_schema_version()
assert any(call.get("readonly") is True for call in calls)
tests/database/test_db_queries.py
+312
View File
@@ -0,0 +1,312 @@
"""Tests for database query functions."""
import time
import pytest
from meshmon.db import (
get_available_metrics,
get_distinct_timestamps,
get_latest_metrics,
get_metric_count,
get_metrics_for_period,
insert_metrics,
)
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())
insert_metrics(ts, "companion", {
"battery_mv": 3850.0,
"contacts": 5,
}, initialized_db)
result = get_metrics_for_period(
"companion", ts - 100, ts + 100, initialized_db
)
assert isinstance(result, dict)
assert "battery_mv" in result
assert "contacts" in result
def test_returns_timestamp_value_tuples(self, initialized_db):
"""Each metric has list of (ts, value) tuples."""
ts = int(time.time())
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
result = get_metrics_for_period(
"companion", ts - 100, ts + 100, initialized_db
)
assert len(result["test"]) == 1
assert result["test"][0] == (ts, 1.0)
def test_sorted_by_timestamp(self, initialized_db):
"""Results are sorted by timestamp ascending."""
base_ts = int(time.time())
# Insert out of order
insert_metrics(base_ts + 200, "companion", {"test": 3.0}, initialized_db)
insert_metrics(base_ts, "companion", {"test": 1.0}, initialized_db)
insert_metrics(base_ts + 100, "companion", {"test": 2.0}, initialized_db)
result = get_metrics_for_period(
"companion", base_ts - 100, base_ts + 300, initialized_db
)
values = [v for ts, v in result["test"]]
assert values == [1.0, 2.0, 3.0]
def test_respects_time_range(self, initialized_db):
"""Only returns data within specified time range."""
base_ts = int(time.time())
insert_metrics(base_ts - 200, "companion", {"test": 1.0}, initialized_db) # Outside
insert_metrics(base_ts, "companion", {"test": 2.0}, initialized_db) # Inside
insert_metrics(base_ts + 200, "companion", {"test": 3.0}, initialized_db) # Outside
result = get_metrics_for_period(
"companion", base_ts - 100, base_ts + 100, initialized_db
)
assert len(result["test"]) == 1
assert result["test"][0][1] == 2.0
def test_filters_by_role(self, initialized_db):
"""Only returns data for specified role."""
ts = int(time.time())
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
insert_metrics(ts, "repeater", {"test": 2.0}, initialized_db)
result = get_metrics_for_period(
"companion", ts - 100, ts + 100, initialized_db
)
assert result["test"][0][1] == 1.0
def test_computes_bat_pct(self, initialized_db):
"""Computes bat_pct from battery voltage."""
ts = int(time.time())
# 4200 mV = 4.2V = 100%
insert_metrics(ts, "companion", {"battery_mv": 4200.0}, initialized_db)
result = get_metrics_for_period(
"companion", ts - 100, ts + 100, initialized_db
)
assert "bat_pct" in result
assert result["bat_pct"][0][1] == pytest.approx(100.0)
def test_bat_pct_for_repeater(self, initialized_db):
"""Computes bat_pct for repeater using 'bat' field."""
ts = int(time.time())
# 3000 mV = 3.0V = 0%
insert_metrics(ts, "repeater", {"bat": 3000.0}, initialized_db)
result = get_metrics_for_period(
"repeater", ts - 100, ts + 100, initialized_db
)
assert "bat_pct" in result
assert result["bat_pct"][0][1] == pytest.approx(0.0)
def test_empty_period_returns_empty(self, initialized_db):
"""Empty time period returns empty dict."""
result = get_metrics_for_period(
"companion", 0, 1, initialized_db
)
assert result == {}
def test_invalid_role_raises(self, initialized_db):
"""Invalid role raises ValueError."""
with pytest.raises(ValueError, match="Invalid role"):
get_metrics_for_period("invalid", 0, 100, initialized_db)
class TestGetLatestMetrics:
"""Tests for get_latest_metrics function."""
def test_returns_most_recent(self, initialized_db):
"""Returns metrics at most recent timestamp."""
base_ts = int(time.time())
insert_metrics(base_ts, "companion", {"test": 1.0}, initialized_db)
insert_metrics(base_ts + 100, "companion", {"test": 2.0}, initialized_db)
result = get_latest_metrics("companion", initialized_db)
assert result["test"] == 2.0
assert result["ts"] == base_ts + 100
def test_includes_ts(self, initialized_db):
"""Result includes 'ts' key with timestamp."""
ts = int(time.time())
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
result = get_latest_metrics("companion", initialized_db)
assert "ts" in result
assert result["ts"] == ts
def test_includes_all_metrics(self, initialized_db):
"""Result includes all metrics at that timestamp."""
ts = int(time.time())
insert_metrics(ts, "companion", {
"battery_mv": 3850.0,
"contacts": 5,
"uptime_secs": 86400,
}, initialized_db)
result = get_latest_metrics("companion", initialized_db)
assert result["battery_mv"] == 3850.0
assert result["contacts"] == 5.0
assert result["uptime_secs"] == 86400.0
def test_computes_bat_pct(self, initialized_db):
"""Computes bat_pct from battery voltage."""
ts = int(time.time())
insert_metrics(ts, "companion", {"battery_mv": 3820.0}, initialized_db)
result = get_latest_metrics("companion", initialized_db)
assert "bat_pct" in result
assert result["bat_pct"] == pytest.approx(50.0)
def test_returns_none_when_empty(self, initialized_db):
"""Returns None when no data exists."""
result = get_latest_metrics("companion", initialized_db)
assert result is None
def test_filters_by_role(self, initialized_db):
"""Only returns data for specified role."""
ts = int(time.time())
insert_metrics(ts, "companion", {"test": 1.0}, initialized_db)
insert_metrics(ts + 100, "repeater", {"test": 2.0}, initialized_db)
result = get_latest_metrics("companion", initialized_db)
assert result["ts"] == ts
assert result["test"] == 1.0
def test_invalid_role_raises(self, initialized_db):
"""Invalid role raises ValueError."""
with pytest.raises(ValueError, match="Invalid role"):
get_latest_metrics("invalid", initialized_db)
class TestGetMetricCount:
"""Tests for get_metric_count function."""
def test_counts_rows(self, initialized_db):
"""Counts total metric rows for role."""
ts = int(time.time())
insert_metrics(ts, "companion", {"a": 1.0, "b": 2.0, "c": 3.0}, initialized_db)
count = get_metric_count("companion", initialized_db)
assert count == 3
def test_filters_by_role(self, initialized_db):
"""Only counts rows for specified role."""
ts = int(time.time())
insert_metrics(ts, "companion", {"a": 1.0}, initialized_db)
insert_metrics(ts, "repeater", {"b": 2.0, "c": 3.0}, initialized_db)
assert get_metric_count("companion", initialized_db) == 1
assert get_metric_count("repeater", initialized_db) == 2
def test_returns_zero_when_empty(self, initialized_db):
"""Returns 0 when no data exists."""
count = get_metric_count("companion", initialized_db)
assert count == 0
def test_invalid_role_raises(self, initialized_db):
"""Invalid role raises ValueError."""
with pytest.raises(ValueError, match="Invalid role"):
get_metric_count("invalid", initialized_db)
class TestGetDistinctTimestamps:
"""Tests for get_distinct_timestamps function."""
def test_counts_unique_timestamps(self, initialized_db):
"""Counts distinct timestamps."""
ts = int(time.time())
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
count = get_distinct_timestamps("companion", initialized_db)
assert count == 2
def test_filters_by_role(self, initialized_db):
"""Only counts timestamps for specified role."""
ts = int(time.time())
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)
assert get_distinct_timestamps("companion", initialized_db) == 2
assert get_distinct_timestamps("repeater", initialized_db) == 1
def test_returns_zero_when_empty(self, initialized_db):
"""Returns 0 when no data exists."""
count = get_distinct_timestamps("companion", initialized_db)
assert count == 0
class TestGetAvailableMetrics:
"""Tests for get_available_metrics function."""
def test_returns_metric_names(self, initialized_db):
"""Returns list of distinct metric names."""
ts = int(time.time())
insert_metrics(ts, "companion", {
"battery_mv": 3850.0,
"contacts": 5,
"recv": 100,
}, initialized_db)
metrics = get_available_metrics("companion", initialized_db)
assert "battery_mv" in metrics
assert "contacts" in metrics
assert "recv" in metrics
def test_sorted_alphabetically(self, initialized_db):
"""Metrics are sorted alphabetically."""
ts = int(time.time())
insert_metrics(ts, "companion", {
"zebra": 1.0,
"apple": 2.0,
"mango": 3.0,
}, initialized_db)
metrics = get_available_metrics("companion", initialized_db)
assert metrics == sorted(metrics)
def test_filters_by_role(self, initialized_db):
"""Only returns metrics for specified role."""
ts = int(time.time())
insert_metrics(ts, "companion", {"companion_metric": 1.0}, initialized_db)
insert_metrics(ts, "repeater", {"repeater_metric": 2.0}, initialized_db)
companion_metrics = get_available_metrics("companion", initialized_db)
repeater_metrics = get_available_metrics("repeater", initialized_db)
assert "companion_metric" in companion_metrics
assert "repeater_metric" not in companion_metrics
assert "repeater_metric" in repeater_metrics
def test_returns_empty_when_no_data(self, initialized_db):
"""Returns empty list when no data exists."""
metrics = get_available_metrics("companion", initialized_db)
assert metrics == []
tests/database/test_db_validation.py
+210
View File
@@ -0,0 +1,210 @@
"""Tests for database validation and security functions."""
import pytest
from meshmon.db import (
VALID_ROLES,
_validate_role,
get_available_metrics,
get_distinct_timestamps,
get_latest_metrics,
get_metric_count,
get_metrics_for_period,
insert_metric,
insert_metrics,
)
class TestValidateRole:
"""Tests for _validate_role function."""
def test_accepts_companion(self):
"""Accepts 'companion' as valid role."""
result = _validate_role("companion")
assert result == "companion"
def test_accepts_repeater(self):
"""Accepts 'repeater' as valid role."""
result = _validate_role("repeater")
assert result == "repeater"
def test_returns_input_on_success(self):
"""Returns the validated role string."""
for role in VALID_ROLES:
result = _validate_role(role)
assert result == role
def test_rejects_invalid_role(self):
"""Rejects invalid role names."""
with pytest.raises(ValueError, match="Invalid role"):
_validate_role("invalid")
def test_rejects_empty_string(self):
"""Rejects empty string as role."""
with pytest.raises(ValueError, match="Invalid role"):
_validate_role("")
def test_rejects_none(self):
"""Rejects None as role."""
with pytest.raises(ValueError):
_validate_role(None)
def test_case_sensitive(self):
"""Role validation is case-sensitive."""
with pytest.raises(ValueError, match="Invalid role"):
_validate_role("Companion")
with pytest.raises(ValueError, match="Invalid role"):
_validate_role("REPEATER")
def test_rejects_whitespace_variants(self):
"""Rejects roles with leading/trailing whitespace."""
with pytest.raises(ValueError, match="Invalid role"):
_validate_role(" companion")
with pytest.raises(ValueError, match="Invalid role"):
_validate_role("repeater ")
with pytest.raises(ValueError, match="Invalid role"):
_validate_role(" companion ")
class TestSqlInjectionPrevention:
"""Tests to verify SQL injection is prevented via role validation."""
@pytest.mark.parametrize("malicious_role", [
"'; DROP TABLE metrics; --",
"admin'; DROP TABLE metrics;--",
"companion OR 1=1",
"companion; DELETE FROM metrics",
"companion' UNION SELECT * FROM db_meta --",
"companion\"; DROP TABLE metrics; --",
"1 OR 1=1",
"companion/*comment*/",
])
def test_insert_metric_rejects_injection(self, initialized_db, malicious_role):
"""insert_metric rejects SQL injection attempts."""
with pytest.raises(ValueError, match="Invalid role"):
insert_metric(1000, malicious_role, "test", 1.0, initialized_db)
@pytest.mark.parametrize("malicious_role", [
"'; DROP TABLE metrics; --",
"companion OR 1=1",
])
def test_insert_metrics_rejects_injection(self, initialized_db, malicious_role):
"""insert_metrics rejects SQL injection attempts."""
with pytest.raises(ValueError, match="Invalid role"):
insert_metrics(1000, malicious_role, {"test": 1.0}, initialized_db)
@pytest.mark.parametrize("malicious_role", [
"'; DROP TABLE metrics; --",
"companion OR 1=1",
])
def test_get_metrics_for_period_rejects_injection(self, initialized_db, malicious_role):
"""get_metrics_for_period rejects SQL injection attempts."""
with pytest.raises(ValueError, match="Invalid role"):
get_metrics_for_period(malicious_role, 0, 100, initialized_db)
@pytest.mark.parametrize("malicious_role", [
"'; DROP TABLE metrics; --",
"companion OR 1=1",
])
def test_get_latest_metrics_rejects_injection(self, initialized_db, malicious_role):
"""get_latest_metrics rejects SQL injection attempts."""
with pytest.raises(ValueError, match="Invalid role"):
get_latest_metrics(malicious_role, initialized_db)
@pytest.mark.parametrize("malicious_role", [
"'; DROP TABLE metrics; --",
"companion OR 1=1",
])
def test_get_metric_count_rejects_injection(self, initialized_db, malicious_role):
"""get_metric_count rejects SQL injection attempts."""
with pytest.raises(ValueError, match="Invalid role"):
get_metric_count(malicious_role, initialized_db)
@pytest.mark.parametrize("malicious_role", [
"'; DROP TABLE metrics; --",
"companion OR 1=1",
])
def test_get_distinct_timestamps_rejects_injection(self, initialized_db, malicious_role):
"""get_distinct_timestamps rejects SQL injection attempts."""
with pytest.raises(ValueError, match="Invalid role"):
get_distinct_timestamps(malicious_role, initialized_db)
@pytest.mark.parametrize("malicious_role", [
"'; DROP TABLE metrics; --",
"companion OR 1=1",
])
def test_get_available_metrics_rejects_injection(self, initialized_db, malicious_role):
"""get_available_metrics rejects SQL injection attempts."""
with pytest.raises(ValueError, match="Invalid role"):
get_available_metrics(malicious_role, initialized_db)
class TestValidRolesConstant:
"""Tests for VALID_ROLES constant."""
def test_contains_companion(self):
"""VALID_ROLES includes 'companion'."""
assert "companion" in VALID_ROLES
def test_contains_repeater(self):
"""VALID_ROLES includes 'repeater'."""
assert "repeater" in VALID_ROLES
def test_is_tuple(self):
"""VALID_ROLES is immutable (tuple)."""
assert isinstance(VALID_ROLES, tuple)
def test_exactly_two_roles(self):
"""There are exactly two valid roles."""
assert len(VALID_ROLES) == 2
class TestMetricNameValidation:
"""Tests for metric name handling (not validated, but should handle safely)."""
def test_metric_name_with_special_chars(self, initialized_db):
"""Metric names with special chars are handled via parameterized queries."""
# These should work because we use parameterized queries
insert_metric(1000, "companion", "test.metric", 1.0, initialized_db)
insert_metric(1001, "companion", "test-metric", 2.0, initialized_db)
insert_metric(1002, "companion", "test_metric", 3.0, initialized_db)
metrics = get_available_metrics("companion", initialized_db)
assert "test.metric" in metrics
assert "test-metric" in metrics
assert "test_metric" in metrics
def test_metric_name_with_spaces(self, initialized_db):
"""Metric names with spaces are handled safely."""
insert_metric(1000, "companion", "test metric", 1.0, initialized_db)
metrics = get_available_metrics("companion", initialized_db)
assert "test metric" in metrics
def test_metric_name_unicode(self, initialized_db):
"""Unicode metric names are handled safely."""
insert_metric(1000, "companion", "température", 1.0, initialized_db)
insert_metric(1001, "companion", "温度", 2.0, initialized_db)
metrics = get_available_metrics("companion", initialized_db)
assert "température" in metrics
assert "温度" in metrics
def test_empty_metric_name(self, initialized_db):
"""Empty metric name is allowed (not validated)."""
# Empty string is allowed as metric name
insert_metric(1000, "companion", "", 1.0, initialized_db)
metrics = get_available_metrics("companion", initialized_db)
assert "" in metrics
def test_very_long_metric_name(self, initialized_db):
"""Very long metric names are handled."""
long_name = "a" * 1000
insert_metric(1000, "companion", long_name, 1.0, initialized_db)
metrics = get_available_metrics("companion", initialized_db)
assert long_name in metrics