openhop_repeater/tests/test_sqlite_upgrade_existing_db.py

"""Regression test: opening a pre-existing database must not break packet recording.

`SQLiteHandler.__init__` runs `_init_database()` and then `_run_migrations()`.
`_init_database` used to create `idx_packets_upstream_time`, an index spanning
`upstream_hash` / `upstream_hash_size` -- columns that only exist on an older
database once `add_upstream_hash_to_packets` has added them, which happens in
the migration step afterwards.

On a fresh install this passed, because the `CREATE TABLE` includes the columns.
On any existing install it raised ``no such column: upstream_hash``, which
`_init_database` caught and logged. Every statement after the failing index was
therefore skipped, and the daemon came up looking completely healthy -- service
active, radio initialised, companions serving, HTTP up -- while recording zero
packets and doing no RX/TX.

These tests build a database with the *old* packets schema and assert that a
handler opens it, ends up with both columns and the index, and can insert.
"""

import os
import shutil
import sqlite3
import sys
import tempfile
from pathlib import Path

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from repeater.data_acquisition.sqlite_handler import (  # noqa: E402
    AIRTIME_INDEX_COLUMNS,
    SQLiteHandler,
)

# The packets columns that predate migration 13. Enough of the real schema for
# the handler to treat this as an existing database rather than a fresh one.
_OLD_PACKETS_SCHEMA = """
CREATE TABLE packets (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp REAL,
    type INTEGER,
    route INTEGER,
    length INTEGER,
    rssi REAL,
    snr REAL,
    score REAL,
    transmitted INTEGER,
    is_duplicate INTEGER,
    drop_reason TEXT,
    src_hash TEXT,
    dst_hash TEXT,
    path_hash TEXT,
    header TEXT,
    transport_codes TEXT,
    payload TEXT,
    payload_length INTEGER,
    packet_hash TEXT
)
"""


def _make_legacy_db(path):
    con = sqlite3.connect(path)
    con.execute(_OLD_PACKETS_SCHEMA)
    con.execute("INSERT INTO packets (timestamp, type, length) VALUES (?, ?, ?)", (1.0, 1, 10))
    con.commit()
    con.close()


def _columns(path, table="packets"):
    con = sqlite3.connect(path)
    try:
        return {row[1] for row in con.execute("PRAGMA table_info(%s)" % table)}
    finally:
        con.close()


def _indexes(path):
    con = sqlite3.connect(path)
    try:
        return {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='index'")}
    finally:
        con.close()


@pytest.fixture
def legacy_db():
    """A storage dir holding a repeater.db with the pre-migration-13 schema."""
    d = tempfile.mkdtemp()
    path = os.path.join(d, "repeater.db")
    _make_legacy_db(path)
    yield d, path
    shutil.rmtree(d, ignore_errors=True)


def test_opens_database_missing_upstream_hash_columns(legacy_db):
    """The upgrade path must complete, not half-apply."""
    d, path = legacy_db
    assert "upstream_hash" not in _columns(path), "fixture is not a legacy db"

    SQLiteHandler(Path(d))

    cols = _columns(path)
    assert "upstream_hash" in cols
    assert "upstream_hash_size" in cols


def test_upstream_index_exists_after_upgrade(legacy_db):
    """The index must still be created -- by the migration that owns the columns."""
    d, path = legacy_db
    SQLiteHandler(Path(d))
    assert "idx_packets_upstream_time" in _indexes(path)


def _tables(path):
    con = sqlite3.connect(path)
    try:
        return {r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")}
    finally:
        con.close()


def test_schema_init_runs_to_completion(legacy_db):
    """The actual damage: `_init_database` aborting part-way.

    This is what the caught exception hid. `idx_packets_upstream_time` was
    created roughly two thirds of the way through `_init_database`; when it
    raised, every statement after it was skipped and the error was swallowed.
    The daemon then ran with a partially-created schema.

    `room_messages` and `room_client_sync` are created *after* that point, so
    their presence is a direct proof that schema init completed. Asserting on
    the upstream columns alone is not enough -- the migration adds those
    afterwards either way, which is exactly why this bug self-heals on the
    second start and is so easy to misread as harmless.
    """
    d, path = legacy_db
    SQLiteHandler(Path(d))

    tables = _tables(path)
    for late in ("room_messages", "room_client_sync"):
        assert late in tables, (
            "%s is missing: _init_database did not run to completion on an existing database" % late
        )


def test_recording_works_after_upgrade(legacy_db):
    """A packet insert must still land once the upgrade has run."""
    d, path = legacy_db
    SQLiteHandler(Path(d))

    con = sqlite3.connect(path)
    try:
        before = con.execute("SELECT COUNT(*) FROM packets").fetchone()[0]
        con.execute(
            "INSERT INTO packets (timestamp, type, length, upstream_hash) VALUES (?, ?, ?, ?)",
            (2.0, 1, 12, "ABCD"),
        )
        con.commit()
        after = con.execute("SELECT COUNT(*) FROM packets").fetchone()[0]
    finally:
        con.close()
    assert after == before + 1


def test_upgrade_adds_tx_radio_ids_without_touching_existing_rows(legacy_db):
    d, path = legacy_db
    assert "tx_radio_ids" not in _columns(path), "fixture is not a legacy db"

    SQLiteHandler(Path(d))

    assert "tx_radio_ids" in _columns(path)
    con = sqlite3.connect(path)
    try:
        rows = con.execute("SELECT timestamp, type, length, tx_radio_ids FROM packets").fetchall()
    finally:
        con.close()
    assert rows == [(1.0, 1, 10, None)]


def test_tx_radio_ids_round_trip_through_every_packet_read():
    d = tempfile.mkdtemp()
    try:
        handler = SQLiteHandler(Path(d))
        base = {"type": 2, "route": 1, "length": 4, "transmitted": True}
        handler.store_packet(
            {
                **base,
                "timestamp": 5.0,
                "packet_hash": "FANOUT0000000001",
                "rx_radio_id": "local",
                "tx_radio_id": "link",
                "tx_radio_ids": ["link", "local"],
            }
        )
        handler.store_packet({**base, "timestamp": 6.0, "packet_hash": "SINGLE0000000002"})

        con = sqlite3.connect(os.path.join(d, "repeater.db"))
        try:
            stored = con.execute(
                "SELECT tx_radio_ids FROM packets WHERE packet_hash = ?", ("FANOUT0000000001",)
            ).fetchone()[0]
        finally:
            con.close()
        assert stored == '["link", "local"]'  # JSON text on disk

        by_hash = handler.get_packet_by_hash("FANOUT0000000001")
        assert by_hash["tx_radio_id"] == "link"
        assert by_hash["tx_radio_ids"] == ["link", "local"]
        assert handler.get_packet_by_id(by_hash["id"])["tx_radio_ids"] == ["link", "local"]

        for rows in (handler.get_recent_packets(10), handler.get_filtered_packets(limit=10)):
            by = {row["packet_hash"]: row for row in rows}
            assert by["FANOUT0000000001"]["tx_radio_ids"] == ["link", "local"]
            assert by["SINGLE0000000002"]["tx_radio_ids"] is None
    finally:
        shutil.rmtree(d, ignore_errors=True)


def test_fresh_database_also_gets_column_and_index():
    """The fresh-install path must be unaffected by moving the index."""
    d = tempfile.mkdtemp()
    try:
        SQLiteHandler(Path(d))
        path = os.path.join(d, "repeater.db")
        assert {"upstream_hash", "upstream_hash_size"} <= _columns(path)
        assert "idx_packets_upstream_time" in _indexes(path)
    finally:
        shutil.rmtree(d, ignore_errors=True)


def _index_columns(path, name):
    con = sqlite3.connect(path)
    try:
        return tuple(row[2] for row in con.execute("PRAGMA index_info(%s)" % name))
    finally:
        con.close()


def test_upgrade_widens_the_existing_airtime_index_to_radio_ids(legacy_db):
    """An install that already built the four-column index gets it rebuilt.

    CREATE INDEX IF NOT EXISTS leaves an existing index as it is, so without the
    migration an upgraded database would keep the narrow index and the airtime
    chart would read the row heap for every packet in its window.
    """
    d, path = legacy_db
    con = sqlite3.connect(path)
    con.execute(
        "CREATE INDEX idx_packets_airtime ON packets(timestamp, length, payload_length, transmitted)"
    )
    con.commit()
    con.close()

    SQLiteHandler(Path(d))

    assert _index_columns(path, "idx_packets_airtime") == AIRTIME_INDEX_COLUMNS
    con = sqlite3.connect(path)
    try:
        assert con.execute("SELECT timestamp, type, length FROM packets").fetchall() == [
            (1.0, 1, 10)
        ]
    finally:
        con.close()


def test_fresh_database_airtime_index_covers_radio_ids():
    d = tempfile.mkdtemp()
    try:
        SQLiteHandler(Path(d))
        path = os.path.join(d, "repeater.db")
        assert _index_columns(path, "idx_packets_airtime") == AIRTIME_INDEX_COLUMNS
    finally:
        shutil.rmtree(d, ignore_errors=True)