summaryrefslogtreecommitdiffstats
path: root/src/pybind/mgr/snap_schedule/tests
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/pybind/mgr/snap_schedule/tests/__init__.py0
-rw-r--r--src/pybind/mgr/snap_schedule/tests/conftest.py34
-rw-r--r--src/pybind/mgr/snap_schedule/tests/fs/__init__.py0
-rw-r--r--src/pybind/mgr/snap_schedule/tests/fs/test_schedule.py256
-rw-r--r--src/pybind/mgr/snap_schedule/tests/fs/test_schedule_client.py37
5 files changed, 327 insertions, 0 deletions
diff --git a/src/pybind/mgr/snap_schedule/tests/__init__.py b/src/pybind/mgr/snap_schedule/tests/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/src/pybind/mgr/snap_schedule/tests/__init__.py
diff --git a/src/pybind/mgr/snap_schedule/tests/conftest.py b/src/pybind/mgr/snap_schedule/tests/conftest.py
new file mode 100644
index 000000000..35255b8d4
--- /dev/null
+++ b/src/pybind/mgr/snap_schedule/tests/conftest.py
@@ -0,0 +1,34 @@
+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)
+ return db
diff --git a/src/pybind/mgr/snap_schedule/tests/fs/__init__.py b/src/pybind/mgr/snap_schedule/tests/fs/__init__.py
new file mode 100644
index 000000000..e69de29bb
--- /dev/null
+++ b/src/pybind/mgr/snap_schedule/tests/fs/__init__.py
diff --git a/src/pybind/mgr/snap_schedule/tests/fs/test_schedule.py b/src/pybind/mgr/snap_schedule/tests/fs/test_schedule.py
new file mode 100644
index 000000000..1e984ab64
--- /dev/null
+++ b/src/pybind/mgr/snap_schedule/tests/fs/test_schedule.py
@@ -0,0 +1,256 @@
+import datetime
+import json
+import pytest
+import random
+import sqlite3
+from ...fs.schedule import Schedule, parse_retention
+
+SELECT_ALL = ('select * from schedules s'
+ ' INNER JOIN schedules_meta sm'
+ ' ON sm.schedule_id = s.id')
+
+
+def assert_updated(new, old, update_expected={}):
+ '''
+ This helper asserts that an object new has been updated in the
+ attributes in the dict updated AND has not changed in other attributes
+ compared to old.
+ if update expected is the empty dict, equality is checked
+ '''
+
+ for var in vars(new):
+ if var in update_expected:
+ expected_val = update_expected.get(var)
+ new_val = getattr(new, var)
+ if isinstance(expected_val, datetime.datetime):
+ assert new_val.year == expected_val.year
+ assert new_val.month == expected_val.month
+ assert new_val.day == expected_val.day
+ assert new_val.hour == expected_val.hour
+ assert new_val.minute == expected_val.minute
+ assert new_val.second == expected_val.second
+ else:
+ assert new_val == expected_val, f'new did not update value for {var}'
+ else:
+ expected_val = getattr(old, var)
+ new_val = getattr(new, var)
+ if isinstance(expected_val, datetime.datetime):
+ assert new_val.year == expected_val.year
+ assert new_val.month == expected_val.month
+ assert new_val.day == expected_val.day
+ assert new_val.hour == expected_val.hour
+ assert new_val.minute == expected_val.minute
+ assert new_val.second == expected_val.second
+ else:
+ assert new_val == expected_val, f'new changed unexpectedly in value for {var}'
+
+
+class TestSchedule(object):
+ '''
+ Test the schedule class basics and that its methods update self as expected
+ '''
+
+ def test_start_default_midnight(self, simple_schedule):
+ now = datetime.datetime.now(datetime.timezone.utc)
+ assert simple_schedule.start.second == 0
+ assert simple_schedule.start.minute == 0
+ assert simple_schedule.start.hour == 0
+ assert simple_schedule.start.day == now.day
+ assert simple_schedule.start.month == now.month
+ assert simple_schedule.start.year == now.year
+ assert simple_schedule.start.tzinfo == now.tzinfo
+
+ def test_created_now(self, simple_schedule):
+ now = datetime.datetime.now(datetime.timezone.utc)
+ assert simple_schedule.created.minute == now.minute
+ assert simple_schedule.created.hour == now.hour
+ assert simple_schedule.created.day == now.day
+ assert simple_schedule.created.month == now.month
+ assert simple_schedule.created.year == now.year
+ assert simple_schedule.created.tzinfo == now.tzinfo
+
+ def test_repeat_valid(self, simple_schedule):
+ repeat = simple_schedule.repeat
+ assert isinstance(repeat, int)
+
+ def test_store_single(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+ row = ()
+ with db:
+ row = db.execute(SELECT_ALL).fetchone()
+
+ db_schedule = Schedule._from_db_row(row, simple_schedule.fs)
+
+ assert_updated(db_schedule, simple_schedule)
+
+ def test_store_multiple(self, db, simple_schedules):
+ [s.store_schedule(db) for s in simple_schedules]
+
+ rows = []
+ with db:
+ rows = db.execute(SELECT_ALL).fetchall()
+
+ assert len(rows) == len(simple_schedules)
+
+ def test_update_last(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+
+ with db:
+ _ = db.execute(SELECT_ALL).fetchone()
+
+ first_time = datetime.datetime.now(datetime.timezone.utc)
+ simple_schedule.update_last(first_time, db)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after, simple_schedule.fs),
+ simple_schedule)
+
+ second_time = datetime.datetime.now(datetime.timezone.utc)
+ simple_schedule.update_last(second_time, db)
+
+ with db:
+ after2 = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after2, simple_schedule.fs),
+ simple_schedule)
+
+ def test_set_inactive_active(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+
+ with db:
+ _ = db.execute(SELECT_ALL).fetchone()
+
+ simple_schedule.set_inactive(db)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after, simple_schedule.fs),
+ simple_schedule)
+
+ simple_schedule.set_active(db)
+
+ with db:
+ after2 = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after2, simple_schedule.fs),
+ simple_schedule)
+
+ def test_update_pruned(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+
+ with db:
+ _ = db.execute(SELECT_ALL).fetchone()
+
+ now = datetime.datetime.now(datetime.timezone.utc)
+ pruned_count = random.randint(1, 1000)
+
+ simple_schedule.update_pruned(now, db, pruned_count)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+
+ assert_updated(Schedule._from_db_row(after, simple_schedule.fs),
+ simple_schedule)
+
+ # TODO test get_schedules and list_schedules
+
+
+class TestScheduleDB(object):
+ '''
+ This class tests that Schedules methods update the DB correctly
+ '''
+
+ def test_update_last(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+
+ with db:
+ before = db.execute(SELECT_ALL).fetchone()
+
+ first_time = datetime.datetime.now(datetime.timezone.utc)
+ simple_schedule.update_last(first_time, db)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after, simple_schedule.fs),
+ Schedule._from_db_row(before, simple_schedule.fs),
+ {'created_count': 1,
+ 'last': first_time,
+ 'first': first_time})
+
+ second_time = datetime.datetime.now(datetime.timezone.utc)
+ simple_schedule.update_last(second_time, db)
+
+ with db:
+ after2 = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after2, simple_schedule.fs),
+ Schedule._from_db_row(after, simple_schedule.fs),
+ {'created_count': 2, 'last': second_time})
+
+ def test_set_inactive_active(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+
+ with db:
+ before = db.execute(SELECT_ALL).fetchone()
+
+ simple_schedule.set_inactive(db)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after, simple_schedule.fs),
+ Schedule._from_db_row(before, simple_schedule.fs),
+ {'active': 0})
+
+ simple_schedule.set_active(db)
+
+ with db:
+ after2 = db.execute(SELECT_ALL).fetchone()
+ assert_updated(Schedule._from_db_row(after2, simple_schedule.fs),
+ Schedule._from_db_row(after, simple_schedule.fs),
+ {'active': 1})
+
+ def test_update_pruned(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+
+ with db:
+ before = db.execute(SELECT_ALL).fetchone()
+
+ now = datetime.datetime.now(datetime.timezone.utc)
+ pruned_count = random.randint(1, 1000)
+
+ simple_schedule.update_pruned(now, db, pruned_count)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+
+ assert_updated(Schedule._from_db_row(after, simple_schedule.fs),
+ Schedule._from_db_row(before, simple_schedule.fs),
+ {'last_pruned': now, 'pruned_count': pruned_count})
+
+ def test_add_retention(self, db, simple_schedule):
+ simple_schedule.store_schedule(db)
+
+ with db:
+ before = db.execute(SELECT_ALL).fetchone()
+
+ retention = "7d12m"
+ simple_schedule.add_retention(db, simple_schedule.path, retention)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+
+ assert after['retention'] == json.dumps(parse_retention(retention))
+
+ retention2 = "4w"
+ simple_schedule.add_retention(db, simple_schedule.path, retention2)
+
+ with db:
+ after = db.execute(SELECT_ALL).fetchone()
+
+ assert after['retention'] == json.dumps(parse_retention(retention + retention2))
+
+ def test_per_path_and_repeat_uniqness(self, db):
+ s1 = Schedule(*('/foo', '24h', 'fs_name', '/foo'))
+ s2 = Schedule(*('/foo', '1d', 'fs_name', '/foo'))
+
+ s1.store_schedule(db)
+ with pytest.raises(sqlite3.IntegrityError):
+ s2.store_schedule(db)
diff --git a/src/pybind/mgr/snap_schedule/tests/fs/test_schedule_client.py b/src/pybind/mgr/snap_schedule/tests/fs/test_schedule_client.py
new file mode 100644
index 000000000..02996146b
--- /dev/null
+++ b/src/pybind/mgr/snap_schedule/tests/fs/test_schedule_client.py
@@ -0,0 +1,37 @@
+from datetime import datetime, timedelta
+from unittest.mock import MagicMock
+import pytest
+from ...fs.schedule_client import get_prune_set, SNAPSHOT_TS_FORMAT
+
+
+class TestScheduleClient(object):
+
+ def test_get_prune_set_empty_retention_no_prune(self):
+ now = datetime.now()
+ candidates = set()
+ for i in range(10):
+ ts = now - timedelta(minutes=i*5)
+ fake_dir = MagicMock()
+ fake_dir.d_name = f'scheduled-{ts.strftime(SNAPSHOT_TS_FORMAT)}'
+ candidates.add((fake_dir, ts))
+ ret = {}
+ prune_set = get_prune_set(candidates, ret)
+ assert prune_set == set(), 'candidates are pruned despite empty retention'
+
+ def test_get_prune_set_two_retention_specs(self):
+ now = datetime.now()
+ candidates = set()
+ for i in range(10):
+ ts = now - timedelta(hours=i*1)
+ fake_dir = MagicMock()
+ fake_dir.d_name = f'scheduled-{ts.strftime(SNAPSHOT_TS_FORMAT)}'
+ candidates.add((fake_dir, ts))
+ for i in range(10):
+ ts = now - timedelta(days=i*1)
+ fake_dir = MagicMock()
+ fake_dir.d_name = f'scheduled-{ts.strftime(SNAPSHOT_TS_FORMAT)}'
+ candidates.add((fake_dir, ts))
+ # should keep 8 snapshots
+ ret = {'h': 6, 'd': 2}
+ prune_set = get_prune_set(candidates, ret)
+ assert len(prune_set) == len(candidates) - 8, 'wrong size of prune set'