|
6 | 6 | from ssl import CertificateError
|
7 | 7 | from socket import timeout, error
|
8 | 8 |
|
| 9 | +import pytest |
9 | 10 |
|
10 | 11 | from django.utils.timezone import now
|
11 | 12 | from datetime import timedelta
|
12 | 13 |
|
13 |
| - |
14 |
| -from django.test import TestCase |
15 | 14 | from django.core.management import call_command
|
16 | 15 |
|
17 |
| - |
18 |
| -from mirrors.tests import create_mirror_url |
19 |
| -from mirrors.models import MirrorLog, CheckLocation |
20 |
| - |
21 |
| - |
22 |
| -class MirrorCheckTest(TestCase): |
23 |
| - def setUp(self): |
24 |
| - self.mirror_url = create_mirror_url() |
25 |
| - |
26 |
| - def tearDown(self): |
27 |
| - self.mirror_url.delete() |
28 |
| - |
29 |
| - @mock.patch('urllib.request.Request') |
30 |
| - @mock.patch('urllib.request.urlopen') |
31 |
| - def test_invalid(self, urlopen, Request): |
32 |
| - urlopen.return_value.read.return_value = 'data' |
33 |
| - Request.get_host.return_value = 'archlinux.org' |
34 |
| - Request.type.return_value = 'https' |
35 |
| - |
36 |
| - call_command('mirrorcheck') |
37 |
| - mirrorlog = MirrorLog.objects.first() |
38 |
| - self.assertNotEqual(mirrorlog.error, '') |
39 |
| - self.assertEqual(mirrorlog.is_success, False) |
40 |
| - |
41 |
| - @mock.patch('urllib.request.Request') |
42 |
| - @mock.patch('urllib.request.urlopen') |
43 |
| - def test_valid(self, urlopen, Request): |
44 |
| - urlopen.return_value.read.return_value = str(int(time.time())) |
45 |
| - Request.get_host.return_value = 'archlinux.org' |
46 |
| - Request.type.return_value = 'https' |
47 |
| - |
48 |
| - call_command('mirrorcheck') |
49 |
| - mirrorlog = MirrorLog.objects.first() |
50 |
| - self.assertEqual(mirrorlog.error, '') |
51 |
| - self.assertEqual(mirrorlog.is_success, True) |
52 |
| - |
53 |
| - @mock.patch('urllib.request.Request') |
54 |
| - @mock.patch('urllib.request.urlopen') |
55 |
| - def test_valid_olddate(self, urlopen, Request): |
56 |
| - urlopen.return_value.read.return_value = str(int(time.time())) |
57 |
| - Request.get_host.return_value = 'archlinux.org' |
58 |
| - Request.type.return_value = 'https' |
59 |
| - |
60 |
| - date = now() - timedelta(days=600) |
61 |
| - MirrorLog.objects.create(url=self.mirror_url, check_time=date) |
62 |
| - call_command('mirrorcheck') |
63 |
| - self.assertEqual(len(MirrorLog.objects.all()), 1) |
64 |
| - |
65 |
| - @mock.patch('urllib.request.Request') |
66 |
| - @mock.patch('urllib.request.urlopen') |
67 |
| - def test_not_found(self, urlopen, Request): |
68 |
| - excp = HTTPError('https://archlinux.org/404.txt', 404, 'Not Found', '', None) |
69 |
| - urlopen.return_value.read.side_effect = excp |
70 |
| - Request.get_host.return_value = 'archlinux.org' |
71 |
| - Request.type.return_value = 'https' |
72 |
| - |
73 |
| - call_command('mirrorcheck') |
74 |
| - mirrorlog = MirrorLog.objects.first() |
75 |
| - self.assertEqual(mirrorlog.error, str(excp)) |
76 |
| - self.assertEqual(mirrorlog.is_success, False) |
77 |
| - |
78 |
| - @mock.patch('urllib.request.Request') |
79 |
| - @mock.patch('urllib.request.urlopen') |
80 |
| - def test_not_found_variant(self, urlopen, Request): |
81 |
| - excp = BadStatusLine('') |
82 |
| - urlopen.return_value.read.side_effect = excp |
83 |
| - Request.get_host.return_value = 'archlinux.org' |
84 |
| - Request.type.return_value = 'https' |
85 |
| - |
86 |
| - call_command('mirrorcheck') |
87 |
| - mirrorlog = MirrorLog.objects.first() |
88 |
| - self.assertIn('Exception in processing', mirrorlog.error) |
89 |
| - self.assertEqual(mirrorlog.is_success, False) |
90 |
| - |
91 |
| - @mock.patch('urllib.request.Request') |
92 |
| - @mock.patch('urllib.request.urlopen') |
93 |
| - def test_cert_error(self, urlopen, Request): |
94 |
| - excp = CertificateError('certificate error') |
95 |
| - urlopen.return_value.read.side_effect = excp |
96 |
| - Request.get_host.return_value = 'archlinux.org' |
97 |
| - Request.type.return_value = 'https' |
98 |
| - |
99 |
| - call_command('mirrorcheck') |
100 |
| - mirrorlog = MirrorLog.objects.first() |
101 |
| - self.assertIn('certificate error', mirrorlog.error) |
102 |
| - self.assertEqual(mirrorlog.is_success, False) |
103 |
| - |
104 |
| - @mock.patch('urllib.request.Request') |
105 |
| - @mock.patch('urllib.request.urlopen') |
106 |
| - def test_general_httpexception(self, urlopen, Request): |
107 |
| - excp = URLError('550 No such file', '550.txt') |
108 |
| - urlopen.return_value.read.side_effect = excp |
109 |
| - Request.get_host.return_value = 'archlinux.org' |
110 |
| - Request.type.return_value = 'https' |
111 |
| - |
112 |
| - call_command('mirrorcheck') |
113 |
| - mirrorlog = MirrorLog.objects.first() |
114 |
| - self.assertIn(excp.reason, mirrorlog.error) |
115 |
| - self.assertEqual(mirrorlog.is_success, False) |
116 |
| - |
117 |
| - @mock.patch('urllib.request.Request') |
118 |
| - @mock.patch('urllib.request.urlopen') |
119 |
| - def test_socket_timeout(self, urlopen, Request): |
120 |
| - excp = timeout('timeout') |
121 |
| - urlopen.return_value.read.side_effect = excp |
122 |
| - Request.get_host.return_value = 'archlinux.org' |
123 |
| - Request.type.return_value = 'https' |
124 |
| - |
125 |
| - call_command('mirrorcheck') |
126 |
| - mirrorlog = MirrorLog.objects.first() |
127 |
| - self.assertEqual('Connection timed out.', mirrorlog.error) |
128 |
| - self.assertEqual(mirrorlog.is_success, False) |
129 |
| - |
130 |
| - @mock.patch('urllib.request.Request') |
131 |
| - @mock.patch('urllib.request.urlopen') |
132 |
| - def test_socket_error(self, urlopen, Request): |
133 |
| - excp = error('error') |
134 |
| - urlopen.return_value.read.side_effect = excp |
135 |
| - Request.get_host.return_value = 'archlinux.org' |
136 |
| - Request.type.return_value = 'https' |
137 |
| - |
138 |
| - call_command('mirrorcheck') |
139 |
| - mirrorlog = MirrorLog.objects.first() |
140 |
| - self.assertEqual(str(excp), mirrorlog.error) |
141 |
| - self.assertEqual(mirrorlog.is_success, False) |
142 |
| - |
143 |
| - def test_checklocation(self): |
144 |
| - with self.assertRaises(CheckLocation.DoesNotExist) as e: |
145 |
| - call_command('mirrorcheck', '-l', '9999999') |
146 |
| - self.assertEqual('CheckLocation matching query does not exist.', str(e.exception)) |
147 |
| - |
148 |
| - def test_checklocation_model(self): |
149 |
| - checkloc = CheckLocation.objects.create(hostname='archlinux.org', |
150 |
| - source_ip='1.1.1.1') |
151 |
| - with mock.patch('mirrors.management.commands.mirrorcheck.logger') as logger: |
152 |
| - call_command('mirrorcheck', '-l', f'{checkloc.id}') |
153 |
| - logger.info.assert_called() |
154 |
| - |
155 |
| - checkloc.delete() |
| 16 | +from mirrors.models import CheckLocation, MirrorLog |
| 17 | +from mirrors.tests.conftest import HOSTNAME, PROTOCOL |
| 18 | + |
| 19 | + |
| 20 | +def mocked_request(urlopen, Request, return_value): |
| 21 | + urlopen.return_value.read.return_value = return_value |
| 22 | + Request.get_host.return_value = HOSTNAME |
| 23 | + Request.type.return_value = PROTOCOL |
| 24 | + |
| 25 | + |
| 26 | +def mocked_request_exception(urlopen, Request, excp): |
| 27 | + urlopen.return_value.read.side_effect = excp |
| 28 | + Request.get_host.return_value = HOSTNAME |
| 29 | + Request.type.return_value = PROTOCOL |
| 30 | + |
| 31 | + |
| 32 | +@mock.patch('urllib.request.Request') |
| 33 | +@mock.patch('urllib.request.urlopen') |
| 34 | +def test_invalid(urlopen, Request, mirrorurl): |
| 35 | + mocked_request(urlopen, Request, 'data') |
| 36 | + call_command('mirrorcheck') |
| 37 | + mirrorlog = MirrorLog.objects.first() |
| 38 | + assert mirrorlog.error != '' |
| 39 | + assert not mirrorlog.is_success |
| 40 | + |
| 41 | + |
| 42 | +@mock.patch('urllib.request.Request') |
| 43 | +@mock.patch('urllib.request.urlopen') |
| 44 | +def test_valid(urlopen, Request, mirrorurl): |
| 45 | + mocked_request(urlopen, Request, str(int(time.time()))) |
| 46 | + call_command('mirrorcheck') |
| 47 | + mirrorlog = MirrorLog.objects.first() |
| 48 | + assert mirrorlog.error == '' |
| 49 | + assert mirrorlog.is_success |
| 50 | + |
| 51 | + |
| 52 | +@mock.patch('urllib.request.Request') |
| 53 | +@mock.patch('urllib.request.urlopen') |
| 54 | +def test_valid_olddate(urlopen, Request, mirrorurl): |
| 55 | + mocked_request(urlopen, Request, str(int(time.time()))) |
| 56 | + date = now() - timedelta(days=600) |
| 57 | + MirrorLog.objects.create(url=mirrorurl, check_time=date) |
| 58 | + call_command('mirrorcheck') |
| 59 | + assert len(MirrorLog.objects.all()) == 1 |
| 60 | + |
| 61 | + |
| 62 | +@mock.patch('urllib.request.Request') |
| 63 | +@mock.patch('urllib.request.urlopen') |
| 64 | +def test_not_found(urlopen, Request, mirrorurl): |
| 65 | + excp = HTTPError('https://archlinux.org/404.txt', 404, 'Not Found', '', None) |
| 66 | + mocked_request_exception(urlopen, Request, excp) |
| 67 | + call_command('mirrorcheck') |
| 68 | + mirrorlog = MirrorLog.objects.first() |
| 69 | + assert mirrorlog.error == str(excp) |
| 70 | + assert not mirrorlog.is_success |
| 71 | + |
| 72 | + |
| 73 | +@mock.patch('urllib.request.Request') |
| 74 | +@mock.patch('urllib.request.urlopen') |
| 75 | +def test_not_found_variant(urlopen, Request, mirrorurl): |
| 76 | + excp = BadStatusLine('') |
| 77 | + mocked_request_exception(urlopen, Request, excp) |
| 78 | + call_command('mirrorcheck') |
| 79 | + mirrorlog = MirrorLog.objects.first() |
| 80 | + assert 'Exception in processing' in mirrorlog.error |
| 81 | + assert not mirrorlog.is_success |
| 82 | + |
| 83 | + |
| 84 | +@mock.patch('urllib.request.Request') |
| 85 | +@mock.patch('urllib.request.urlopen') |
| 86 | +def test_cert_error(urlopen, Request, mirrorurl): |
| 87 | + excp = CertificateError('certificate error') |
| 88 | + mocked_request_exception(urlopen, Request, excp) |
| 89 | + call_command('mirrorcheck') |
| 90 | + mirrorlog = MirrorLog.objects.first() |
| 91 | + assert 'certificate error' in mirrorlog.error |
| 92 | + assert not mirrorlog.is_success |
| 93 | + |
| 94 | + |
| 95 | +@mock.patch('urllib.request.Request') |
| 96 | +@mock.patch('urllib.request.urlopen') |
| 97 | +def test_general_httpexception(urlopen, Request, mirrorurl): |
| 98 | + excp = URLError('550 No such file', '550.txt') |
| 99 | + mocked_request_exception(urlopen, Request, excp) |
| 100 | + call_command('mirrorcheck') |
| 101 | + mirrorlog = MirrorLog.objects.first() |
| 102 | + assert excp.reason in mirrorlog.error |
| 103 | + assert not mirrorlog.is_success |
| 104 | + |
| 105 | + |
| 106 | +@mock.patch('urllib.request.Request') |
| 107 | +@mock.patch('urllib.request.urlopen') |
| 108 | +def test_socket_timeout(urlopen, Request, mirrorurl): |
| 109 | + excp = timeout('timeout') |
| 110 | + mocked_request_exception(urlopen, Request, excp) |
| 111 | + |
| 112 | + call_command('mirrorcheck') |
| 113 | + mirrorlog = MirrorLog.objects.first() |
| 114 | + assert 'Connection timed out.' in mirrorlog.error |
| 115 | + assert not mirrorlog.is_success |
| 116 | + |
| 117 | + |
| 118 | +@mock.patch('urllib.request.Request') |
| 119 | +@mock.patch('urllib.request.urlopen') |
| 120 | +def test_socket_error(urlopen, Request, mirrorurl): |
| 121 | + excp = error('error') |
| 122 | + mocked_request_exception(urlopen, Request, excp) |
| 123 | + |
| 124 | + call_command('mirrorcheck') |
| 125 | + mirrorlog = MirrorLog.objects.first() |
| 126 | + assert str(excp) in mirrorlog.error |
| 127 | + assert not mirrorlog.is_success |
| 128 | + |
| 129 | + |
| 130 | +def test_checklocation_fail(db): |
| 131 | + with pytest.raises(CheckLocation.DoesNotExist) as e: |
| 132 | + call_command('mirrorcheck', '-l', '1') |
| 133 | + assert 'CheckLocation matching query does not exist.' == str(e.value) |
| 134 | + |
| 135 | + |
| 136 | +def test_checklocation_model(checklocation): |
| 137 | + with mock.patch('mirrors.management.commands.mirrorcheck.logger') as logger: |
| 138 | + call_command('mirrorcheck', '-l', '1') |
| 139 | + logger.info.assert_called() |
0 commit comments