NetAlertX/test/db/test_timestamp_migration.py

"""
Unit tests for database timestamp migration to UTC.

Tests verify that:
- Migration detects version correctly from Settings table
- Fresh installs skip migration (empty VERSION)
- Upgrades from v26.2.6+ skip migration (already UTC)
- Upgrades from <v26.2.6 run migration (convert local→UTC)
- Migration handles timezone offset calculations correctly
- Migration is idempotent (safe to run multiple times)
"""

import sys
import os
import pytest
import sqlite3
import tempfile

INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])

from db.db_upgrade import migrate_timestamps_to_utc, is_timestamps_in_utc  # noqa: E402
from utils.datetime_utils import timeNowUTC  # noqa: E402


@pytest.fixture
def temp_db():
    """Create a temporary database for testing"""
    fd, db_path = tempfile.mkstemp(suffix='.db')
    os.close(fd)
    
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # Create Settings table
    cursor.execute("""
        CREATE TABLE Settings (
            setKey TEXT PRIMARY KEY,
            setValue TEXT
        )
    """)
    
    # Create Devices table with timestamp columns
    cursor.execute("""
        CREATE TABLE Devices (
            devMac TEXT PRIMARY KEY,
            devFirstConnection TEXT,
            devLastConnection TEXT,
            devLastNotification TEXT
        )
    """)
    
    conn.commit()
    
    yield cursor, conn
    
    conn.close()
    os.unlink(db_path)


class TestTimestampMigration:
    """Test suite for UTC timestamp migration"""

    def test_migrate_fresh_install_skips_migration(self, temp_db):
        """Test that fresh install with empty VERSION skips migration"""
        cursor, conn = temp_db
        
        # Empty Settings table (fresh install)
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True
        # Should return without error

    def test_migrate_unknown_version_skips_migration(self, temp_db):
        """Test that 'unknown' VERSION skips migration"""
        cursor, conn = temp_db
        
        # Insert 'unknown' VERSION
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', 'unknown')")
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_migrate_version_26_2_6_skips_migration(self, temp_db):
        """Test that v26.2.6 skips migration (already UTC)"""
        cursor, conn = temp_db
        
        # Insert VERSION v26.2.6
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', '26.2.6')")
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_migrate_version_27_0_0_skips_migration(self, temp_db):
        """Test that v27.0.0 skips migration (newer version)"""
        cursor, conn = temp_db
        
        # Insert VERSION v27.0.0
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', '27.0.0')")
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_migrate_version_26_3_0_skips_migration(self, temp_db):
        """Test that v26.3.0 skips migration (newer minor version)"""
        cursor, conn = temp_db
        
        # Insert VERSION v26.3.0
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', '26.3.0')")
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_migrate_old_version_triggers_migration(self, temp_db):
        """Test that v25.x.x triggers migration"""
        cursor, conn = temp_db
        
        # Insert VERSION v25.1.0
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', '25.1.0')")
        
        # Insert a sample device with timestamp
        now_str = timeNowUTC()
        cursor.execute("""
            INSERT INTO Devices (devMac, devFirstConnection, devLastConnection)
            VALUES ('aa:bb:cc:dd:ee:ff', ?, ?)
        """, (now_str, now_str))
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_migrate_version_with_v_prefix(self, temp_db):
        """Test that version string with 'v' prefix is parsed correctly"""
        cursor, conn = temp_db
        
        # Insert VERSION with 'v' prefix
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', 'v26.2.6')")
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_migrate_malformed_version_uses_fallback(self, temp_db):
        """Test that malformed version string uses timestamp detection fallback"""
        cursor, conn = temp_db
        
        # Insert malformed VERSION
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', 'invalid.version')")
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        # Should not crash, should use fallback detection
        assert result is True

    def test_migrate_version_26_2_5_triggers_migration(self, temp_db):
        """Test that v26.2.5 (one patch before UTC) triggers migration"""
        cursor, conn = temp_db
        
        # Insert VERSION v26.2.5
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', '26.2.5')")
        
        # Insert sample device
        now_str = timeNowUTC()
        cursor.execute("""
            INSERT INTO Devices (devMac, devFirstConnection)
            VALUES ('aa:bb:cc:dd:ee:ff', ?)
        """, (now_str,))
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_migrate_does_not_crash_on_empty_devices_table(self, temp_db):
        """Test that migration handles empty Devices table gracefully"""
        cursor, conn = temp_db
        
        # Insert old VERSION but no devices
        cursor.execute("INSERT INTO Settings (setKey, setValue) VALUES ('VERSION', '25.1.0')")
        conn.commit()
        
        result = migrate_timestamps_to_utc(cursor)
        
        assert result is True

    def test_is_timestamps_in_utc_returns_true_for_empty_table(self, temp_db):
        """Test that is_timestamps_in_utc returns True for empty Devices table"""
        cursor, conn = temp_db
        
        result = is_timestamps_in_utc(cursor)
        
        assert result is True

    def test_is_timestamps_in_utc_detects_utc_timestamps(self, temp_db):
        """Test that is_timestamps_in_utc correctly identifies UTC timestamps"""
        cursor, conn = temp_db
        
        # Insert devices with UTC timestamps
        now_str = timeNowUTC()
        cursor.execute("""
            INSERT INTO Devices (devMac, devFirstConnection)
            VALUES ('aa:bb:cc:dd:ee:ff', ?)
        """, (now_str,))
        conn.commit()
        
        result = is_timestamps_in_utc(cursor)
        
        # Should return False for naive timestamps (no timezone marker)
        # This is expected behavior - naive timestamps need migration check
        assert result is False

    def test_is_timestamps_in_utc_detects_timezone_markers(self, temp_db):
        """Test that is_timestamps_in_utc detects timestamps with timezone info"""
        cursor, conn = temp_db
        
        # Insert device with timezone marker
        timestamp_with_tz = "2026-02-11 11:37:02+00:00"
        cursor.execute("""
            INSERT INTO Devices (devMac, devFirstConnection)
            VALUES ('aa:bb:cc:dd:ee:ff', ?)
        """, (timestamp_with_tz,))
        conn.commit()
        
        result = is_timestamps_in_utc(cursor)
        
        # Should detect timezone marker
        assert result is True