Developer
83 lines
3.3 KiB
Python
83 lines
3.3 KiB
Python
import sqlite3
|
|
|
|
from cryptography.fernet import Fernet
|
|
|
|
from media_library_viewer_api.services.secrets import decrypt_secrets, reset_encryption_key_cache
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
|
|
|
|
def test_migrates_legacy_ssh_machine_and_ssh_task_service(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("MANAGE_ENCRYPTION_KEY", Fernet.generate_key().decode())
|
|
reset_encryption_key_cache()
|
|
db_path = tmp_path / "settings.sqlite"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript("""
|
|
CREATE TABLE monitoring_machines (
|
|
id TEXT PRIMARY KEY, name TEXT, mode TEXT, enabled INTEGER,
|
|
config_json TEXT, created_at INTEGER, updated_at INTEGER
|
|
);
|
|
CREATE TABLE services (
|
|
id TEXT PRIMARY KEY, service_type TEXT, name TEXT, config_json TEXT,
|
|
secrets_json TEXT, enabled INTEGER, created_at INTEGER, updated_at INTEGER
|
|
);
|
|
""")
|
|
conn.execute(
|
|
"INSERT INTO monitoring_machines VALUES (?, ?, 'ssh', 1, ?, 10, 11)",
|
|
(
|
|
"remote-1",
|
|
"Storage",
|
|
'{"host":"storage","port":2222,"username":"ops","ssh_private_key":"PRIVATE","ssh_private_key_passphrase":"phrase","password":"pw"}',
|
|
),
|
|
)
|
|
conn.execute("INSERT INTO monitoring_machines VALUES (?, ?, 'local', 1, '{}', 10, 11)", ("local", "This machine"))
|
|
conn.execute("INSERT INTO services VALUES ('task-service', 'ssh_tasks', 'Tasks', '{}', '{}', 1, 1, 1)")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
store = SettingsStore(db_path)
|
|
store.init_schema()
|
|
remote = store.get_service("remote-1")
|
|
assert remote and remote["service_type"] == "remote_machine"
|
|
assert remote["config"] == {
|
|
"host": "storage",
|
|
"port": 2222,
|
|
"username": "ops",
|
|
"ssh_key_id": "legacy-key-remote-1",
|
|
"timeout_seconds": 30,
|
|
}
|
|
assert decrypt_secrets(remote["secrets"]) == {"passphrase": "phrase", "password": "pw"}
|
|
assert store.get_ssh_key("legacy-key-remote-1")["private_key"] == "PRIVATE"
|
|
assert store.get_service("task-service")["service_type"] == "remote_machine"
|
|
with store.connect() as check:
|
|
assert (
|
|
check.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='monitoring_machines'").fetchone()
|
|
is None
|
|
)
|
|
store.init_schema()
|
|
assert store.get_service("remote-1")["id"] == "remote-1"
|
|
|
|
|
|
def test_migrates_task_default_service_id_to_service_id(tmp_path):
|
|
db_path = tmp_path / "settings.sqlite"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE saved_tasks (
|
|
id TEXT PRIMARY KEY, name TEXT NOT NULL, task_type TEXT NOT NULL,
|
|
content TEXT NOT NULL, enabled INTEGER NOT NULL, default_service_id TEXT NOT NULL,
|
|
notes TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
conn.execute("INSERT INTO saved_tasks VALUES ('task-1', 'Check', 'shell', 'true', 1, 'remote-1', '', 1, 1)")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
store = SettingsStore(db_path)
|
|
store.init_schema()
|
|
assert store.get_task("task-1")["service_id"] == "remote-1"
|
|
with store.connect() as check:
|
|
columns = {row[1] for row in check.execute("PRAGMA table_info(saved_tasks)")}
|
|
assert "service_id" in columns
|
|
assert "default_service_id" not in columns
|