mirror of
https://github.com/certbot/certbot.git
synced 2026-06-06 23:32:06 -04:00
9e5bca4bbf
Part of #7550 This PR makes appropriate corrections to run pylint on Python 3. Why not keeping the dependencies unchanged and just run pylint on Python 3? Because the old version of pylint breaks horribly on Python 3 because of unsupported version of astroid. Why updating pylint + astroid to the latest version ? Because this version only fixes some internal errors occuring during the lint of Certbot code, and is also ready to run gracefully on Python 3.8. Why upgrading mypy ? Because the old version does not support the new version of astroid required to run pylint correctly. Why not upgrading mypy to its latest version ? Because this latest version includes a new typshed version, that adds a lot of new type definitions, and brings dozens of new errors on the Certbot codebase. I would like to fix that in a future PR. That said so, the work has been to find the correct set of new dependency versions, then configure pylint for sane configuration errors in our situation, disable irrelevant lintings errors, then fixing (or ignoring for good reason) the remaining mypy errors. I also made PyLint and MyPy checks run correctly on Windows. * Start configuration * Reconfigure travis * Suspend a check specific to python 3. Start fixing code. * Repair call_args * Fix return + elif lints * Reconfigure development to run mainly on python3 * Remove incompatible Python 3.4 jobs * Suspend pylint in some assertions * Remove pylint in dev * Take first mypy that supports typed-ast>=1.4.0 to limit the migration path * Various return + else lint errors * Find a set of deps that is working with current mypy version * Update local oldest requirements * Remove all current pylint errors * Rebuild letsencrypt-auto * Update mypy to fix pylint with new astroid version, and fix mypy issues * Explain type: ignore * Reconfigure tox, fix none path * Simplify pinning * Remove useless directive * Remove debugging code * Remove continue * Update requirements * Disable unsubscriptable-object check * Disable one check, enabling two more * Plug certbot dev version for oldest requirements * Remove useless disable directives * Remove useless no-member disable * Remove no-else-* checks. Use elif in symetric branches. * Add back assertion * Add new line * Remove unused pylint disable * Remove other pylint disable
331 lines
16 KiB
Python
331 lines
16 KiB
Python
"""Tests for certbot_dns_google._internal.dns_google."""
|
|
|
|
import unittest
|
|
|
|
from googleapiclient import discovery
|
|
from googleapiclient.errors import Error
|
|
from googleapiclient.http import HttpMock
|
|
from httplib2 import ServerNotFoundError
|
|
import mock
|
|
|
|
from certbot import errors
|
|
from certbot.compat import os
|
|
from certbot.errors import PluginError
|
|
from certbot.plugins import dns_test_common
|
|
from certbot.plugins.dns_test_common import DOMAIN
|
|
from certbot.tests import util as test_util
|
|
|
|
ACCOUNT_JSON_PATH = '/not/a/real/path.json'
|
|
API_ERROR = Error()
|
|
PROJECT_ID = "test-test-1"
|
|
|
|
|
|
class AuthenticatorTest(test_util.TempDirTestCase, dns_test_common.BaseAuthenticatorTest):
|
|
|
|
def setUp(self):
|
|
super(AuthenticatorTest, self).setUp()
|
|
|
|
from certbot_dns_google._internal.dns_google import Authenticator
|
|
|
|
path = os.path.join(self.tempdir, 'file.json')
|
|
open(path, "wb").close()
|
|
|
|
super(AuthenticatorTest, self).setUp()
|
|
self.config = mock.MagicMock(google_credentials=path,
|
|
google_propagation_seconds=0) # don't wait during tests
|
|
|
|
self.auth = Authenticator(self.config, "google")
|
|
|
|
self.mock_client = mock.MagicMock()
|
|
# _get_google_client | pylint: disable=protected-access
|
|
self.auth._get_google_client = mock.MagicMock(return_value=self.mock_client)
|
|
|
|
def test_perform(self):
|
|
self.auth.perform([self.achall])
|
|
|
|
expected = [mock.call.add_txt_record(DOMAIN, '_acme-challenge.'+DOMAIN, mock.ANY, mock.ANY)]
|
|
self.assertEqual(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, mock.ANY)]
|
|
self.assertEqual(expected, self.mock_client.mock_calls)
|
|
|
|
@mock.patch('httplib2.Http.request', side_effect=ServerNotFoundError)
|
|
def test_without_auth(self, unused_mock):
|
|
self.config.google_credentials = None
|
|
self.assertRaises(PluginError, self.auth.perform, [self.achall])
|
|
|
|
|
|
class GoogleClientTest(unittest.TestCase):
|
|
record_name = "foo"
|
|
record_content = "bar"
|
|
record_ttl = 42
|
|
zone = "ZONE_ID"
|
|
change = "an-id"
|
|
|
|
def _setUp_client_with_mock(self, zone_request_side_effect):
|
|
from certbot_dns_google._internal.dns_google import _GoogleClient
|
|
|
|
pwd = os.path.dirname(__file__)
|
|
rel_path = 'testdata/discovery.json'
|
|
discovery_file = os.path.join(pwd, rel_path)
|
|
http_mock = HttpMock(discovery_file, {'status': '200'})
|
|
dns_api = discovery.build('dns', 'v1', http=http_mock)
|
|
|
|
client = _GoogleClient(ACCOUNT_JSON_PATH, dns_api)
|
|
|
|
# Setup
|
|
mock_mz = mock.MagicMock()
|
|
mock_mz.list.return_value.execute.side_effect = zone_request_side_effect
|
|
|
|
mock_rrs = mock.MagicMock()
|
|
rrsets = {"rrsets": [{"name": "_acme-challenge.example.org.", "type": "TXT",
|
|
"rrdatas": ["\"example-txt-contents\""]}]}
|
|
mock_rrs.list.return_value.execute.return_value = rrsets
|
|
mock_changes = mock.MagicMock()
|
|
|
|
client.dns.managedZones = mock.MagicMock(return_value=mock_mz)
|
|
client.dns.changes = mock.MagicMock(return_value=mock_changes)
|
|
client.dns.resourceRecordSets = mock.MagicMock(return_value=mock_rrs)
|
|
|
|
return client, mock_changes
|
|
|
|
@mock.patch('googleapiclient.discovery.build')
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google._GoogleClient.get_project_id')
|
|
def test_client_without_credentials(self, get_project_id_mock, credential_mock,
|
|
unused_discovery_mock):
|
|
from certbot_dns_google._internal.dns_google import _GoogleClient
|
|
_GoogleClient(None)
|
|
self.assertFalse(credential_mock.called)
|
|
self.assertTrue(get_project_id_mock.called)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
@mock.patch('certbot_dns_google._internal.dns_google._GoogleClient.get_project_id')
|
|
def test_add_txt_record(self, get_project_id_mock, credential_mock):
|
|
client, changes = self._setUp_client_with_mock([{'managedZones': [{'id': self.zone}]}])
|
|
credential_mock.assert_called_once_with('/not/a/real/path.json', mock.ANY)
|
|
self.assertFalse(get_project_id_mock.called)
|
|
|
|
client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
expected_body = {
|
|
"kind": "dns#change",
|
|
"additions": [
|
|
{
|
|
"kind": "dns#resourceRecordSet",
|
|
"type": "TXT",
|
|
"name": self.record_name + ".",
|
|
"rrdatas": [self.record_content, ],
|
|
"ttl": self.record_ttl,
|
|
},
|
|
],
|
|
}
|
|
|
|
changes.create.assert_called_with(body=expected_body,
|
|
managedZone=self.zone,
|
|
project=PROJECT_ID)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_add_txt_record_and_poll(self, unused_credential_mock):
|
|
client, changes = self._setUp_client_with_mock([{'managedZones': [{'id': self.zone}]}])
|
|
changes.create.return_value.execute.return_value = {'status': 'pending', 'id': self.change}
|
|
changes.get.return_value.execute.return_value = {'status': 'done'}
|
|
|
|
client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
changes.create.assert_called_with(body=mock.ANY,
|
|
managedZone=self.zone,
|
|
project=PROJECT_ID)
|
|
|
|
changes.get.assert_called_with(changeId=self.change,
|
|
managedZone=self.zone,
|
|
project=PROJECT_ID)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_add_txt_record_delete_old(self, unused_credential_mock):
|
|
client, changes = self._setUp_client_with_mock(
|
|
[{'managedZones': [{'id': self.zone}]}])
|
|
# pylint: disable=line-too-long
|
|
mock_get_rrs = "certbot_dns_google._internal.dns_google._GoogleClient.get_existing_txt_rrset"
|
|
with mock.patch(mock_get_rrs) as mock_rrs:
|
|
mock_rrs.return_value = ["sample-txt-contents"]
|
|
client.add_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
self.assertTrue(changes.create.called)
|
|
self.assertTrue("sample-txt-contents" in
|
|
changes.create.call_args_list[0][1]["body"]["deletions"][0]["rrdatas"])
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_add_txt_record_noop(self, unused_credential_mock):
|
|
client, changes = self._setUp_client_with_mock(
|
|
[{'managedZones': [{'id': self.zone}]}])
|
|
client.add_txt_record(DOMAIN, "_acme-challenge.example.org",
|
|
"example-txt-contents", self.record_ttl)
|
|
self.assertFalse(changes.create.called)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_add_txt_record_error_during_zone_lookup(self, unused_credential_mock):
|
|
client, unused_changes = self._setUp_client_with_mock(API_ERROR)
|
|
|
|
self.assertRaises(errors.PluginError, client.add_txt_record,
|
|
DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_add_txt_record_zone_not_found(self, unused_credential_mock):
|
|
client, unused_changes = self._setUp_client_with_mock([{'managedZones': []},
|
|
{'managedZones': []}])
|
|
|
|
self.assertRaises(errors.PluginError, client.add_txt_record,
|
|
DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_add_txt_record_error_during_add(self, unused_credential_mock):
|
|
client, changes = self._setUp_client_with_mock([{'managedZones': [{'id': self.zone}]}])
|
|
changes.create.side_effect = API_ERROR
|
|
|
|
self.assertRaises(errors.PluginError, client.add_txt_record,
|
|
DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_del_txt_record(self, unused_credential_mock):
|
|
client, changes = self._setUp_client_with_mock([{'managedZones': [{'id': self.zone}]}])
|
|
|
|
# pylint: disable=line-too-long
|
|
mock_get_rrs = "certbot_dns_google._internal.dns_google._GoogleClient.get_existing_txt_rrset"
|
|
with mock.patch(mock_get_rrs) as mock_rrs:
|
|
mock_rrs.return_value = ["\"sample-txt-contents\"",
|
|
"\"example-txt-contents\""]
|
|
client.del_txt_record(DOMAIN, "_acme-challenge.example.org",
|
|
"example-txt-contents", self.record_ttl)
|
|
|
|
expected_body = {
|
|
"kind": "dns#change",
|
|
"deletions": [
|
|
{
|
|
"kind": "dns#resourceRecordSet",
|
|
"type": "TXT",
|
|
"name": "_acme-challenge.example.org.",
|
|
"rrdatas": ["\"sample-txt-contents\"", "\"example-txt-contents\""],
|
|
"ttl": self.record_ttl,
|
|
},
|
|
],
|
|
"additions": [
|
|
{
|
|
"kind": "dns#resourceRecordSet",
|
|
"type": "TXT",
|
|
"name": "_acme-challenge.example.org.",
|
|
"rrdatas": ["\"sample-txt-contents\"", ],
|
|
"ttl": self.record_ttl,
|
|
},
|
|
],
|
|
}
|
|
|
|
changes.create.assert_called_with(body=expected_body,
|
|
managedZone=self.zone,
|
|
project=PROJECT_ID)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_del_txt_record_error_during_zone_lookup(self, unused_credential_mock):
|
|
client, unused_changes = self._setUp_client_with_mock(API_ERROR)
|
|
|
|
client.del_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_del_txt_record_zone_not_found(self, unused_credential_mock):
|
|
client, unused_changes = self._setUp_client_with_mock([{'managedZones': []},
|
|
{'managedZones': []}])
|
|
|
|
client.del_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_del_txt_record_error_during_delete(self, unused_credential_mock):
|
|
client, changes = self._setUp_client_with_mock([{'managedZones': [{'id': self.zone}]}])
|
|
changes.create.side_effect = API_ERROR
|
|
|
|
client.del_txt_record(DOMAIN, self.record_name, self.record_content, self.record_ttl)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_get_existing(self, unused_credential_mock):
|
|
client, unused_changes = self._setUp_client_with_mock(
|
|
[{'managedZones': [{'id': self.zone}]}])
|
|
# Record name mocked in setUp
|
|
found = client.get_existing_txt_rrset(self.zone, "_acme-challenge.example.org")
|
|
self.assertEqual(found, ["\"example-txt-contents\""])
|
|
not_found = client.get_existing_txt_rrset(self.zone, "nonexistent.tld")
|
|
self.assertEqual(not_found, None)
|
|
|
|
@mock.patch('oauth2client.service_account.ServiceAccountCredentials.from_json_keyfile_name')
|
|
@mock.patch('certbot_dns_google._internal.dns_google.open',
|
|
mock.mock_open(read_data='{"project_id": "' + PROJECT_ID + '"}'), create=True)
|
|
def test_get_existing_fallback(self, unused_credential_mock):
|
|
client, unused_changes = self._setUp_client_with_mock(
|
|
[{'managedZones': [{'id': self.zone}]}])
|
|
mock_execute = client.dns.resourceRecordSets.return_value.list.return_value.execute
|
|
mock_execute.side_effect = API_ERROR
|
|
|
|
rrset = client.get_existing_txt_rrset(self.zone, "_acme-challenge.example.org")
|
|
self.assertFalse(rrset)
|
|
|
|
def test_get_project_id(self):
|
|
from certbot_dns_google._internal.dns_google import _GoogleClient
|
|
|
|
response = DummyResponse()
|
|
response.status = 200
|
|
|
|
with mock.patch('httplib2.Http.request', return_value=(response, 'test-test-1')):
|
|
project_id = _GoogleClient.get_project_id()
|
|
self.assertEqual(project_id, 'test-test-1')
|
|
|
|
with mock.patch('httplib2.Http.request', return_value=(response, b'test-test-1')):
|
|
project_id = _GoogleClient.get_project_id()
|
|
self.assertEqual(project_id, 'test-test-1')
|
|
|
|
failed_response = DummyResponse()
|
|
failed_response.status = 404
|
|
|
|
with mock.patch('httplib2.Http.request',
|
|
return_value=(failed_response, "some detailed http error response")):
|
|
self.assertRaises(ValueError, _GoogleClient.get_project_id)
|
|
|
|
with mock.patch('httplib2.Http.request', side_effect=ServerNotFoundError):
|
|
self.assertRaises(ServerNotFoundError, _GoogleClient.get_project_id)
|
|
|
|
|
|
class DummyResponse(object):
|
|
"""
|
|
Dummy object to create a fake HTTPResponse (the actual one requires a socket and we only
|
|
need the status attribute)
|
|
"""
|
|
def __init__(self):
|
|
self.status = 200
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() # pragma: no cover
|