mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2026-03-25 10:53:05 -04:00
c7399215ec
- Updated test cases to reflect new column names (eve_MAC -> eveMac, eve_DateTime -> eveDateTime, etc.) across various test files. - Modified SQL table definitions in the database cleanup and migration tests to use camelCase naming conventions. - Implemented migration tests to ensure legacy column names are correctly renamed to camelCase equivalents. - Ensured that existing data is preserved during the migration process and that views referencing old column names are dropped before renaming. - Verified that the migration function is idempotent, allowing for safe re-execution without data loss.
308 lines
11 KiB
Python
308 lines
11 KiB
Python
"""
|
|
Unit tests for migrate_to_camelcase() in db_upgrade.
|
|
|
|
Covers:
|
|
- Already-migrated schema (eveMac present) → skip, return True
|
|
- Unrecognised schema (neither eveMac nor eve_MAC) → skip, return True
|
|
- Legacy Events columns renamed to camelCase equivalents
|
|
- Legacy Sessions columns renamed to camelCase equivalents
|
|
- Legacy Online_History columns renamed to camelCase equivalents
|
|
- Legacy Plugins_Objects columns renamed to camelCase equivalents
|
|
- Legacy Plugins_Language_Strings columns renamed to camelCase equivalents
|
|
- Missing tables are silently skipped without error
|
|
- Existing row data is preserved through the column rename
|
|
- Views referencing old column names are dropped before ALTER TABLE runs
|
|
- Migration is idempotent (second call detects eveMac and returns early)
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import sqlite3
|
|
|
|
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_to_camelcase # noqa: E402
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_cursor():
|
|
"""Return an in-memory SQLite cursor and its parent connection."""
|
|
conn = sqlite3.connect(":memory:")
|
|
return conn.cursor(), conn
|
|
|
|
|
|
def _col_names(cursor, table):
|
|
"""Return the set of column names for a given table."""
|
|
cursor.execute(f'PRAGMA table_info("{table}")')
|
|
return {row[1] for row in cursor.fetchall()}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Legacy DDL fixtures (pre-migration schema with old column names)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_LEGACY_EVENTS_DDL = """
|
|
CREATE TABLE Events (
|
|
eve_MAC TEXT NOT NULL,
|
|
eve_IP TEXT NOT NULL,
|
|
eve_DateTime DATETIME NOT NULL,
|
|
eve_EventType TEXT NOT NULL,
|
|
eve_AdditionalInfo TEXT DEFAULT '',
|
|
eve_PendingAlertEmail INTEGER NOT NULL DEFAULT 1,
|
|
eve_PairEventRowid INTEGER
|
|
)
|
|
"""
|
|
|
|
_LEGACY_SESSIONS_DDL = """
|
|
CREATE TABLE Sessions (
|
|
ses_MAC TEXT,
|
|
ses_IP TEXT,
|
|
ses_EventTypeConnection TEXT,
|
|
ses_DateTimeConnection DATETIME,
|
|
ses_EventTypeDisconnection TEXT,
|
|
ses_DateTimeDisconnection DATETIME,
|
|
ses_StillConnected INTEGER,
|
|
ses_AdditionalInfo TEXT
|
|
)
|
|
"""
|
|
|
|
_LEGACY_ONLINE_HISTORY_DDL = """
|
|
CREATE TABLE Online_History (
|
|
"Index" INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
"Scan_Date" TEXT,
|
|
"Online_Devices" INTEGER,
|
|
"Down_Devices" INTEGER,
|
|
"All_Devices" INTEGER,
|
|
"Archived_Devices" INTEGER,
|
|
"Offline_Devices" INTEGER
|
|
)
|
|
"""
|
|
|
|
_LEGACY_PLUGINS_OBJECTS_DDL = """
|
|
CREATE TABLE Plugins_Objects (
|
|
"Index" INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
Plugin TEXT NOT NULL,
|
|
Object_PrimaryID TEXT NOT NULL,
|
|
Object_SecondaryID TEXT NOT NULL,
|
|
DateTimeCreated TEXT NOT NULL,
|
|
DateTimeChanged TEXT NOT NULL,
|
|
Watched_Value1 TEXT NOT NULL,
|
|
Watched_Value2 TEXT NOT NULL,
|
|
Watched_Value3 TEXT NOT NULL,
|
|
Watched_Value4 TEXT NOT NULL,
|
|
Status TEXT NOT NULL,
|
|
Extra TEXT NOT NULL,
|
|
UserData TEXT NOT NULL,
|
|
ForeignKey TEXT NOT NULL,
|
|
SyncHubNodeName TEXT,
|
|
HelpVal1 TEXT,
|
|
HelpVal2 TEXT,
|
|
HelpVal3 TEXT,
|
|
HelpVal4 TEXT,
|
|
ObjectGUID TEXT
|
|
)
|
|
"""
|
|
|
|
_LEGACY_PLUGINS_LANG_DDL = """
|
|
CREATE TABLE Plugins_Language_Strings (
|
|
"Index" INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
Language_Code TEXT NOT NULL,
|
|
String_Key TEXT NOT NULL,
|
|
String_Value TEXT NOT NULL,
|
|
Extra TEXT NOT NULL
|
|
)
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMigrateToCamelCase:
|
|
|
|
def test_returns_true_if_already_camelcase(self):
|
|
"""DB already on camelCase schema → skip silently, return True."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute("""
|
|
CREATE TABLE Events (
|
|
eveMac TEXT NOT NULL, eveIp TEXT NOT NULL,
|
|
eveDateTime DATETIME NOT NULL, eveEventType TEXT NOT NULL,
|
|
eveAdditionalInfo TEXT, evePendingAlertEmail INTEGER,
|
|
evePairEventRowid INTEGER
|
|
)
|
|
""")
|
|
conn.commit()
|
|
|
|
result = migrate_to_camelcase(cur)
|
|
|
|
assert result is True
|
|
assert "eveMac" in _col_names(cur, "Events")
|
|
|
|
def test_returns_true_if_unknown_schema(self):
|
|
"""Events exists but has neither eve_MAC nor eveMac → skip, return True."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute("CREATE TABLE Events (someOtherCol TEXT)")
|
|
conn.commit()
|
|
|
|
result = migrate_to_camelcase(cur)
|
|
|
|
assert result is True
|
|
|
|
def test_events_legacy_columns_renamed(self):
|
|
"""All legacy eve_* columns are renamed to their camelCase equivalents."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
conn.commit()
|
|
|
|
result = migrate_to_camelcase(cur)
|
|
|
|
assert result is True
|
|
cols = _col_names(cur, "Events")
|
|
expected_new = {
|
|
"eveMac", "eveIp", "eveDateTime", "eveEventType",
|
|
"eveAdditionalInfo", "evePendingAlertEmail", "evePairEventRowid",
|
|
}
|
|
old_names = {
|
|
"eve_MAC", "eve_IP", "eve_DateTime", "eve_EventType",
|
|
"eve_AdditionalInfo", "eve_PendingAlertEmail", "eve_PairEventRowid",
|
|
}
|
|
assert expected_new.issubset(cols), f"Missing new columns: {expected_new - cols}"
|
|
assert not old_names & cols, f"Old columns still present: {old_names & cols}"
|
|
|
|
def test_sessions_legacy_columns_renamed(self):
|
|
"""All legacy ses_* columns are renamed to their camelCase equivalents."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
cur.execute(_LEGACY_SESSIONS_DDL)
|
|
conn.commit()
|
|
|
|
result = migrate_to_camelcase(cur)
|
|
|
|
assert result is True
|
|
cols = _col_names(cur, "Sessions")
|
|
assert {
|
|
"sesMac", "sesIp", "sesEventTypeConnection", "sesDateTimeConnection",
|
|
"sesEventTypeDisconnection", "sesDateTimeDisconnection",
|
|
"sesStillConnected", "sesAdditionalInfo",
|
|
}.issubset(cols)
|
|
assert not {"ses_MAC", "ses_IP", "ses_DateTimeConnection"} & cols
|
|
|
|
def test_online_history_legacy_columns_renamed(self):
|
|
"""Quoted legacy Online_History column names are renamed to camelCase."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
cur.execute(_LEGACY_ONLINE_HISTORY_DDL)
|
|
conn.commit()
|
|
|
|
migrate_to_camelcase(cur)
|
|
|
|
cols = _col_names(cur, "Online_History")
|
|
assert {
|
|
"scanDate", "onlineDevices", "downDevices",
|
|
"allDevices", "archivedDevices", "offlineDevices",
|
|
}.issubset(cols)
|
|
assert not {"Scan_Date", "Online_Devices", "Down_Devices"} & cols
|
|
|
|
def test_plugins_objects_legacy_columns_renamed(self):
|
|
"""All renamed Plugins_Objects columns receive their camelCase names."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
cur.execute(_LEGACY_PLUGINS_OBJECTS_DDL)
|
|
conn.commit()
|
|
|
|
migrate_to_camelcase(cur)
|
|
|
|
cols = _col_names(cur, "Plugins_Objects")
|
|
assert {
|
|
"plugin", "objectPrimaryId", "objectSecondaryId",
|
|
"dateTimeCreated", "dateTimeChanged",
|
|
"watchedValue1", "watchedValue2", "watchedValue3", "watchedValue4",
|
|
"status", "extra", "userData", "foreignKey", "syncHubNodeName",
|
|
"helpVal1", "helpVal2", "helpVal3", "helpVal4", "objectGuid",
|
|
}.issubset(cols)
|
|
assert not {
|
|
"Object_PrimaryID", "Watched_Value1", "ObjectGUID",
|
|
"ForeignKey", "UserData", "Plugin",
|
|
} & cols
|
|
|
|
def test_plugins_language_strings_renamed(self):
|
|
"""Plugins_Language_Strings legacy column names are renamed to camelCase."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
cur.execute(_LEGACY_PLUGINS_LANG_DDL)
|
|
conn.commit()
|
|
|
|
migrate_to_camelcase(cur)
|
|
|
|
cols = _col_names(cur, "Plugins_Language_Strings")
|
|
assert {"languageCode", "stringKey", "stringValue", "extra"}.issubset(cols)
|
|
assert not {"Language_Code", "String_Key", "String_Value"} & cols
|
|
|
|
def test_missing_table_silently_skipped(self):
|
|
"""Tables in the migration map that don't exist are skipped without error."""
|
|
cur, conn = _make_cursor()
|
|
# Only Events (legacy) exists — all other mapped tables are absent
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
conn.commit()
|
|
|
|
result = migrate_to_camelcase(cur)
|
|
|
|
assert result is True
|
|
assert "eveMac" in _col_names(cur, "Events")
|
|
|
|
def test_data_preserved_after_rename(self):
|
|
"""Existing rows remain accessible under the new camelCase column names."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
cur.execute(
|
|
"INSERT INTO Events (eve_MAC, eve_IP, eve_DateTime, eve_EventType) "
|
|
"VALUES ('aa:bb:cc:dd:ee:ff', '192.168.1.1', '2025-01-01 12:00:00', 'Connected')"
|
|
)
|
|
conn.commit()
|
|
|
|
migrate_to_camelcase(cur)
|
|
|
|
cur.execute(
|
|
"SELECT eveMac, eveIp, eveEventType FROM Events WHERE eveMac = 'aa:bb:cc:dd:ee:ff'"
|
|
)
|
|
row = cur.fetchone()
|
|
assert row is not None, "Row missing after camelCase migration"
|
|
assert row[0] == "aa:bb:cc:dd:ee:ff"
|
|
assert row[1] == "192.168.1.1"
|
|
assert row[2] == "Connected"
|
|
|
|
def test_views_dropped_before_migration(self):
|
|
"""Views referencing old column names do not block ALTER TABLE RENAME COLUMN."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
# A view that references old column names would normally block the rename
|
|
cur.execute("CREATE VIEW Events_Devices AS SELECT eve_MAC, eve_IP FROM Events")
|
|
conn.commit()
|
|
|
|
result = migrate_to_camelcase(cur)
|
|
|
|
assert result is True
|
|
assert "eveMac" in _col_names(cur, "Events")
|
|
# View is dropped (ensure_views() is responsible for recreation separately)
|
|
cur.execute("SELECT name FROM sqlite_master WHERE type='view' AND name='Events_Devices'")
|
|
assert cur.fetchone() is None
|
|
|
|
def test_idempotent_second_run(self):
|
|
"""Running migration twice is safe — second call detects eveMac and exits early."""
|
|
cur, conn = _make_cursor()
|
|
cur.execute(_LEGACY_EVENTS_DDL)
|
|
conn.commit()
|
|
|
|
first = migrate_to_camelcase(cur)
|
|
second = migrate_to_camelcase(cur)
|
|
|
|
assert first is True
|
|
assert second is True
|
|
cols = _col_names(cur, "Events")
|
|
assert "eveMac" in cols
|
|
assert "eve_MAC" not in cols
|