reivilibre/yama
1
0
Fork 0
Code Issues 27 Pull Requests Releases 16 Wiki Activity

Add some checking and GC tests

Browse Source
This commit is contained in:
Olivier 'reivilibre' 2021-09-05 22:05:20 +01:00
parent cc85ec24a3
commit f86b37845a
1 changed files with 101 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
testsuite/yamatests/test_check_and_gc.py Normal file
View File

@ -0,0 +1,101 @@
import subprocess
from pathlib import Path
from random import Random
from tempfile import TemporaryDirectory
from unittest import TestCase
from helpers import generate_random_dir
from helpers.datman_helpers import set_up_simple_datman
from helpers.yama_helpers import set_up_simple_yama
class TestYamaCheck(TestCase):
def test_check_on_empty_pile_is_ok(self):
td = TemporaryDirectory("test_check_on_empty_pile_is_ok")
tdpath = Path(td.name)
datman_path = tdpath.joinpath("datman")
_src_path = datman_path.joinpath("srca")
yama_path = datman_path.joinpath("main")
set_up_simple_datman(datman_path)
set_up_simple_yama(yama_path)
subprocess.check_call(("yama", "check", "--shallow"), cwd=yama_path)
subprocess.check_call(("yama", "check", "--deep"), cwd=yama_path)
output = subprocess.check_output(("yama", "check", "--shallow", "--apply-gc"), cwd=yama_path, stderr=subprocess.STDOUT)
self.assertIn(b" 0 chunks", output)
output = subprocess.check_output(("yama", "check", "--deep", "--apply-gc"), cwd=yama_path, stderr=subprocess.STDOUT)
self.assertIn(b" 0 chunks", output)
td.cleanup()
def test_check_on_pile_with_store_is_ok(self):
td = TemporaryDirectory("test_check_on_pile_with_store_is_ok")
tdpath = Path(td.name)
datman_path = tdpath.joinpath("datman")
src_path = datman_path.joinpath("srca")
yama_path = datman_path.joinpath("main")
set_up_simple_datman(datman_path)
set_up_simple_yama(yama_path)
rng = Random()
seed = rng.randint(0, 9001)
print(f"seed: {seed}")
rng.seed(seed)
later_expected_descriptor, _ = generate_random_dir(rng, src_path, 32)
subprocess.check_call(("datman", "backup-one", "srca", "main"), cwd=datman_path)
subprocess.check_call(("yama", "check", "--shallow"), cwd=yama_path)
subprocess.check_call(("yama", "check", "--deep"), cwd=yama_path)
output = subprocess.check_output(("yama", "check", "--shallow", "--apply-gc"), cwd=yama_path,
stderr=subprocess.STDOUT)
self.assertIn(b" 0 chunks", output)
output = subprocess.check_output(("yama", "check", "--deep", "--apply-gc"), cwd=yama_path,
stderr=subprocess.STDOUT)
self.assertIn(b" 0 chunks", output)
td.cleanup()
def test_check_fails_after_random_corruption(self):
td = TemporaryDirectory("test_check_fails_after_random_corruption")
tdpath = Path(td.name)
datman_path = tdpath.joinpath("datman")
src_path = datman_path.joinpath("srca")
yama_path = datman_path.joinpath("main")
set_up_simple_datman(datman_path)
set_up_simple_yama(yama_path)
rng = Random()
seed = rng.randint(0, 9001)
print(f"seed: {seed}")
rng.seed(seed)
later_expected_descriptor, _ = generate_random_dir(rng, src_path, 32)
subprocess.check_call(("datman", "backup-one", "srca", "main"), cwd=datman_path)
victim_bloblog = yama_path.joinpath("bloblog", "0")
size_in_bytes = victim_bloblog.stat().st_size
with victim_bloblog.open("r+b") as fvictim:
byte_to_eat = rng.randint(0, size_in_bytes - 1)
fvictim.seek(byte_to_eat)
existing_byte = fvictim.read(1)
fvictim.seek(byte_to_eat)
if existing_byte == b"\0":
fvictim.write(b"\x01")
else:
fvictim.write(b"\0")
print(f"Corrupted byte {byte_to_eat} of {size_in_bytes}. Was {existing_byte!r}.")
ec_shallow = subprocess.Popen(("yama", "check", "--shallow"), cwd=yama_path).wait()
ec_deep = subprocess.Popen(("yama", "check", "--deep"), cwd=yama_path).wait()
# shallow checks won't always raise the issue
self.assertIn(ec_shallow, (0, 1))
# deep checks should always raise the issue
self.assertEqual(ec_deep, 1)
td.cleanup()