1# Copyright 2016 Google Inc. All rights reserved.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import datetime
16import json
17
18import httplib2
19import mock
20from six.moves import http_client
21import unittest2
22
23from oauth2client.contrib import _metadata
24
25PATH = 'instance/service-accounts/default'
26DATA = {'foo': 'bar'}
27EXPECTED_URL = (
28    'http://metadata.google.internal/computeMetadata/v1/instance'
29    '/service-accounts/default')
30EXPECTED_KWARGS = dict(headers=_metadata.METADATA_HEADERS)
31
32
33def request_mock(status, content_type, content):
34    return mock.MagicMock(return_value=(
35        httplib2.Response(
36            {'status': status, 'content-type': content_type}
37        ),
38        content.encode('utf-8')
39    ))
40
41
42class TestMetadata(unittest2.TestCase):
43
44    def test_get_success_json(self):
45        http_request = request_mock(
46            http_client.OK, 'application/json', json.dumps(DATA))
47        self.assertEqual(
48            _metadata.get(http_request, PATH),
49            DATA
50        )
51        http_request.assert_called_once_with(EXPECTED_URL, **EXPECTED_KWARGS)
52
53    def test_get_success_string(self):
54        http_request = request_mock(
55            http_client.OK, 'text/html', '<p>Hello World!</p>')
56        self.assertEqual(
57            _metadata.get(http_request, PATH),
58            '<p>Hello World!</p>'
59        )
60        http_request.assert_called_once_with(EXPECTED_URL, **EXPECTED_KWARGS)
61
62    def test_get_failure(self):
63        http_request = request_mock(
64            http_client.NOT_FOUND, 'text/html', '<p>Error</p>')
65        with self.assertRaises(httplib2.HttpLib2Error):
66            _metadata.get(http_request, PATH)
67
68        http_request.assert_called_once_with(EXPECTED_URL, **EXPECTED_KWARGS)
69
70    @mock.patch(
71        'oauth2client.client._UTCNOW',
72        return_value=datetime.datetime.min)
73    def test_get_token_success(self, now):
74        http_request = request_mock(
75            http_client.OK,
76            'application/json',
77            json.dumps({'access_token': 'a', 'expires_in': 100})
78        )
79        token, expiry = _metadata.get_token(http_request=http_request)
80        self.assertEqual(token, 'a')
81        self.assertEqual(
82            expiry, datetime.datetime.min + datetime.timedelta(seconds=100))
83        http_request.assert_called_once_with(
84            EXPECTED_URL + '/token',
85            **EXPECTED_KWARGS
86        )
87        now.assert_called_once_with()
88
89    def test_service_account_info(self):
90        http_request = request_mock(
91            http_client.OK, 'application/json', json.dumps(DATA))
92        info = _metadata.get_service_account_info(http_request)
93        self.assertEqual(info, DATA)
94        http_request.assert_called_once_with(
95            EXPECTED_URL + '/?recursive=True',
96            **EXPECTED_KWARGS
97        )
98