1import os.path
2import tempfile
3import shutil
4import unittest
5import librepo
6
7from tests.base import Context, TestCase, TEST_DATA
8
9# Repositories used in download tests
10#REPOS_YUM = "tests/python/tests/servermock/yum_mock/static/"
11
12REPO_YUM_01_PATH = TEST_DATA+"/repo_yum_01/"
13REPO_YUM_02_PATH = TEST_DATA+"/repo_yum_02/"
14PUB_KEY = TEST_DATA+"/key.pub"
15MIRRORLIST = TEST_DATA+"/mirrorlist"
16METALINK = TEST_DATA+"/metalink.xml"
17
18class TestCaseYumRepoLocating(TestCase):
19
20    def setUp(self):
21        self.tmpdir = tempfile.mkdtemp(prefix="librepotest-")
22        # Import public key into the temporary gpg keyring
23        self._gnupghome = os.environ.get('GNUPGHOME')
24        gpghome = os.path.join(self.tmpdir, "keyring")
25        os.mkdir(gpghome, 0o700)
26        os.environ['GNUPGHOME'] = gpghome
27        self.ctx = Context()
28        self.ctx.op_import(open(PUB_KEY, 'rb'))
29
30    def tearDown(self):
31        if self._gnupghome is None:
32            os.environ.pop('GNUPGHOME')
33        else:
34            os.environ['GNUPGHOME'] = self._gnupghome
35        shutil.rmtree(self.tmpdir, True)
36
37    def test_read_mirrorlist(self):
38        h = librepo.Handle()
39        r = librepo.Result()
40        h.mirrorlist = MIRRORLIST
41        h.repotype = librepo.LR_YUMREPO
42        h.destdir = self.tmpdir
43        h.fetchmirrors = True
44        h.perform(r)
45        self.assertEqual(h.mirrors, ['http://127.0.0.1:5000/yum/static/01/'])
46        self.assertEqual(h.metalink, None)
47
48    def test_read_metalink(self):
49        h = librepo.Handle()
50        r = librepo.Result()
51        h.mirrorlist = METALINK
52        h.repotype = librepo.LR_YUMREPO
53        h.destdir = self.tmpdir
54        h.fetchmirrors = True
55        h.perform(r)
56        self.assertEqual(h.mirrors, ['http://127.0.0.1:5000/yum/static/01/'])
57        self.assertEqual(h.metalink,
58            {'timestamp': 1347459931,
59             'hashes': [
60                 ('md5', 'f76409f67a84bcd516131d5cc98e57e1'),
61                 ('sha1', '75125e73304c21945257d9041a908d0d01d2ca16'),
62                 ('sha256', 'bef5d33dc68f47adc7b31df448851b1e9e6bae27840f28700fff144881482a6a'),
63                 ('sha512', 'e40060c747895562e945a68967a04d1279e4bd8507413681f83c322479aa564027fdf3962c2d875089bfcb9317d3a623465f390dc1f4acef294711168b807af0')],
64             'size': 2621,
65             'urls': [{
66                 'url': 'http://127.0.0.1:5000/yum/static/01/repodata/repomd.xml',
67                 'type': 'http',
68                 'protocol': 'http',
69                 'location': 'CZ',
70                 'preference': 100}],
71             'filename': 'repomd.xml'})
72
73    def test_read_metalink_of_local_repo(self):
74        # At first, download whole repository
75        # Check if local metalink.xml will be copied as well
76        h = librepo.Handle()
77        r = librepo.Result()
78
79        h.urls = [REPO_YUM_01_PATH]
80        h.repotype = librepo.LR_YUMREPO
81        h.destdir = self.tmpdir
82        h.gpgcheck = True
83        h.perform(r)
84
85        yum_repo_downloaded   = r.getinfo(librepo.LRR_YUM_REPO)
86        yum_repomd_downloaded = r.getinfo(librepo.LRR_YUM_REPOMD)
87
88        self.assertFalse(yum_repo_downloaded["mirrorlist"])
89        self.assertTrue(yum_repo_downloaded["metalink"])
90        self.assertTrue(yum_repo_downloaded["metalink"].endswith("metalink.xml"))
91
92        # Now try to read metalink of the repository
93        h = librepo.Handle()
94        r = librepo.Result()
95
96        h.urls = [self.tmpdir]
97        h.repotype = librepo.LR_YUMREPO
98        h.local = True
99        h.fetchmirrors = True
100        h.perform(r)
101
102        self.assertEqual(h.mirrors, ['http://127.0.0.1:5000/yum/static/01/'])
103        self.assertEqual(h.metalink,
104            {'timestamp': 1347459931,
105             'hashes': [
106                 ('md5', 'f76409f67a84bcd516131d5cc98e57e1'),
107                 ('sha1', '75125e73304c21945257d9041a908d0d01d2ca16'),
108                 ('sha256', 'bef5d33dc68f47adc7b31df448851b1e9e6bae27840f28700fff144881482a6a'),
109                 ('sha512', 'e40060c747895562e945a68967a04d1279e4bd8507413681f83c322479aa564027fdf3962c2d875089bfcb9317d3a623465f390dc1f4acef294711168b807af0')],
110             'size': 2621,
111             'urls': [{
112                 'url': 'http://127.0.0.1:5000/yum/static/01/repodata/repomd.xml',
113                 'type': 'http',
114                 'protocol': 'http',
115                 'location': 'CZ',
116                 'preference': 100}],
117             'filename': 'repomd.xml'}
118            )
119
120    def test_locate_repo_01(self):
121        # At first, download whole repository
122        h = librepo.Handle()
123        r = librepo.Result()
124
125        h.urls = [REPO_YUM_01_PATH]
126        h.repotype = librepo.LR_YUMREPO
127        h.destdir = self.tmpdir
128        h.gpgcheck = True
129        h.perform(r)
130
131        yum_repo_downloaded   = r.getinfo(librepo.LRR_YUM_REPO)
132        yum_repomd_downloaded = r.getinfo(librepo.LRR_YUM_REPOMD)
133
134        # Now try to localize the existing repository and all its files
135        h = librepo.Handle()
136        r = librepo.Result()
137
138        h.urls = [self.tmpdir]
139        h.repotype = librepo.LR_YUMREPO
140        h.local = True
141        h.perform(r)
142
143        yum_repo   = r.getinfo(librepo.LRR_YUM_REPO)
144        yum_repomd = r.getinfo(librepo.LRR_YUM_REPOMD)
145
146        # Compare results
147        yum_repo_downloaded["url"] = None
148        self.assertEqual(yum_repo, yum_repo_downloaded)
149        self.assertEqual(yum_repomd, yum_repomd_downloaded)
150
151    def test_locate_repo_02(self):
152        # At first, download whole repository
153        h = librepo.Handle()
154        r = librepo.Result()
155
156        h.urls = [REPO_YUM_02_PATH]
157        h.repotype = librepo.LR_YUMREPO
158        h.destdir = self.tmpdir
159        h.gpgcheck = False
160        h.perform(r)
161
162        yum_repo_downloaded   = r.getinfo(librepo.LRR_YUM_REPO)
163        yum_repomd_downloaded = r.getinfo(librepo.LRR_YUM_REPOMD)
164
165        # Now try to localize the existing repository and all its files
166        h = librepo.Handle()
167        r = librepo.Result()
168
169        h.urls = [self.tmpdir]
170        h.repotype = librepo.LR_YUMREPO
171        h.local = True
172        h.perform(r)
173
174        yum_repo   = r.getinfo(librepo.LRR_YUM_REPO)
175        yum_repomd = r.getinfo(librepo.LRR_YUM_REPOMD)
176
177        # Compare results
178        yum_repo_downloaded["url"] = None
179        self.assertEqual(yum_repo, yum_repo_downloaded)
180        self.assertEqual(yum_repomd, yum_repomd_downloaded)
181
182    def test_locate_incomplete_repo_01(self):
183        # At first, download only some files from the repository
184        h = librepo.Handle()
185        r = librepo.Result()
186
187        h.urls = [REPO_YUM_01_PATH]
188        h.repotype = librepo.LR_YUMREPO
189        h.destdir = self.tmpdir
190        h.gpgcheck = True
191        h.yumdlist = ["primary"]
192        h.perform(r)
193
194        yum_repo_downloaded   = r.getinfo(librepo.LRR_YUM_REPO)
195        yum_repomd_downloaded = r.getinfo(librepo.LRR_YUM_REPOMD)
196
197        # Now try to localize the existing repository
198        h = librepo.Handle()
199        r = librepo.Result()
200
201        h.urls = [self.tmpdir]
202        h.repotype = librepo.LR_YUMREPO
203        h.local = True
204        self.assertRaises(librepo.LibrepoException, h.perform, (r))
205
206    def test_locate_incomplete_repo_01_2(self):
207        # At first, download whole repository
208        h = librepo.Handle()
209        r = librepo.Result()
210
211        h.urls = [REPO_YUM_01_PATH]
212        h.repotype = librepo.LR_YUMREPO
213        h.destdir = self.tmpdir
214        h.gpgcheck = True
215        h.yumdlist = ["primary"]
216        h.perform(r)
217
218        yum_repo_downloaded   = r.getinfo(librepo.LRR_YUM_REPO)
219        yum_repomd_downloaded = r.getinfo(librepo.LRR_YUM_REPOMD)
220
221        # Now try to localize the existing repository and all its files
222        h = librepo.Handle()
223        r = librepo.Result()
224
225        h.urls = [self.tmpdir]
226        h.repotype = librepo.LR_YUMREPO
227        h.local = True
228        h.ignoremissing = True
229        h.perform(r)
230
231        yum_repo   = r.getinfo(librepo.LRR_YUM_REPO)
232        yum_repomd = r.getinfo(librepo.LRR_YUM_REPOMD)
233
234        # Compare results
235        yum_repo_downloaded["url"] = None
236        self.assertEqual(yum_repo, yum_repo_downloaded)
237        self.assertEqual(yum_repomd, yum_repomd_downloaded)
238
239    def test_locate_with_gpgcheck_enabled_but_without_signature(self):
240        # At first, download whole repository
241        h = librepo.Handle()
242        h.urls = [REPO_YUM_02_PATH]
243        h.repotype = librepo.LR_YUMREPO
244        h.gpgcheck = True
245        h.local = True
246        self.assertRaises(librepo.LibrepoException, h.perform)
247
248    def test_locate_with_metalink_and_mirrorlist_urls_set(self):
249        # See: https://github.com/Tojaj/librepo/issues/41
250        h = librepo.Handle()
251        h.urls = [REPO_YUM_02_PATH]
252        # These bad URLs should not raise any error because
253        # librepo shoudn't try to download them
254        h.metalinkurl = "xyz://really-bad-url-sf987243kj.com"
255        h.mirrorlisturl = "xyz://really-bad-url-mi-qcwmi.com"
256        h.repotype = librepo.LR_YUMREPO
257        h.local = True
258        h.perform()
259
260    def test_locate_with_offline(self):
261        # Offline and Local options shouldn't collide in any way
262        h = librepo.Handle()
263        h.urls = [REPO_YUM_02_PATH]
264        h.metalinkurl = "xyz://really-bad-url-sf987243kj.com"
265        h.mirrorlisturl = "xyz://really-bad-url-mi-qcwmi.com"
266        h.repotype = librepo.LR_YUMREPO
267        h.local = True
268        h.offline = True
269        h.perform()
270