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
|
"""Tests for certbot_dns_digitalocean._internal.dns_digitalocean."""
import sys
import unittest
from unittest import mock
import digitalocean
import pytest
from certbot import errors
from certbot.compat import os
from certbot.plugins import dns_test_common
from certbot.plugins.dns_test_common import DOMAIN
from certbot.tests import util as test_util
API_ERROR = digitalocean.DataReadError()
TOKEN = 'a-token'
class AuthenticatorTest(test_util.TempDirTestCase, dns_test_common.BaseAuthenticatorTest):
def setUp(self):
from certbot_dns_digitalocean._internal.dns_digitalocean import Authenticator
super().setUp()
path = os.path.join(self.tempdir, 'file.ini')
dns_test_common.write({"digitalocean_token": TOKEN}, path)
self.config = mock.MagicMock(digitalocean_credentials=path,
digitalocean_propagation_seconds=0) # don't wait during tests
self.auth = Authenticator(self.config, "digitalocean")
self.mock_client = mock.MagicMock()
# _get_digitalocean_client | pylint: disable=protected-access
# workaround for wont-fix https://github.com/python/mypy/issues/2427 that works with
# both strict and non-strict mypy
setattr(self.auth, '_get_digitalocean_client',
mock.MagicMock(return_value=self.mock_client))
@test_util.patch_display_util()
def test_perform(self, unused_mock_get_utility):
self.auth.perform([self.achall])
expected = [mock.call.add_txt_record(DOMAIN, '_acme-challenge.'+DOMAIN, mock.ANY, 30)]
assert expected == self.mock_client.mock_calls
def test_cleanup(self):
# _attempt_cleanup | pylint: disable=protected-access
self.auth._attempt_cleanup = True
self.auth.cleanup([self.achall])
expected = [mock.call.del_txt_record(DOMAIN, '_acme-challenge.'+DOMAIN, mock.ANY)]
assert expected == self.mock_client.mock_calls
class DigitalOceanClientTest(unittest.TestCase):
id_num = 1
record_prefix = "_acme-challenge"
record_name = record_prefix + "." + DOMAIN
record_content = "bar"
record_ttl = 60
def setUp(self):
from certbot_dns_digitalocean._internal.dns_digitalocean import _DigitalOceanClient
self.digitalocean_client = _DigitalOceanClient(TOKEN)
self.manager = mock.MagicMock()
self.digitalocean_client.manager = self.manager
def test_add_txt_record(self):
wrong_domain_mock = mock.MagicMock()
wrong_domain_mock.name = "other.invalid"
wrong_domain_mock.create_new_domain_record.side_effect = AssertionError('Wrong Domain')
domain_mock = mock.MagicMock()
domain_mock.name = DOMAIN
domain_mock.create_new_domain_record.return_value = {'domain_record': {'id': self.id_num}}
self.manager.get_all_domains.return_value = [wrong_domain_mock, domain_mock]
self.digitalocean_client.add_txt_record(DOMAIN, self.record_name, self.record_content,
self.record_ttl)
domain_mock.create_new_domain_record.assert_called_with(type='TXT',
name=self.record_prefix,
data=self.record_content,
ttl=self.record_ttl)
def test_add_txt_record_fail_to_find_domain(self):
self.manager.get_all_domains.return_value = []
with pytest.raises(errors.PluginError):
self.digitalocean_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
def test_add_txt_record_error_finding_domain(self):
self.manager.get_all_domains.side_effect = API_ERROR
with pytest.raises(errors.PluginError):
self.digitalocean_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
def test_add_txt_record_error_creating_record(self):
domain_mock = mock.MagicMock()
domain_mock.name = DOMAIN
domain_mock.create_new_domain_record.side_effect = API_ERROR
self.manager.get_all_domains.return_value = [domain_mock]
with pytest.raises(errors.PluginError):
self.digitalocean_client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
def test_del_txt_record(self):
first_record_mock = mock.MagicMock()
first_record_mock.type = 'TXT'
first_record_mock.name = "DIFFERENT"
first_record_mock.data = self.record_content
correct_record_mock = mock.MagicMock()
correct_record_mock.type = 'TXT'
correct_record_mock.name = self.record_prefix
correct_record_mock.data = self.record_content
last_record_mock = mock.MagicMock()
last_record_mock.type = 'TXT'
last_record_mock.name = self.record_prefix
last_record_mock.data = "DIFFERENT"
domain_mock = mock.MagicMock()
domain_mock.name = DOMAIN
domain_mock.get_records.return_value = [first_record_mock,
correct_record_mock,
last_record_mock]
self.manager.get_all_domains.return_value = [domain_mock]
self.digitalocean_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
assert correct_record_mock.destroy.called
assert not first_record_mock.destroy.call_args_list
assert not last_record_mock.destroy.call_args_list
def test_del_txt_record_error_finding_domain(self):
self.manager.get_all_domains.side_effect = API_ERROR
self.digitalocean_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
def test_del_txt_record_error_finding_record(self):
domain_mock = mock.MagicMock()
domain_mock.name = DOMAIN
domain_mock.get_records.side_effect = API_ERROR
self.manager.get_all_domains.return_value = [domain_mock]
self.digitalocean_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
def test_del_txt_record_error_deleting_record(self):
record_mock = mock.MagicMock()
record_mock.type = 'TXT'
record_mock.name = self.record_prefix
record_mock.data = self.record_content
record_mock.destroy.side_effect = API_ERROR
domain_mock = mock.MagicMock()
domain_mock.name = DOMAIN
domain_mock.get_records.return_value = [record_mock]
self.manager.get_all_domains.return_value = [domain_mock]
self.digitalocean_client.del_txt_record(DOMAIN, self.record_name, self.record_content)
if __name__ == "__main__":
sys.exit(pytest.main(sys.argv[1:] + [__file__])) # pragma: no cover
|
