1 files changed, 256 insertions, 0 deletions
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)