-
Notifications
You must be signed in to change notification settings - Fork 189
/
test_util.py
1588 lines (1336 loc) · 64.3 KB
/
test_util.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Copyright © 2007-2019 Red Hat, Inc. and others.
#
# This file is part of Bodhi.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
from datetime import datetime, timedelta
from unittest import mock
from xml.etree import ElementTree
import gzip
import os
import shutil
import subprocess
import tempfile
from webob.multidict import MultiDict
import bleach
import createrepo_c
import packaging
import pytest
from bodhi.server import models, util
from bodhi.server.config import config
from bodhi.server.exceptions import RepodataException
from bodhi.server.models import ReleaseState, TestGatingStatus, Update
from . import base
class TestAvatar:
"""Test the avatar() function."""
def test_libravatar_disabled(self):
"""If libravatar_enabled is False, libravatar.org should be returned."""
config['libravatar_enabled'] = False
context = {'request': mock.MagicMock()}
def cache_on_arguments():
"""A fake cache - we aren't testing this so let's just return f."""
return lambda x: x
context['request'].cache.cache_on_arguments = cache_on_arguments
assert util.avatar(context, 'bowlofeggs', 50) == 'libravatar.org'
@mock.patch('bodhi.server.util.libravatar.libravatar_url', return_value='cool url')
def test_libravatar_dns_set_ssl_false(self, libravatar_url):
"""Test the correct return value when libravatar_dns is set in config."""
config.update({
'libravatar_enabled': True,
'libravatar_dns': True,
'libravatar_prefer_tls': False,
})
context = {'request': mock.MagicMock()}
context['request'].registry.settings = config
def cache_on_arguments():
"""A fake cache - we aren't testing this so let's just return f."""
return lambda x: x
context['request'].cache.cache_on_arguments = cache_on_arguments
assert util.avatar(context, 'bowlofeggs', 50) == 'cool url'
openid_user_host = config['openid_template'].format(username='bowlofeggs')
libravatar_url.assert_called_once_with(openid=f'http://{openid_user_host}/',
https=False, size=50, default='retro')
@mock.patch('bodhi.server.util.libravatar.libravatar_url', return_value='cool url')
def test_libravatar_dns_set_ssl_true(self, libravatar_url):
"""Test the correct return value when libravatar_dns is set in config."""
config.update({
'libravatar_enabled': True,
'libravatar_dns': True,
'libravatar_prefer_tls': True,
})
context = {'request': mock.MagicMock()}
context['request'].registry.settings = config
def cache_on_arguments():
"""A fake cache - we aren't testing this so let's just return f."""
return lambda x: x
context['request'].cache.cache_on_arguments = cache_on_arguments
assert util.avatar(context, 'bowlofeggs', 50) == 'cool url'
openid_user_host = config['openid_template'].format(username='bowlofeggs')
libravatar_url.assert_called_once_with(openid=f'http://{openid_user_host}/',
https=True, size=50, default='retro')
class TestBugLink:
"""Tests for the bug_link() function."""
def test_short_false_with_title(self):
"""Test a call to bug_link() with short=False on a Bug that has a title."""
bug = mock.MagicMock()
bug.bug_id = 1234567
bug.title = "Lucky bug number"
link = util.bug_link(None, bug)
assert link == \
("<a target='_blank' href='https://bugzilla.redhat.com/show_bug.cgi?id=1234567' "
"class='notblue'>BZ#1234567</a> Lucky bug number")
def test_short_false_with_title_sanitizes_safe_tags(self):
"""
Test that a call to bug_link() with short=False on a Bug that has a title sanitizes even
safe tags because really they should be rendered human readable.
"""
bug = mock.MagicMock()
bug.bug_id = 1234567
bug.title = 'Check <b>this</b> out'
link = util.bug_link(None, bug)
assert link == \
("<a target='_blank' href='https://bugzilla.redhat.com/show_bug.cgi?id=1234567' "
"class='notblue'>BZ#1234567</a> Check <b>this</b> out")
def test_short_false_with_title_sanitizes_unsafe_tags(self):
"""
Test that a call to bug_link() with short=False on a Bug that has a title sanitizes unsafe
tags.
"""
bug = mock.MagicMock()
bug.bug_id = 1473091
bug.title = '<disk> <driver name="..."> should be optional'
link = util.bug_link(None, bug)
# bleach v3 fixed a bug that closed out tags when sanitizing. so we check for
# either possible results here.
# https://github.com/mozilla/bleach/issues/392
if packaging.version.parse(bleach.__version__) >= packaging.version.parse('3.0.0'):
assert link == \
("<a target='_blank' href='https://bugzilla.redhat.com/show_bug.cgi?id=1473091' "
"class='notblue'>BZ#1473091</a> <disk> <driver name=\"...\"> should "
"be optional")
else:
assert link == \
("<a target='_blank' href='https://bugzilla.redhat.com/show_bug.cgi?id=1473091' "
"class='notblue'>BZ#1473091</a> <disk> <driver name=\"...\"> should "
"be optional</driver></disk>")
def test_short_false_without_title(self):
"""Test a call to bug_link() with short=False on a Bug that has no title."""
bug = mock.MagicMock()
bug.bug_id = 1234567
bug.title = None
link = util.bug_link(None, bug)
assert link == \
("<a target='_blank' href='https://bugzilla.redhat.com/show_bug.cgi?id=1234567' "
"class='notblue'>BZ#1234567</a> <i class='fa fa-spinner fa-spin fa-fw'></i>")
def test_short_true(self):
"""Test a call to bug_link() with short=True."""
bug = mock.MagicMock()
bug.bug_id = 1234567
bug.title = "Lucky bug number"
link = util.bug_link(None, bug, True)
assert link == \
("<a target='_blank' href='https://bugzilla.redhat.com/show_bug.cgi?id=1234567' "
"class='notblue'>BZ#1234567</a>")
@mock.patch('bodhi.server.util.time.sleep')
class TestCallAPI:
"""Test the call_api() function."""
@mock.patch('bodhi.server.util.http_session.get')
def test_retries_failure(self, get, sleep):
"""Assert correct operation of the retries argument when they never succeed."""
class FakeResponse(object):
def __init__(self, status_code):
self.status_code = status_code
@property
def text(self):
return "Some stuff"
def json(self):
return {'some': 'stuff'}
get.side_effect = [FakeResponse(503), FakeResponse(503)]
with pytest.raises(RuntimeError) as exc:
util.call_api('url', 'service_name', retries=1)
assert str(exc.value) == \
('Bodhi failed to get a resource from '
'service_name at the following URL "url". The '
'status code was "503". The error was "{\'some\': \'stuff\'}".')
assert get.mock_calls == [mock.call('url', timeout=60), mock.call('url', timeout=60)]
sleep.assert_called_once_with(1)
@mock.patch('bodhi.server.util.http_session.get')
def test_retries_success(self, get, sleep):
"""Assert correct operation of the retries argument when they succeed eventually."""
class FakeResponse(object):
def __init__(self, status_code):
self.status_code = status_code
def json(self):
return {'some': 'stuff'}
get.side_effect = [FakeResponse(503), FakeResponse(200)]
res = util.call_api('url', 'service_name', retries=1)
assert res == {'some': 'stuff'}
assert get.mock_calls == [mock.call('url', timeout=60), mock.call('url', timeout=60)]
sleep.assert_called_once_with(1)
class TestMemoized:
"""Test the memoized class."""
def test_caching(self):
"""Ensure that caching works for hashable parameters."""
return_value = True
@util.memoized
def some_function(arg):
return return_value
assert some_function(42)
# Let's flip the value of return_value just to make sure the cached value is used and not
# the new value.
return_value = False
# It should still return True, indicating that some_function() was not called again.
assert some_function(42)
def test_caching_different_args(self):
"""Ensure that caching works for hashable parameters, but is sensitive to arguments."""
return_value = True
@util.memoized
def some_function(arg):
return return_value
assert some_function(42)
# Let's flip the value of return_value just to make sure the cached value is not used.
return_value = False
# It should return False because the argument is different.
assert not some_function(41)
def test_dont_cache_lists(self):
"""memoized should not cache calls with list arguments."""
return_value = True
@util.memoized
def some_function(arg):
return return_value
assert some_function(['some', 'list'])
# Let's flip the value of return_value just to make sure it isn't cached.
return_value = False
assert not some_function(['some', 'list'])
def test___get__(self):
"""__get__() should allow us to set the function as an attribute of another object."""
@util.memoized
def some_function(arg):
"""Some docblock"""
return 42
class some_class(object):
thing = some_function
assert some_class().thing() == 42
class TestNoAutoflush:
"""Test the no_autoflush context manager."""
def test_autoflush_disabled(self):
"""Test correct behavior when autoflush is disabled."""
session = mock.MagicMock()
session.autoflush = False
with util.no_autoflush(session):
assert session.autoflush is False
# autoflush should still be False since that was the starting condition.
assert session.autoflush is False
def test_autoflush_enabled(self):
"""Test correct behavior when autoflush is enabled."""
session = mock.MagicMock()
session.autoflush = True
with util.no_autoflush(session):
assert not session.autoflush
# autoflush should again be True since that was the starting condition.
assert session.autoflush
class TestCanWaiveTestResults(base.BasePyTestCase):
"""Test the can_waive_test_results() function."""
def test_access_token_undefined(self):
"""If Bodhi is not configured with an access token, the result should be False."""
config.update({
'test_gating.required': True,
'waiverdb.access_token': None
})
u = Update.query.all()[0]
u.test_gating_status = TestGatingStatus.failed
u.status = models.UpdateStatus.testing
assert not util.can_waive_test_results(None, u)
def test_can_waive_test_results(self):
config.update({
'test_gating.required': True,
'waiverdb.access_token': "secret"
})
u = Update.query.all()[0]
u.test_gating_status = TestGatingStatus.failed
u.status = models.UpdateStatus.testing
assert util.can_waive_test_results(None, u)
def test_gating_required_false(self):
"""
Assert that it should return false if test_gating is not enabled, even if
other conditions are met.
"""
config.update({
'test_gating.required': False,
'waiverdb.access_token': "secret"
})
u = Update.query.all()[0]
u.test_gating_status = TestGatingStatus.failed
u.status = models.UpdateStatus.testing
assert not util.can_waive_test_results(None, u)
def test_all_tests_passed(self):
"""
Assert that it should return false if all tests passed, even if
other conditions are met.
"""
config.update({
'test_gating.required': True,
'waiverdb.access_token': "secret"
})
u = Update.query.all()[0]
u.test_gating_status = TestGatingStatus.passed
u.status = models.UpdateStatus.testing
assert not util.can_waive_test_results(None, u)
def test_update_is_stable(self):
"""
Assert that it should return false if the update is stable, even if
other conditions are met.
"""
config.update({
'test_gating.required': True,
'waiverdb.access_token': "secret"
})
u = Update.query.all()[0]
u.test_gating_status = TestGatingStatus.failed
u.status = models.UpdateStatus.stable
assert not util.can_waive_test_results(None, u)
class TestPagesList:
"""Test the pages_list() function."""
def test_page_in_middle(self):
"""Test for when the current page is in the middle of the pages."""
val = util.pages_list(mock.MagicMock(), 15, 30)
assert val == [1, "..."] + list(range(11, 20)) + ['...', 30]
def test_page_near_end(self):
"""Test for when the current page is near the end of the pages."""
val = util.pages_list(mock.MagicMock(), 6, 7)
assert val == list(range(1, 8))
class TestSanityCheckRepodata(base.BasePyTestCase):
"""Test the sanity_check_repodata() function."""
def setup_method(self, method):
super().setup_method(method)
self.tempdir = tempfile.mkdtemp('bodhi')
def teardown_method(self, method):
shutil.rmtree(self.tempdir)
super().teardown_method(method)
def test_correct_yum_repo_with_xz_compress(self):
"""No Exception should be raised if the repo is normal.
This is using default XZ compression.
"""
base.mkmetadatadir(self.tempdir)
# No exception should be raised here.
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
def test_correct_yum_repo_with_gz_compress(self):
"""No Exception should be raised if the repo is normal.
This is using GZ compression.
"""
base.mkmetadatadir(self.tempdir, compress_type='gz')
# No exception should be raised here.
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
def test_correct_yum_repo_with_bz2_compress(self):
"""No Exception should be raised if the repo is normal.
This is using BZ2 compression.
"""
base.mkmetadatadir(self.tempdir, compress_type='bz2')
# No exception should be raised here.
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
@pytest.mark.skipif(
packaging.version.parse(createrepo_c.VERSION) < packaging.version.parse('1.0.0'),
reason='ZSTD compression requires createrepo_c 1.0.0 or higher'
)
def test_correct_yum_repo_with_zstd_compress(self):
"""No Exception should be raised if the repo is normal.
This is using ZSTD compression.
"""
base.mkmetadatadir(self.tempdir, compress_type='zstd')
# No exception should be raised here.
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
def test_invalid_repo_type(self):
"""A ValueError should be raised with invalid repo type."""
with pytest.raises(ValueError) as excinfo:
util.sanity_check_repodata("so", "wrong", drpms=True)
assert str(excinfo.value) == 'repo_type must be one of module, source, or yum.'
@mock.patch('bodhi.server.util.librepo')
def test_librepo_exception(self, librepo):
"""Verify that LibrepoExceptions are re-wrapped."""
class MockException(Exception):
pass
librepo.LibrepoException = MockException
librepo.Handle.return_value.perform.side_effect = MockException(-1, 'msg', 'general_msg')
with pytest.raises(RepodataException) as excinfo:
util.sanity_check_repodata('/tmp/', 'yum', drpms=True)
assert str(excinfo.value) == 'msg'
def _mkmetadatadir_w_modules(self):
base.mkmetadatadir(self.tempdir)
# We need to add a modules tag to repomd.
repomd_path = os.path.join(self.tempdir, 'repodata', 'repomd.xml')
repomd_tree = ElementTree.parse(repomd_path)
ElementTree.register_namespace('', 'http://linux.duke.edu/metadata/repo')
root = repomd_tree.getroot()
modules_elem = ElementTree.SubElement(root, 'data', type='modules')
# ensure librepo finds something
ElementTree.SubElement(modules_elem, 'location', href='repodata/modules.yaml.gz')
with gzip.open(os.path.join(self.tempdir, 'repodata', 'modules.yaml.gz'), 'w'):
pass
for data in root.findall('{http://linux.duke.edu/metadata/repo}data'):
# module repos don't have drpms or comps.
if data.attrib['type'] in ('group', 'prestodelta'):
root.remove(data)
repomd_tree.write(repomd_path, encoding='UTF-8', xml_declaration=True)
@mock.patch('subprocess.check_output', return_value='Some output')
def test_correct_module_repo(self, *args):
"""No Exception should be raised if the repo is a normal module repo."""
self._mkmetadatadir_w_modules()
# No exception should be raised here.
util.sanity_check_repodata(self.tempdir, repo_type='module', drpms=True)
@mock.patch('subprocess.check_output', return_value='')
def test_module_repo_no_dnf_output(self, *args):
"""No Exception should be raised if the repo is a normal module repo."""
self._mkmetadatadir_w_modules()
with pytest.raises(util.RepodataException) as exc:
util.sanity_check_repodata(self.tempdir, repo_type='module', drpms=True)
assert str(exc.value) == \
("DNF did not return expected output when running test!"
" Test: ['module', 'list'], expected: .*, output: ")
def test_updateinfo_empty_tags(self):
"""RepodataException should be raised if <id/> is found in updateinfo."""
updateinfo = os.path.join(self.tempdir, 'updateinfo.xml')
with open(updateinfo, 'w') as uinfo:
uinfo.write('<id/>')
base.mkmetadatadir(self.tempdir, updateinfo=updateinfo)
with pytest.raises(util.RepodataException) as exc:
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
assert str(exc.value) == 'updateinfo.xml.gz contains empty ID tags'
def test_comps_invalid_notxml(self):
"""RepodataException should be raised if comps is invalid."""
comps = os.path.join(self.tempdir, 'comps.xml')
with open(comps, 'w') as uinfo:
uinfo.write('this is not even xml')
base.mkmetadatadir(self.tempdir, comps=comps)
with pytest.raises(util.RepodataException) as exc:
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
assert str(exc.value) == 'Comps file unable to be parsed'
def test_comps_invalid_nonsense(self):
"""RepodataException should be raised if comps is invalid."""
comps = os.path.join(self.tempdir, 'comps.xml')
with open(comps, 'w') as uinfo:
uinfo.write('<whatever />')
base.mkmetadatadir(self.tempdir, comps=comps)
with pytest.raises(util.RepodataException) as exc:
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
assert str(exc.value) == 'Comps file empty'
def test_repomd_missing_updateinfo(self):
"""If the updateinfo data tag is missing in repomd.xml, an Exception should be raised."""
base.mkmetadatadir(self.tempdir)
repomd_path = os.path.join(self.tempdir, 'repodata', 'repomd.xml')
repomd = ElementTree.parse(repomd_path)
ElementTree.register_namespace('', 'http://linux.duke.edu/metadata/repo')
root = repomd.getroot()
# Find the <data type="updateinfo"> tag and delete it
for data in root.findall('{http://linux.duke.edu/metadata/repo}data'):
if data.attrib['type'] == 'updateinfo':
root.remove(data)
repomd.write(repomd_path, encoding='UTF-8', xml_declaration=True)
with pytest.raises(util.RepodataException) as exc:
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
assert str(exc.value) == 'Required parts not in repomd.xml: updateinfo'
def test_repomd_missing_prestodelta(self):
"""If the prestodelta data tag is missing in repomd.xml, an Exception should be raised."""
base.mkmetadatadir(self.tempdir)
repomd_path = os.path.join(self.tempdir, 'repodata', 'repomd.xml')
repomd = ElementTree.parse(repomd_path)
ElementTree.register_namespace('', 'http://linux.duke.edu/metadata/repo')
root = repomd.getroot()
for data in root.findall('{http://linux.duke.edu/metadata/repo}data'):
if data.attrib['type'] == 'prestodelta':
root.remove(data)
repomd.write(repomd_path, encoding='UTF-8', xml_declaration=True)
with pytest.raises(util.RepodataException) as exc:
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=True)
assert str(exc.value) == 'Required parts not in repomd.xml: prestodelta'
def test_repomd_drpms_disabled(self):
"""If the prestodelta data tag is missing in a repo without DRPMs is fine."""
base.mkmetadatadir(self.tempdir)
repomd_path = os.path.join(self.tempdir, 'repodata', 'repomd.xml')
repomd = ElementTree.parse(repomd_path)
ElementTree.register_namespace('', 'http://linux.duke.edu/metadata/repo')
root = repomd.getroot()
for data in root.findall('{http://linux.duke.edu/metadata/repo}data'):
if data.attrib['type'] == 'prestodelta':
root.remove(data)
repomd.write(repomd_path, encoding='UTF-8', xml_declaration=True)
# No exception should be raised.
util.sanity_check_repodata(self.tempdir, repo_type='yum', drpms=False)
def test_source_true(self):
"""It should not fail source repos for missing prestodelta or comps."""
base.mkmetadatadir(self.tempdir)
repomd_path = os.path.join(self.tempdir, 'repodata', 'repomd.xml')
repomd = ElementTree.parse(repomd_path)
ElementTree.register_namespace('', 'http://linux.duke.edu/metadata/repo')
root = repomd.getroot()
for data in root.findall('{http://linux.duke.edu/metadata/repo}data'):
# Source repos don't have drpms or comps.
if data.attrib['type'] in ('group', 'prestodelta'):
root.remove(data)
repomd.write(repomd_path, encoding='UTF-8', xml_declaration=True)
# No exception should be raised.
util.sanity_check_repodata(self.tempdir, repo_type='source', drpms=True)
class TestTestcaseLink(base.BasePyTestCase):
"""Test the testcase_link() function."""
base_url = 'http://example.com/'
displayed_name = 'test case name'
def setup_method(self, method):
super().setup_method(method)
self.test = mock.Mock()
self.test.name = 'QA:Testcase ' + self.displayed_name
@property
def expected_url(self):
return self.base_url + self.test.name
@pytest.mark.parametrize('short', (False, True))
def test_fn(self, short):
"""Test the function."""
print(self.base_url)
config["test_case_base_url"] = self.base_url
print(config["test_case_base_url"])
retval = util.testcase_link(None, self.test, short=short)
if short:
assert not retval.startswith('Test Case ')
else:
assert retval.startswith('Test Case ')
assert f"href='{self.expected_url}'" in retval
assert f">{self.displayed_name}<" in retval
class TestType2Color:
"""Test the type2color() function."""
def test_colors(self):
"""Test type2color() output."""
context = {'request': mock.MagicMock()}
assert util.type2color(context, 'bugfix') == 'rgba(150,180,205,0.5)'
assert util.type2color(context, 'security') == 'rgba(205,150,180,0.5)'
assert util.type2color(context, 'newpackage') == 'rgba(150,205,180,0.5)'
assert util.type2color(context, 'enhancement') == 'rgba(205,205,150,0.5)'
assert util.type2color(context, 'something_else') == 'rgba(200,200,200,0.5)'
class TestType2Icon:
"""Test the type2icon() function."""
def test_consonant(self):
"""Test type2icon() with a kind that starts with a consonant."""
assert util.type2icon(None, 'security') == \
("<span data-toggle='tooltip' title='This is a security update'>"
"<i class='fa fa-fw fa-shield'></i></span>")
def test_vowel(self):
"""Test type2icon() with a kind that starts with a vowel."""
assert util.type2icon(None, 'enhancement') == \
("<span data-toggle='tooltip' title='This is an enhancement update'>"
"<i class='fa fa-fw fa-bolt'></i></span>")
class TestUtils(base.BasePyTestCase):
def test_get_critpath_components_dummy(self):
""" Ensure that critpath packages can be found using the hardcoded
list.
"""
config.update({
'critpath.type': None,
'critpath_pkgs': ['kernel', 'glibc']
})
assert util.get_critpath_components() == ['kernel', 'glibc']
@mock.patch('bodhi.server.util.log')
def test_get_critpath_components_not_json_not_rpm(self, mock_log):
""" Ensure a warning is logged when the critpath system is not json
and the type of components to search for is not rpm.
"""
config.update({
'critpath.type': None,
'critpath_pkgs': ['kernel', 'glibc']
})
pkgs = util.get_critpath_components('f25', 'module')
assert 'kernel' in pkgs
warning = ('The critpath.type of "(default)" does not support searching '
'for non-RPM components')
mock_log.warning.assert_called_once_with(warning)
@mock.patch('bodhi.server.util.log')
def test_get_critpath_components_json_no_file(self, mock_log, critpath_json_config):
"""Ensure we log a warning and return an empty list when
trying to retrieve critpath info from a non-existent JSON
file.
"""
config.update({'critpath.type': 'json'})
pkgs = util.get_critpath_components('f34')
assert pkgs == []
warning = 'No JSON file found for collection f34'
mock_log.warning.assert_called_once_with(warning)
@mock.patch('bodhi.server.util.log')
def test_get_critpath_components_json_bad_file(self, mock_log, critpath_json_config):
"""Ensure we log a warning and return an empty list when
trying to retrieve critpath info from an invalid JSON
file.
"""
(tempdir, _) = critpath_json_config
config.update({
'critpath.type': 'json',
'critpath.jsonpath': tempdir
})
pkgs = util.get_critpath_components('f35')
assert pkgs == []
warning = 'JSON file for collection f35 is invalid'
mock_log.warning.assert_called_once_with(warning)
def test_get_critpath_components_json_success(self, critpath_json_config):
"""Ensure that critpath packages can be found using JSON
files.
"""
(tempdir, _) = critpath_json_config
config.update({
'critpath.type': 'json',
'critpath.jsonpath': tempdir
})
pkgs = util.get_critpath_components('f36')
assert sorted(pkgs) == [
'ModemManager-glib',
'NetworkManager',
'TurboGears',
'abattis-cantarell-fonts',
'adobe-source-code-pro-fonts'
]
@mock.patch('bodhi.server.util.log')
def test_get_grouped_critpath_components_json_no_file(self, mock_log, critpath_json_config):
"""Ensure we log a warning and return an empty list when
trying to retrieve critpath info from a non-existent JSON
file (from get_grouped_critpath_components).
"""
config.update({'critpath.type': 'json'})
pkgs = util.get_grouped_critpath_components('f34')
assert pkgs == {}
warning = 'No JSON file found for collection f34'
mock_log.warning.assert_called_once_with(warning)
@mock.patch('bodhi.server.util.log')
def test_get_grouped_critpath_components_json_bad_file(self, mock_log, critpath_json_config):
"""Ensure we log a warning and return an empty list when
trying to retrieve critpath info from an invalid JSON
file (from get_grouped_critpath_components).
"""
(tempdir, _) = critpath_json_config
config.update({
'critpath.type': 'json',
'critpath.jsonpath': tempdir
})
pkgs = util.get_grouped_critpath_components('f35')
assert pkgs == {}
warning = 'JSON file for collection f35 is invalid'
mock_log.warning.assert_called_once_with(warning)
def test_get_grouped_critpath_components_not_supported(self):
"""Ensure that get_grouped_critpath_components raises
ValueError when critpath.type doesn't support groups.
"""
config.update({'critpath.type': None})
with pytest.raises(ValueError) as exc:
util.get_grouped_critpath_components('f36')
error = 'critpath.type (default) does not support groups'
assert str(exc.value) == error
def test_get_grouped_critpath_components_success(self, critpath_json_config):
"""Ensure that get_grouped_critpath_components works when
using JSON files.
"""
(tempdir, testdata) = critpath_json_config
config.update({
'critpath.type': 'json',
'critpath.jsonpath': tempdir
})
grouped = util.get_grouped_critpath_components('f36')
assert grouped == testdata['rpm']
# now test with components arg
filtered = util.get_grouped_critpath_components(
'f36', components=['ModemManager-glib'])
assert filtered == {'core': ['ModemManager-glib']}
# now test against 'f35' to confirm we get an empty dict if
# there are no matches
grouped = util.get_grouped_critpath_components('f35')
assert grouped == {}
@mock.patch('bodhi.server.util.http_session')
def test_pagure_api_get(self, session):
""" Ensure that an API request to Pagure works as expected.
"""
session.get.return_value.status_code = 200
expected_json = {
"access_groups": {
"admin": [],
"commit": [],
"ticket": []
},
"access_users": {
"admin": [],
"commit": [],
"owner": [
"mprahl"
],
"ticket": []
},
"close_status": [],
"custom_keys": [],
"date_created": "1494947106",
"description": "Python",
"fullname": "rpms/python",
"id": 2,
"milestones": {},
"name": "python",
"namespace": "rpms",
"parent": None,
"priorities": {},
"tags": [],
"user": {
"fullname": "Matt Prahl",
"name": "mprahl"
}
}
session.get.return_value.json.return_value = expected_json
rv = util.pagure_api_get('http://domain.local/api/0/rpms/python')
assert rv == expected_json
@mock.patch('bodhi.server.util.http_session')
@mock.patch('bodhi.server.util.time.sleep')
def test_pagure_api_get_non_500_error(self, sleep, session):
""" Ensure that an API request to Pagure that raises an error that is
not a 500 error returns the actual error message from the JSON.
"""
session.get.return_value.status_code = 404
session.get.return_value.json.return_value = {
"error": "Project not found",
"error_code": "ENOPROJECT"
}
with pytest.raises(RuntimeError) as exc:
util.pagure_api_get('http://domain.local/api/0/rpms/python')
expected_error = (
'Bodhi failed to get a resource from Pagure at the following URL '
'"http://domain.local/api/0/rpms/python". The status code was '
'"404". The error was "Project not found".')
assert str(exc.value) == expected_error
assert sleep.mock_calls == [mock.call(1), mock.call(1), mock.call(1)]
@mock.patch('bodhi.server.util.http_session')
@mock.patch('bodhi.server.util.time.sleep')
def test_pagure_api_get_500_error(self, sleep, session):
""" Ensure that an API request to Pagure that triggers a 500 error
raises the expected error message.
"""
session.get.return_value.status_code = 500
with pytest.raises(RuntimeError) as exc:
util.pagure_api_get('http://domain.local/api/0/rpms/python')
expected_error = (
'Bodhi failed to get a resource from Pagure at the following URL '
'"http://domain.local/api/0/rpms/python". The status code was '
'"500".')
assert str(exc.value) == expected_error
assert sleep.mock_calls, [mock.call(1), mock.call(1), mock.call(1)]
@mock.patch('bodhi.server.util.http_session')
@mock.patch('bodhi.server.util.time.sleep')
def test_pagure_api_get_non_500_error_no_json(self, sleep, session):
""" Ensure that an API request to Pagure that raises an error that is
not a 500 error and has no JSON returns an error.
"""
session.get.return_value.status_code = 404
session.get.return_value.json.side_effect = ValueError('Not JSON')
with pytest.raises(RuntimeError) as exc:
util.pagure_api_get('http://domain.local/api/0/rpms/python')
expected_error = (
'Bodhi failed to get a resource from Pagure at the following URL '
'"http://domain.local/api/0/rpms/python". The status code was '
'"404". The error was "".')
assert str(exc.value) == expected_error
assert sleep.mock_calls, [mock.call(1), mock.call(1), mock.call(1)]
@mock.patch('bodhi.server.util.http_session')
def test_greenwave_api_post(self, session):
""" Ensure that a POST request to Greenwave works as expected.
"""
session.post.return_value.status_code = 200
expected_json = {
'policies_satisfied': True,
'summary': 'All tests passed',
'applicable_policies': ['bodhiupdate_bodhipush_openqa_workstation'],
'unsatisfied_requirements': []
}
session.post.return_value.json.return_value = expected_json
data = {
'product_version': 'fedora-26',
'decision_context': 'bodhi_push_update_stable',
'subjects': ['foo-1.0.0-1.f26']
}
decision = util.greenwave_api_post('http://domain.local/api/v1.0/decision',
data)
assert decision == expected_json
@mock.patch('bodhi.server.util.http_session')
@mock.patch('bodhi.server.util.time.sleep')
def test_greenwave_api_post_500_error(self, sleep, session):
""" Ensure that a POST request to Greenwave that triggers a 500 error
raises the expected error message.
"""
session.post.return_value.status_code = 500
with pytest.raises(RuntimeError) as exc:
data = {
'product_version': 'fedora-26',
'decision_context': 'bodhi_push_update_stable',
'subjects': ['foo-1.0.0-1.f26']
}
util.greenwave_api_post('http://domain.local/api/v1.0/decision',
data)
expected_error = (
'Bodhi failed to send POST request to Greenwave at the following URL '
'"http://domain.local/api/v1.0/decision". The status code was "500".')
assert str(exc.value) == expected_error
assert sleep.mock_calls == [mock.call(1), mock.call(1), mock.call(1)]
@mock.patch('bodhi.server.util.http_session')
@mock.patch('bodhi.server.util.time.sleep')
def test_greenwave_api_post_non_500_error(self, sleep, session):
""" Ensure that a POST request to Greenwave that raises an error that is
not a 500 error returns the returned JSON.
"""
session.post.return_value.status_code = 404
session.post.return_value.json.return_value = {
"message": "Not found."
}
with pytest.raises(RuntimeError) as exc:
data = {
'product_version': 'fedora-26',
'decision_context': 'bodhi_push_update_stable',
'subjects': ['foo-1.0.0-1.f26']
}
util.greenwave_api_post('http://domain.local/api/v1.0/decision',
data)
expected_error = (
'Bodhi failed to send POST request to Greenwave at the following URL '
'"http://domain.local/api/v1.0/decision". The status code was "404". '
'The error was "{\'message\': \'Not found.\'}".')
assert str(exc.value) == expected_error
assert sleep.mock_calls == [mock.call(1), mock.call(1), mock.call(1)]
@mock.patch('bodhi.server.util.http_session')
@mock.patch('bodhi.server.util.time.sleep')
def test_greenwave_api_post_non_500_error_no_json(self, sleep, session):
""" Ensure that a POST request to Greenwave that raises an error that is
not a 500 error and has no JSON returns an error.
"""
session.post.return_value.status_code = 404
session.post.return_value.json.side_effect = ValueError('Not JSON')
with pytest.raises(RuntimeError) as exc:
data = {
'product_version': 'fedora-26',
'decision_context': 'bodhi_push_update_stable',
'subjects': ['foo-1.0.0-1.f26']
}
util.greenwave_api_post('http://domain.local/api/v1.0/decision',
data)
expected_error = (
'Bodhi failed to send POST request to Greenwave at the following URL '
'"http://domain.local/api/v1.0/decision". The status code was "404". '
'The error was "".')
assert str(exc.value) == expected_error
assert sleep.mock_calls == [mock.call(1), mock.call(1), mock.call(1)]
@mock.patch('bodhi.server.util.http_session')
def test_waiverdb_api_post(self, session):
""" Ensure that a POST request to WaiverDB works as expected.
"""
session.post.return_value.status_code = 200
expected_json = {
'comment': 'this is not true!',
'id': 15,
'product_version': 'fedora-26',
'result_subject': {'productmd.compose.id': 'Fedora-9000-19700101.n.18'},
'result_testcase': 'compose.install_no_user',
'timestamp': '2017-11-28T17:42:04.209638',
'username': 'foo',
'waived': True,
'proxied_by': 'bodhi'
}
session.post.return_value.json.return_value = expected_json
data = {
'product_version': 'fedora-26',
'waived': True,