1import os 2import tarfile 3 4import pytest 5 6from ..constants import * # NOQA 7from ..crypto.key import KeyfileKey 8from ..upgrader import AtticRepositoryUpgrader, AtticKeyfileKey 9from ..helpers import get_keys_dir 10from ..repository import Repository 11from . import are_hardlinks_supported 12 13 14# tar with a repo and repo keyfile from attic 15ATTIC_TAR = os.path.join(os.path.dirname(__file__), 'attic.tar.gz') 16 17 18def untar(tarfname, path, what): 19 """ 20 extract <tarfname> tar archive to <path>, all stuff starting with <what>. 21 22 return path to <what>. 23 """ 24 25 def files(members): 26 for tarinfo in members: 27 if tarinfo.name.startswith(what): 28 yield tarinfo 29 30 with tarfile.open(tarfname, 'r') as tf: 31 tf.extractall(path, members=files(tf)) 32 33 return os.path.join(path, what) 34 35 36def repo_valid(path): 37 """ 38 utility function to check if borg can open a repository 39 40 :param path: the path to the repository 41 :returns: if borg can check the repository 42 """ 43 with Repository(str(path), exclusive=True, create=False) as repository: 44 # can't check raises() because check() handles the error 45 return repository.check() 46 47 48def key_valid(path): 49 """ 50 check that the new keyfile is alright 51 52 :param path: the path to the key file 53 :returns: if the file starts with the borg magic string 54 """ 55 keyfile = os.path.join(get_keys_dir(), 56 os.path.basename(path)) 57 with open(keyfile, 'r') as f: 58 return f.read().startswith(KeyfileKey.FILE_ID) 59 60 61def make_attic_repo(dir): 62 """ 63 create an attic repo with some stuff in it 64 65 :param dir: path to the repository to be created 66 :returns: path to attic repository 67 """ 68 # there is some stuff in that repo, copied from `RepositoryTestCase.test1` 69 return untar(ATTIC_TAR, str(dir), 'repo') 70 71 72@pytest.fixture() 73def attic_repo(tmpdir): 74 return make_attic_repo(tmpdir) 75 76 77@pytest.fixture(params=[True, False]) 78def inplace(request): 79 return request.param 80 81 82def test_convert_segments(attic_repo, inplace): 83 """test segment conversion 84 85 this will load the given attic repository, list all the segments 86 then convert them one at a time. we need to close the repo before 87 conversion otherwise we have errors from borg 88 89 :param attic_repo: a populated attic repository (fixture) 90 """ 91 repo_path = attic_repo 92 with pytest.raises(Repository.AtticRepository): 93 repo_valid(repo_path) 94 repository = AtticRepositoryUpgrader(repo_path, create=False) 95 with repository: 96 segments = [filename for i, filename in repository.io.segment_iterator()] 97 repository.convert_segments(segments, dryrun=False, inplace=inplace) 98 repository.convert_cache(dryrun=False) 99 assert repo_valid(repo_path) 100 101 102@pytest.fixture() 103def attic_key_file(tmpdir, monkeypatch): 104 """ 105 create an attic key file from the given repo, in the keys 106 subdirectory of the given tmpdir 107 108 :param tmpdir: a temporary directory (a builtin fixture) 109 :returns: path to key file 110 """ 111 keys_dir = untar(ATTIC_TAR, str(tmpdir), 'keys') 112 113 # we use the repo dir for the created keyfile, because we do 114 # not want to clutter existing keyfiles 115 monkeypatch.setenv('ATTIC_KEYS_DIR', keys_dir) 116 117 # we use the same directory for the converted files, which 118 # will clutter the previously created one, which we don't care 119 # about anyways. in real runs, the original key will be retained. 120 monkeypatch.setenv('BORG_KEYS_DIR', keys_dir) 121 monkeypatch.setenv('ATTIC_PASSPHRASE', 'test') 122 123 return os.path.join(keys_dir, 'repo') 124 125 126def test_keys(attic_repo, attic_key_file): 127 """test key conversion 128 129 test that we can convert the given key to a properly formatted 130 borg key. assumes that the ATTIC_KEYS_DIR and BORG_KEYS_DIR have 131 been properly populated by the attic_key_file fixture. 132 133 :param attic_repo: path to an attic repository (fixture defined above) 134 :param attic_key_file: path to an attic key file (fixture defined above) 135 """ 136 keyfile_path = attic_key_file 137 assert not key_valid(keyfile_path) # not upgraded yet 138 with AtticRepositoryUpgrader(attic_repo, create=False) as repository: 139 keyfile = AtticKeyfileKey.find_key_file(repository) 140 AtticRepositoryUpgrader.convert_keyfiles(keyfile, dryrun=False) 141 assert key_valid(keyfile_path) 142 143 144@pytest.mark.skipif(not are_hardlinks_supported(), reason='hardlinks not supported') 145def test_convert_all(attic_repo, attic_key_file, inplace): 146 """test all conversion steps 147 148 this runs everything. mostly redundant test, since everything is 149 done above. yet we expect a NotImplementedError because we do not 150 convert caches yet. 151 152 :param attic_repo: path to an attic repository (fixture defined above) 153 :param attic_key_file: path to an attic key file (fixture defined above) 154 """ 155 repo_path = attic_repo 156 157 with pytest.raises(Repository.AtticRepository): 158 repo_valid(repo_path) 159 160 def stat_segment(path): 161 return os.stat(os.path.join(path, 'data', '0', '0')) 162 163 def first_inode(path): 164 return stat_segment(path).st_ino 165 166 orig_inode = first_inode(repo_path) 167 with AtticRepositoryUpgrader(repo_path, create=False) as repository: 168 # replicate command dispatch, partly 169 os.umask(UMASK_DEFAULT) 170 backup = repository.upgrade(dryrun=False, inplace=inplace) # note: uses hardlinks internally 171 if inplace: 172 assert backup is None 173 assert first_inode(repository.path) == orig_inode 174 else: 175 assert backup 176 assert first_inode(repository.path) != first_inode(backup) 177 # i have seen cases where the copied tree has world-readable 178 # permissions, which is wrong 179 if 'BORG_TESTS_IGNORE_MODES' not in os.environ: 180 assert stat_segment(backup).st_mode & UMASK_DEFAULT == 0 181 182 assert key_valid(attic_key_file) 183 assert repo_valid(repo_path) 184 185 186@pytest.mark.skipif(not are_hardlinks_supported(), reason='hardlinks not supported') 187def test_hardlink(tmpdir, inplace): 188 """test that we handle hard links properly 189 190 that is, if we are in "inplace" mode, hardlinks should *not* 191 change (ie. we write to the file directly, so we do not rewrite the 192 whole file, and we do not re-create the file). 193 194 if we are *not* in inplace mode, then the inode should change, as 195 we are supposed to leave the original inode alone.""" 196 a = str(tmpdir.join('a')) 197 with open(a, 'wb') as tmp: 198 tmp.write(b'aXXX') 199 b = str(tmpdir.join('b')) 200 os.link(a, b) 201 AtticRepositoryUpgrader.header_replace(b, b'a', b'b', inplace=inplace) 202 if not inplace: 203 assert os.stat(a).st_ino != os.stat(b).st_ino 204 else: 205 assert os.stat(a).st_ino == os.stat(b).st_ino 206 with open(b, 'rb') as tmp: 207 assert tmp.read() == b'bXXX' 208