EVOLUTION-MANAGER

Edit File: dns_google_test.py

"""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)

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)

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"])

self.assertRaises(errors.PluginError, client.add_txt_record,
                          DOMAIN, self.record_name, self.record_content, self.record_ttl)

# 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)

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)

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