Use regex to parse user snapshot name input
This commit is contained in:
parent
51bc28e62b
commit
8a407292c9
@ -6,6 +6,8 @@
|
|||||||
"""
|
"""
|
||||||
import datetime
|
import datetime
|
||||||
import logging
|
import logging
|
||||||
|
import re
|
||||||
|
import shutil
|
||||||
|
|
||||||
from rbackup.struct.hierarchy import Hierarchy
|
from rbackup.struct.hierarchy import Hierarchy
|
||||||
from rbackup.struct.snapshot import Snapshot
|
from rbackup.struct.snapshot import Snapshot
|
||||||
@ -18,6 +20,8 @@ syslog = logging.getLogger(__name__)
|
|||||||
DIRMODE = 0o755
|
DIRMODE = 0o755
|
||||||
FILEMODE = 0o644
|
FILEMODE = 0o644
|
||||||
|
|
||||||
|
VALID_SNAPSHOT_NAME = r"[\w._+-]+[^/]*"
|
||||||
|
|
||||||
|
|
||||||
# ========== Classes ==========
|
# ========== Classes ==========
|
||||||
class Repository(Hierarchy):
|
class Repository(Hierarchy):
|
||||||
@ -121,23 +125,22 @@ class Repository(Hierarchy):
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def is_valid_snapshot_name(name):
|
def is_valid_snapshot_name(name):
|
||||||
"""Check if the given name is a valid name. If it is a duplicate,
|
"""Check if the given name is a valid name.
|
||||||
log a warning. If it is invalid, raise a ValueError.
|
|
||||||
|
|
||||||
Invalid Names:
|
Invalid Names:
|
||||||
--------------
|
--------------
|
||||||
* Contain slashes
|
* Contain slashes
|
||||||
* Are empty values i.e. '' or []
|
* Are empty values
|
||||||
|
|
||||||
|
Valid names match the regex
|
||||||
|
r'[\w]+[^/]*'
|
||||||
|
|
||||||
:param name: name to validate
|
:param name: name to validate
|
||||||
:type name: str
|
:type name: str
|
||||||
:returns: true if this name is deemed valid
|
:returns: true if this name is deemed valid
|
||||||
:rtype: bool
|
:rtype: bool
|
||||||
"""
|
"""
|
||||||
if not str(name) or "/" in name:
|
return bool(re.match(VALID_SNAPSHOT_NAME, name))
|
||||||
return False
|
|
||||||
else:
|
|
||||||
return True
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def snapshot_dir(self):
|
def snapshot_dir(self):
|
||||||
|
@ -3,14 +3,12 @@
|
|||||||
|
|
||||||
Tests for the rbackup.struct.repository module.
|
Tests for the rbackup.struct.repository module.
|
||||||
"""
|
"""
|
||||||
# TODO test that the snapshot returned is actually in the repository
|
import re
|
||||||
# TODO test creating snapshots, returned snapshot is an instance of Snapshot, etc.
|
|
||||||
|
|
||||||
import unittest
|
import unittest
|
||||||
from unittest.mock import PropertyMock, patch
|
from unittest.mock import PropertyMock, patch
|
||||||
|
|
||||||
from hypothesis import given
|
from hypothesis import given
|
||||||
from hypothesis.strategies import characters, lists, text
|
from hypothesis.strategies import from_regex, lists, text
|
||||||
|
|
||||||
from rbackup.struct.repository import Repository
|
from rbackup.struct.repository import Repository
|
||||||
from rbackup.struct.snapshot import Snapshot
|
from rbackup.struct.snapshot import Snapshot
|
||||||
@ -20,7 +18,7 @@ TESTING_PACKAGE = "rbackup.struct"
|
|||||||
REPO_MODULE = f"{TESTING_PACKAGE}.repository"
|
REPO_MODULE = f"{TESTING_PACKAGE}.repository"
|
||||||
SS_MODULE = f"{TESTING_PACKAGE}.snapshot"
|
SS_MODULE = f"{TESTING_PACKAGE}.snapshot"
|
||||||
|
|
||||||
UNWANTED_SNAPSHOT_CHARS = ["/"]
|
VALID_SNAPSHOT_NAME = r"[\w._+-]+[^/]*"
|
||||||
|
|
||||||
|
|
||||||
# ========== Integration Tests ==========
|
# ========== Integration Tests ==========
|
||||||
@ -59,15 +57,7 @@ class TestRepositoryPreCreate(unittest.TestCase):
|
|||||||
|
|
||||||
self.mocked_path.return_value.exists.return_value = True
|
self.mocked_path.return_value.exists.return_value = True
|
||||||
|
|
||||||
@given(
|
@given(lists(from_regex(VALID_SNAPSHOT_NAME, fullmatch=True), unique=True))
|
||||||
lists(
|
|
||||||
text(
|
|
||||||
alphabet=characters(blacklist_characters=UNWANTED_SNAPSHOT_CHARS),
|
|
||||||
min_size=1,
|
|
||||||
),
|
|
||||||
unique=True,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
def test_empty(self, snapshots):
|
def test_empty(self, snapshots):
|
||||||
self.mocked_r_metadata.return_value = snapshots.copy()
|
self.mocked_r_metadata.return_value = snapshots.copy()
|
||||||
repo = Repository("backup")
|
repo = Repository("backup")
|
||||||
@ -103,7 +93,7 @@ class TestRepositoryPreCreate(unittest.TestCase):
|
|||||||
def test_valid_name(self, name):
|
def test_valid_name(self, name):
|
||||||
self.mocked_r_metadata.return_value = []
|
self.mocked_r_metadata.return_value = []
|
||||||
|
|
||||||
if not name or "/" in name:
|
if not re.match(VALID_SNAPSHOT_NAME, name):
|
||||||
self.assertFalse(Repository.is_valid_snapshot_name(name))
|
self.assertFalse(Repository.is_valid_snapshot_name(name))
|
||||||
else:
|
else:
|
||||||
self.assertTrue(Repository.is_valid_snapshot_name(name))
|
self.assertTrue(Repository.is_valid_snapshot_name(name))
|
||||||
@ -152,16 +142,8 @@ class TestRepositoryPostCreate(unittest.TestCase):
|
|||||||
self.mocked_w_metadata = self.patched_w_metadata.start()
|
self.mocked_w_metadata = self.patched_w_metadata.start()
|
||||||
self.mocked_snapshot = self.patched_snapshot.start()
|
self.mocked_snapshot = self.patched_snapshot.start()
|
||||||
|
|
||||||
@given(
|
@given(lists(from_regex(VALID_SNAPSHOT_NAME, fullmatch=True), unique=True))
|
||||||
lists(
|
def test_dunder_len(self, snapshots):
|
||||||
text(
|
|
||||||
alphabet=characters(blacklist_characters=UNWANTED_SNAPSHOT_CHARS),
|
|
||||||
min_size=1,
|
|
||||||
),
|
|
||||||
unique=True,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
def test_empty(self, snapshots):
|
|
||||||
self.mocked_r_metadata.return_value = snapshots.copy()
|
self.mocked_r_metadata.return_value = snapshots.copy()
|
||||||
repo = Repository("backup")
|
repo = Repository("backup")
|
||||||
|
|
||||||
@ -169,17 +151,16 @@ class TestRepositoryPostCreate(unittest.TestCase):
|
|||||||
|
|
||||||
self.assertFalse(repo.empty)
|
self.assertFalse(repo.empty)
|
||||||
|
|
||||||
@given(
|
@given(from_regex(VALID_SNAPSHOT_NAME, fullmatch=True))
|
||||||
lists(
|
def test_dunder_contains(self, name):
|
||||||
text(
|
self.mocked_path.return_value.exists.return_value = False
|
||||||
alphabet=characters(blacklist_characters=UNWANTED_SNAPSHOT_CHARS),
|
repo = Repository("backup")
|
||||||
min_size=1,
|
|
||||||
),
|
repo.create_snapshot(name)
|
||||||
unique=True,
|
self.assertTrue(name in repo)
|
||||||
)
|
|
||||||
)
|
def test_empty(self):
|
||||||
def test_len(self, snapshots):
|
self.mocked_r_metadata.return_value = []
|
||||||
self.mocked_r_metadata.return_value = snapshots.copy()
|
|
||||||
repo = Repository("backup")
|
repo = Repository("backup")
|
||||||
|
|
||||||
repo.create_snapshot()
|
repo.create_snapshot()
|
||||||
|
Loading…
x
Reference in New Issue
Block a user