1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
import pytest
import sqlite3
from ..fs.schedule import Schedule
# simple_schedule fixture returns schedules without any timing arguments
# the tuple values correspong to ctor args for Schedule
_simple_schedules = [
('/foo', '6h', 'fs_name', '/foo'),
('/foo', '24h', 'fs_name', '/foo'),
('/bar', '1d', 'fs_name', '/bar'),
('/fnord', '1w', 'fs_name', '/fnord'),
]
@pytest.fixture(params=_simple_schedules)
def simple_schedule(request):
return Schedule(*request.param)
@pytest.fixture
def simple_schedules():
return [Schedule(*s) for s in _simple_schedules]
@pytest.fixture
def db():
db = sqlite3.connect(':memory:',
check_same_thread=False)
with db:
db.row_factory = sqlite3.Row
db.execute("PRAGMA FOREIGN_KEYS = 1")
db.executescript(Schedule.CREATE_TABLES)
_create_snap_schedule_kv_db(db)
_upgrade_snap_schedule_db_schema(db)
return db
def _create_snap_schedule_kv_db(db):
SQL = """
CREATE TABLE IF NOT EXISTS SnapScheduleModuleKV (
key TEXT PRIMARY KEY,
value NOT NULL
) WITHOUT ROWID;
INSERT OR IGNORE INTO SnapScheduleModuleKV (key, value) VALUES ('__snap_schedule_db_version', 1);
"""
db.executescript(SQL)
def _get_snap_schedule_db_version(db):
SQL = """
SELECT value
FROM SnapScheduleModuleKV
WHERE key = '__snap_schedule_db_version';
"""
cur = db.execute(SQL)
row = cur.fetchone()
assert row is not None
return int(row[0])
# add all upgrades here
def _upgrade_snap_schedule_db_schema(db):
# add a column to hold the subvolume group name
if _get_snap_schedule_db_version(db) < 2:
SQL = """
ALTER TABLE schedules
ADD COLUMN group_name TEXT;
"""
db.executescript(SQL)
# bump up the snap-schedule db version to 2
SQL = "UPDATE OR ROLLBACK SnapScheduleModuleKV SET value = ? WHERE key = '__snap_schedule_db_version';"
db.execute(SQL, (2,))
|
