Utilize pathlib over os.path for internal path handling
This commit is contained in:
parent
eab5bb108c
commit
30cb9e03a4
@ -48,7 +48,7 @@ Backup Directory Hierarchy
|
||||
|
||||
Implementation Notes
|
||||
--------------------
|
||||
* os.path is used for path handling
|
||||
* pathlib is used for path handling
|
||||
* Use --link-dest=
|
||||
* Use --suffix=, --backup, and --backup-dir=
|
||||
|
||||
|
@ -3,13 +3,21 @@
|
||||
.. module:: rbackup.hierarchy.hierarchy
|
||||
:synopsis: Classes for creating the backup hierarchy.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# ========== Classes ==========
|
||||
class Hierarchy:
|
||||
"""A class for organizing the backup root hierarchy.
|
||||
|
||||
|
||||
Upon creation of a Hierarchy object, it is up to the caller
|
||||
to call either shutil.mkdir() or a related method to create
|
||||
the directory structure it emulates.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
* path
|
||||
* name
|
||||
"""
|
||||
|
||||
def __init__(self, dest):
|
||||
@ -17,14 +25,14 @@ class Hierarchy:
|
||||
|
||||
Example
|
||||
-------
|
||||
>>> hier = Hierarchy('/tmp')
|
||||
>>> hier = Hierarchy('backup')
|
||||
>>> hier.path
|
||||
'/tmp'
|
||||
PosixPath('backup')
|
||||
|
||||
:param dest: the root directory of the backup hierarchy
|
||||
:type dest: str, bytes
|
||||
:type dest: str, path-like object
|
||||
"""
|
||||
self._dest = dest
|
||||
self._path = Path(dest)
|
||||
|
||||
@property
|
||||
def path(self):
|
||||
@ -32,13 +40,27 @@ class Hierarchy:
|
||||
|
||||
Example
|
||||
-------
|
||||
>>> hier = Hierarchy('/tmp')
|
||||
>>> hier = Hierarchy('backup')
|
||||
>>> hier.path
|
||||
'/tmp'
|
||||
PosixPath('backup')
|
||||
|
||||
:rtype: path-like object
|
||||
"""
|
||||
return self._path
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of this hierarchy.
|
||||
|
||||
Example
|
||||
-------
|
||||
>>> hier = Hierarchy('backup/data/snapshot-one')
|
||||
>>> hier.name
|
||||
'snapshot-one'
|
||||
|
||||
:rtype: str
|
||||
"""
|
||||
return self._dest
|
||||
return self.path.name
|
||||
|
||||
|
||||
# ========== Functions ==========
|
||||
|
@ -4,9 +4,7 @@
|
||||
:synopsis: Class for structuring a backup repository.
|
||||
"""
|
||||
import logging
|
||||
import os.path
|
||||
import datetime
|
||||
import glob
|
||||
|
||||
from rbackup.hierarchy.hierarchy import Hierarchy
|
||||
from rbackup.hierarchy.snapshot import Snapshot
|
||||
@ -32,6 +30,7 @@ class Repository(Hierarchy):
|
||||
Attributes
|
||||
----------
|
||||
* path (inherited from Hierarchy)
|
||||
* name (inherited from Hierarchy)
|
||||
* curr_snapshot - return either the most recent snapshot
|
||||
before running create_snapshot() or the new snapshot
|
||||
created after running create_snapshot()
|
||||
@ -40,8 +39,6 @@ class Repository(Hierarchy):
|
||||
Methods
|
||||
-------
|
||||
* create_snapshot() - create a new snapshot, then update curr_snapshot
|
||||
* update_snapshots() - update the list of snapshots this repository
|
||||
contains
|
||||
|
||||
Directory Structure
|
||||
-------------------
|
||||
@ -57,14 +54,22 @@ class Repository(Hierarchy):
|
||||
"""Default constructor for the Repository class."""
|
||||
super().__init__(dest)
|
||||
|
||||
self._snapshot_dir = os.path.join(self.path, "data")
|
||||
self.update_snapshots()
|
||||
self._snapshot_dir = self.path / "data"
|
||||
self._snapshots = [
|
||||
Snapshot(s) for s in self._snapshot_dir.glob("*") if s.is_dir()
|
||||
]
|
||||
|
||||
if self._snapshots == []:
|
||||
self._curr_snapshot = None
|
||||
else:
|
||||
self._curr_snapshot = self._snapshots[-1]
|
||||
|
||||
def __len__(self):
|
||||
return len(self._snapshots)
|
||||
|
||||
def __getitem__(self, position):
|
||||
return self._snapshots[position]
|
||||
|
||||
@property
|
||||
def snapshots(self):
|
||||
"""Return a list of snapshots stored in this Repository.
|
||||
@ -96,7 +101,7 @@ class Repository(Hierarchy):
|
||||
|
||||
:rtype: bool
|
||||
"""
|
||||
return self._snapshots == []
|
||||
return self.snapshots == []
|
||||
|
||||
@property
|
||||
def curr_snapshot(self):
|
||||
@ -116,7 +121,10 @@ class Repository(Hierarchy):
|
||||
|
||||
:rtype: Snapshot object
|
||||
"""
|
||||
return self._curr_snapshot
|
||||
try:
|
||||
return self.snapshots[-1]
|
||||
except IndexError:
|
||||
return None
|
||||
|
||||
def create_snapshot(
|
||||
self, name=datetime.datetime.utcnow().isoformat().replace(":", "-")
|
||||
@ -128,7 +136,6 @@ class Repository(Hierarchy):
|
||||
|
||||
Example
|
||||
-------
|
||||
directive ::= "#" "doctest":" ELLIPSIS
|
||||
>>> repo = Repository('/tmp')
|
||||
>>> repo.snapshots
|
||||
[]
|
||||
@ -140,22 +147,14 @@ class Repository(Hierarchy):
|
||||
:return: a new Snapshot object
|
||||
"""
|
||||
syslog.debug("Creating snapshot")
|
||||
path = os.path.join(self._snapshot_dir, f"snapshot-{name}")
|
||||
path = self._snapshot_dir / f"snapshot-{name}"
|
||||
|
||||
self._curr_snapshot = Snapshot(path)
|
||||
self._snapshots.append(self._curr_snapshot)
|
||||
self.snapshots.append(self._curr_snapshot)
|
||||
|
||||
syslog.debug("Snapshot created")
|
||||
syslog.debug(f"Snapshot name: {self.curr_snapshot.name}")
|
||||
|
||||
def update_snapshots(self):
|
||||
"""Update the list of snapshots in this repository."""
|
||||
self._snapshots = [
|
||||
Snapshot(s)
|
||||
for s in glob.glob(f"{self._snapshot_dir}/*")
|
||||
if os.path.isdir(s)
|
||||
]
|
||||
|
||||
|
||||
# ========== Functions ==========
|
||||
if __name__ == "__main__":
|
||||
|
@ -4,7 +4,6 @@
|
||||
:synopsis: Classes for creating the /tmp hierarchy.
|
||||
"""
|
||||
import logging
|
||||
import os.path
|
||||
|
||||
from rbackup.hierarchy.hierarchy import Hierarchy
|
||||
|
||||
@ -19,7 +18,7 @@ class Snapshot(Hierarchy):
|
||||
Attributes
|
||||
----------
|
||||
* path (inherited from Hierarchy)
|
||||
* name
|
||||
* name (inherited from Hierarchy)
|
||||
* boot_dir
|
||||
* etc_dir
|
||||
* home_dir
|
||||
@ -30,24 +29,10 @@ class Snapshot(Hierarchy):
|
||||
"""Default constructor for the Snapshot class."""
|
||||
super().__init__(path)
|
||||
|
||||
self._boot_dir = os.path.join(self.path, "boot")
|
||||
self._etc_dir = os.path.join(self.path, "etc")
|
||||
self._home_dir = os.path.join(self.path, "home")
|
||||
self._root_home_dir = os.path.join(self.path, "root")
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
"""Return the name of this snapshot.
|
||||
|
||||
Example
|
||||
-------
|
||||
>>> s = Snapshot('/tmp/data/snapshot-new')
|
||||
>>> s.name
|
||||
'snapshot-new'
|
||||
|
||||
:rtype: str
|
||||
"""
|
||||
return os.path.basename(self.path)
|
||||
self._boot_dir = self.path / "boot"
|
||||
self._etc_dir = self.path / "etc"
|
||||
self._home_dir = self.path / "home"
|
||||
self._root_home_dir = self.path / "root"
|
||||
|
||||
@property
|
||||
def boot_dir(self):
|
||||
@ -57,9 +42,9 @@ class Snapshot(Hierarchy):
|
||||
-------
|
||||
>>> s = Snapshot('/tmp/data/snapshot-new')
|
||||
>>> s.boot_dir
|
||||
'/tmp/data/snapshot-new/boot'
|
||||
PosixPath('/tmp/data/snapshot-new/boot')
|
||||
|
||||
:rtype: str
|
||||
:rtype: path-like object
|
||||
"""
|
||||
return self._boot_dir
|
||||
|
||||
@ -71,9 +56,9 @@ class Snapshot(Hierarchy):
|
||||
-------
|
||||
>>> s = Snapshot('/tmp/data/snapshot-new')
|
||||
>>> s.etc_dir
|
||||
'/tmp/data/snapshot-new/etc'
|
||||
PosixPath('/tmp/data/snapshot-new/etc')
|
||||
|
||||
:rtype: str
|
||||
:rtype: path-like object
|
||||
"""
|
||||
return self._etc_dir
|
||||
|
||||
@ -85,9 +70,9 @@ class Snapshot(Hierarchy):
|
||||
-------
|
||||
>>> s = Snapshot('/tmp/data/snapshot-new')
|
||||
>>> s.home_dir
|
||||
'/tmp/data/snapshot-new/home'
|
||||
PosixPath('/tmp/data/snapshot-new/home')
|
||||
|
||||
:rtype: str
|
||||
:rtype: path-like object
|
||||
"""
|
||||
return self._home_dir
|
||||
|
||||
@ -99,9 +84,9 @@ class Snapshot(Hierarchy):
|
||||
-------
|
||||
>>> s = Snapshot('/tmp/data/snapshot-new')
|
||||
>>> s.root_home_dir
|
||||
'/tmp/data/snapshot-new/root'
|
||||
PosixPath('/tmp/data/snapshot-new/root')
|
||||
|
||||
:rtype: str
|
||||
:rtype: path-like object
|
||||
"""
|
||||
return self._root_home_dir
|
||||
|
||||
|
@ -1,15 +1,13 @@
|
||||
import doctest
|
||||
import unittest
|
||||
|
||||
from pathlib import PosixPath
|
||||
from rbackup.hierarchy.repository import Repository
|
||||
from rbackup.hierarchy.snapshot import Snapshot
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import patch, PropertyMock
|
||||
|
||||
# ========== Constants ==========
|
||||
TESTING_MODULE = "rbackup.hierarchy.repository"
|
||||
OS_PATH = f"{TESTING_MODULE}.os.path"
|
||||
OS_PATH_ISDIR = f"{OS_PATH}.isdir"
|
||||
GLOB_GLOB = f"{TESTING_MODULE}.glob.glob"
|
||||
|
||||
|
||||
# ========== Functions ==========
|
||||
@ -19,47 +17,67 @@ def load_tests(loader, tests, ignore):
|
||||
|
||||
|
||||
# ========== Integration Tests ==========
|
||||
class TestRepository(unittest.TestCase):
|
||||
class TestPopulatedRepository(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.patched_isdir = patch(OS_PATH_ISDIR)
|
||||
self.mocked_isdir = self.patched_isdir.start()
|
||||
|
||||
self.mocked_isdir.return_value = True
|
||||
|
||||
self.patched_glob = patch(GLOB_GLOB)
|
||||
self.mocked_glob = self.patched_glob.start()
|
||||
|
||||
self.snapshots = [
|
||||
"backup/data/snapshot-first",
|
||||
"backup/data/snapshot-second",
|
||||
"backup/data/snapshot-third",
|
||||
Snapshot("backup/data/snapshot-first"),
|
||||
Snapshot("backup/data/snapshot-second"),
|
||||
Snapshot("backup/data/snapshot-third"),
|
||||
]
|
||||
|
||||
self.mocked_glob.return_value = self.snapshots
|
||||
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.snapshots
|
||||
|
||||
self.repo_basepath = "backup"
|
||||
|
||||
self.repo = Repository(self.repo_basepath)
|
||||
|
||||
def test_snapshots(self):
|
||||
found_snapshots = [s.path for s in self.repo.snapshots]
|
||||
found_snapshots = [s for s in self.repo.snapshots]
|
||||
self.assertListEqual(found_snapshots, self.snapshots)
|
||||
|
||||
def test_curr_snapshot_pre_create(self):
|
||||
snapshot_name = "third"
|
||||
snapshot_path = f"backup/data/snapshot-{snapshot_name}"
|
||||
last_snapshot = Snapshot(f"backup/data/snapshot-{snapshot_name}")
|
||||
|
||||
self.assertEqual(self.repo.curr_snapshot.path, snapshot_path)
|
||||
self.assertEqual(self.repo.curr_snapshot.path, last_snapshot.path)
|
||||
self.assertIsInstance(self.repo.curr_snapshot, Snapshot)
|
||||
|
||||
def test_curr_snapshot_post_create(self):
|
||||
snapshot_name = "new"
|
||||
snapshot_path = f"backup/data/snapshot-{snapshot_name}"
|
||||
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.assertIsInstance(self.repo.curr_snapshot, Snapshot)
|
||||
|
||||
def tearDown(self):
|
||||
self.patched_isdir.stop()
|
||||
self.patched_glob.stop()
|
||||
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 tearDown(self):
|
||||
self.patched_snapshots.stop()
|
||||
|
@ -6,12 +6,13 @@ Unit tests for the Snapshot class.
|
||||
import doctest
|
||||
import unittest
|
||||
|
||||
from pathlib import Path
|
||||
from rbackup.hierarchy.snapshot import Snapshot
|
||||
from unittest.mock import patch
|
||||
|
||||
# ========== Constants ==========
|
||||
TESTING_MODULE = "rbackup.hierarchy.repository"
|
||||
|
||||
|
||||
# ========== Functions ==========
|
||||
def load_tests(loader, tests, ignore):
|
||||
tests.addTests(doctest.DocTestSuite(TESTING_MODULE))
|
||||
@ -21,33 +22,31 @@ def load_tests(loader, tests, ignore):
|
||||
# ========== Test Cases ==========
|
||||
class TestSnapshot(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.patched_isdir = patch("rbackup.hierarchy.snapshot.os.path.isdir")
|
||||
self.mocked_isdir = self.patched_isdir.start()
|
||||
self.snapshot_fullpath = Path("backup/data/snapshot-new")
|
||||
self.test_snapshot = Snapshot(self.snapshot_fullpath)
|
||||
|
||||
self.mocked_isdir.return_value = True
|
||||
|
||||
self.snapshot_fullpath = "backup/data/snapshot-new"
|
||||
self.snapshot_name = "snapshot-new"
|
||||
|
||||
self.snapshot = Snapshot(self.snapshot_fullpath)
|
||||
|
||||
def test_path(self):
|
||||
self.assertEqual(self.snapshot.path, self.snapshot_fullpath)
|
||||
def test_fullpath(self):
|
||||
self.assertEqual(self.test_snapshot.path, self.snapshot_fullpath)
|
||||
|
||||
def test_name(self):
|
||||
self.assertEqual(self.snapshot.name, self.snapshot_name)
|
||||
self.assertEqual(self.test_snapshot.name, "snapshot-new")
|
||||
|
||||
def test_boot_dir(self):
|
||||
self.assertEqual(self.snapshot.boot_dir, f"{self.snapshot_fullpath}/boot")
|
||||
self.assertEqual(
|
||||
self.test_snapshot.boot_dir, self.snapshot_fullpath / "boot"
|
||||
)
|
||||
|
||||
def test_etc_dir(self):
|
||||
self.assertEqual(self.snapshot.etc_dir, f"{self.snapshot_fullpath}/etc")
|
||||
self.assertEqual(
|
||||
self.test_snapshot.etc_dir, self.snapshot_fullpath / "etc"
|
||||
)
|
||||
|
||||
def test_home_dir(self):
|
||||
self.assertEqual(self.snapshot.home_dir, f"{self.snapshot_fullpath}/home")
|
||||
self.assertEqual(
|
||||
self.test_snapshot.home_dir, self.snapshot_fullpath / "home"
|
||||
)
|
||||
|
||||
def test_root_home_dir(self):
|
||||
self.assertEqual(self.snapshot.root_home_dir, f"{self.snapshot_fullpath}/root")
|
||||
|
||||
def tearDown(self):
|
||||
self.patched_isdir.stop()
|
||||
self.assertEqual(
|
||||
self.test_snapshot.root_home_dir, self.snapshot_fullpath / "root"
|
||||
)
|
||||
|
Loading…
x
Reference in New Issue
Block a user