etorres/rbackup
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Allow for means of iteration through Snapshots in a Repository

Browse Source
This commit is contained in:
Eric Torres 2019-03-18 10:36:30 -07:00
parent 6d4bc32c4e
commit ccccdb6b77
2 changed files with 100 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
rbackup/hierarchy/repository.py
View File

@ -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
"""

102
tests/test_repository.py
View File

@ -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()