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