From ccccdb6b7756514b6fd5c4a7c7d17f0a17441537 Mon Sep 17 00:00:00 2001 From: Eric Torres Date: Mon, 18 Mar 2019 10:36:30 -0700 Subject: [PATCH] Allow for means of iteration through Snapshots in a Repository --- rbackup/hierarchy/repository.py | 27 ++++++++- tests/test_repository.py | 102 +++++++++++++++++++++++--------- 2 files changed, 100 insertions(+), 29 deletions(-) diff --git a/rbackup/hierarchy/repository.py b/rbackup/hierarchy/repository.py index 5e0de40..4eb6f20 100644 --- a/rbackup/hierarchy/repository.py +++ b/rbackup/hierarchy/repository.py @@ -48,6 +48,11 @@ class Repository(Hierarchy): * rsync hardlinks unchanged files between snapshots * A symlink in the root of the repository symlinking to the most recent snapshot + + Iteration + --------- + To support checking all snapshots for hardlinking, the Repository class + can be iterated through. """ def __init__(self, dest): @@ -65,10 +70,25 @@ class Repository(Hierarchy): self._curr_snapshot = self._snapshots[-1] def __len__(self): - return len(self._snapshots) + """Return the number of snapshots in this Repository.""" + return len(self.snapshots) def __getitem__(self, position): - return self._snapshots[position] + """Retrieve a Snapshot at a certain index.""" + return self.snapshots[position] + + def __iter__(self): + self.snapshot_index = 0 + return self + + def __next__(self): + """Return the next Snapshot in this Repository.""" + try: + result = self.snapshots[self.snapshot_index] + self.snapshot_index += 1 + return result + except IndexError: + raise StopIteration @property def snapshots(self): @@ -98,6 +118,9 @@ class Repository(Hierarchy): >>> repo = Repository('/tmp') >>> repo.empty True + >>> repo.create_snapshot() + >>> repo.empty + False :rtype: bool """ diff --git a/tests/test_repository.py b/tests/test_repository.py index 0fbd969..b8f6fa8 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -17,6 +17,54 @@ def load_tests(loader, tests, ignore): # ========== Integration Tests ========== +class TestEmptyRepository(unittest.TestCase): + def setUp(self): + self.patched_snapshots = patch( + f"{TESTING_MODULE}.Repository.snapshots", new_callable=PropertyMock + ) + self.mocked_snapshots = self.patched_snapshots.start() + self.mocked_snapshots.return_value = list() + + self.repo_basepath = "backup" + self.repo = Repository(self.repo_basepath) + + def test_curr_snapshot_pre_create(self): + self.assertIsNone(self.repo.curr_snapshot) + self.assertEqual(len(self.repo), 0) + + def test_iteration_pre_create(self): + result = list() + for snapshot in self.repo: + result.append(snapshot.path) + + self.assertListEqual(result, []) + + def test_subscript_pre_create(self): + with self.assertRaises(IndexError): + self.assertRaises(IndexError, self.repo[0]) + + def test_curr_snapshot_post_create(self): + snapshot_name = "new" + new_snapshot = Snapshot(f"backup/data/snapshot-{snapshot_name}") + + self.repo.create_snapshot(snapshot_name) + self.assertEqual(self.repo.curr_snapshot.path, new_snapshot.path) + self.assertEqual(len(self.repo), 1) + self.assertIsInstance(self.repo.curr_snapshot, Snapshot) + + result = list() + for snapshot in self.repo: + result.append(snapshot.path) + + self.assertListEqual(result, [PosixPath("backup/data/snapshot-new")]) + + self.assertEqual(self.repo[0], self.repo.curr_snapshot) + self.assertEqual(self.repo[-1], self.repo.curr_snapshot) + + def tearDown(self): + self.patched_snapshots.stop() + + class TestPopulatedRepository(unittest.TestCase): def setUp(self): self.snapshots = [ @@ -29,7 +77,7 @@ class TestPopulatedRepository(unittest.TestCase): f"{TESTING_MODULE}.Repository.snapshots", new_callable=PropertyMock ) self.mocked_snapshots = self.patched_snapshots.start() - self.mocked_snapshots.return_value = self.snapshots + self.mocked_snapshots.return_value = list(self.snapshots) self.repo_basepath = "backup" self.repo = Repository(self.repo_basepath) @@ -37,47 +85,47 @@ class TestPopulatedRepository(unittest.TestCase): def test_snapshots(self): found_snapshots = [s for s in self.repo.snapshots] self.assertListEqual(found_snapshots, self.snapshots) + self.assertEqual(len(self.repo), len(self.snapshots)) def test_curr_snapshot_pre_create(self): snapshot_name = "third" last_snapshot = Snapshot(f"backup/data/snapshot-{snapshot_name}") self.assertEqual(self.repo.curr_snapshot.path, last_snapshot.path) + self.assertEqual(len(self.repo), len(self.snapshots)) self.assertIsInstance(self.repo.curr_snapshot, Snapshot) + def test_iteration_pre_create(self): + result = list() + for snapshot in self.repo: + result.append(snapshot) + + self.assertListEqual(result, self.snapshots) + + def test_subscript_pre_create(self): + self.assertEqual(self.repo[0], self.snapshots[0]) + self.assertEqual(self.repo[-1], self.snapshots[-1]) + self.assertEqual(self.repo[len(self.repo) - 1], self.snapshots[-1]) + def test_curr_snapshot_post_create(self): snapshot_name = "new" snapshot_path = PosixPath(f"backup/data/snapshot-{snapshot_name}") self.repo.create_snapshot(snapshot_name) self.assertEqual(self.repo.curr_snapshot.path, snapshot_path) + self.assertEqual(len(self.repo), len(self.snapshots) + 1) self.assertIsInstance(self.repo.curr_snapshot, Snapshot) - - def tearDown(self): - self.patched_snapshots.stop() - - -class TestEmptyRepository(unittest.TestCase): - def setUp(self): - self.patched_snapshots = patch( - f"{TESTING_MODULE}.Repository.snapshots", new_callable=PropertyMock - ) - self.mocked_snapshots = self.patched_snapshots.start() - self.mocked_snapshots.return_value = [] - - self.repo_basepath = "backup" - self.repo = Repository(self.repo_basepath) - - def test_curr_snapshot_pre_create(self): - self.assertIsNone(self.repo.curr_snapshot) - - def test_curr_snapshot_post_create(self): - snapshot_name = "new" - new_snapshot = Snapshot(f"backup/data/snapshot-{snapshot_name}") - - self.repo.create_snapshot(snapshot_name) - self.assertEqual(self.repo.curr_snapshot.path, new_snapshot.path) - self.assertIsInstance(self.repo.curr_snapshot, Snapshot) + + def test_iteration_post_create(self): + result = list() + for snapshot in self.repo: + result.append(snapshot) + + self.assertListEqual(result, self.snapshots) + + def test_subscript_post_create(self): + self.assertEqual(self.repo[0], self.snapshots[0]) + self.assertEqual(self.repo[-1], self.repo.curr_snapshot) def tearDown(self): self.patched_snapshots.stop()