1#   Copyright 2013 Nebula Inc.
2#
3#   Licensed under the Apache License, Version 2.0 (the "License"); you may
4#   not use this file except in compliance with the License. You may obtain
5#   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, WITHOUT
11#   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12#   License for the specific language governing permissions and limitations
13#   under the License.
14#
15
16import json
17import sys
18from unittest import mock
19
20from keystoneauth1 import fixture
21import requests
22import six
23
24
25AUTH_TOKEN = "foobar"
26AUTH_URL = "http://0.0.0.0"
27USERNAME = "itchy"
28PASSWORD = "scratchy"
29PROJECT_NAME = "poochie"
30REGION_NAME = "richie"
31INTERFACE = "catchy"
32VERSION = "3"
33
34TEST_RESPONSE_DICT = fixture.V2Token(token_id=AUTH_TOKEN,
35                                     user_name=USERNAME)
36_s = TEST_RESPONSE_DICT.add_service('identity', name='keystone')
37_s.add_endpoint(AUTH_URL + ':5000/v2.0')
38_s = TEST_RESPONSE_DICT.add_service('network', name='neutron')
39_s.add_endpoint(AUTH_URL + ':9696')
40_s = TEST_RESPONSE_DICT.add_service('compute', name='nova')
41_s.add_endpoint(AUTH_URL + ':8774/v2.1')
42_s = TEST_RESPONSE_DICT.add_service('image', name='glance')
43_s.add_endpoint(AUTH_URL + ':9292')
44_s = TEST_RESPONSE_DICT.add_service('object', name='swift')
45_s.add_endpoint(AUTH_URL + ':8080/v1')
46
47TEST_RESPONSE_DICT_V3 = fixture.V3Token(user_name=USERNAME)
48TEST_RESPONSE_DICT_V3.set_project_scope()
49
50TEST_VERSIONS = fixture.DiscoveryList(href=AUTH_URL)
51
52
53def to_unicode_dict(catalog_dict):
54    """Converts dict to unicode dict
55
56    """
57    if isinstance(catalog_dict, dict):
58        return {to_unicode_dict(key): to_unicode_dict(value)
59                for key, value in catalog_dict.items()}
60    elif isinstance(catalog_dict, list):
61        return [to_unicode_dict(element) for element in catalog_dict]
62    elif isinstance(catalog_dict, str):
63        return catalog_dict + u""
64    else:
65        return catalog_dict
66
67
68class FakeStdout(object):
69
70    def __init__(self):
71        self.content = []
72
73    def write(self, text):
74        self.content.append(text)
75
76    def make_string(self):
77        result = ''
78        for line in self.content:
79            result = result + line
80        return result
81
82
83class FakeLog(object):
84
85    def __init__(self):
86        self.messages = {}
87
88    def debug(self, msg):
89        self.messages['debug'] = msg
90
91    def info(self, msg):
92        self.messages['info'] = msg
93
94    def warning(self, msg):
95        self.messages['warning'] = msg
96
97    def error(self, msg):
98        self.messages['error'] = msg
99
100    def critical(self, msg):
101        self.messages['critical'] = msg
102
103
104class FakeApp(object):
105
106    def __init__(self, _stdout, _log):
107        self.stdout = _stdout
108        self.client_manager = None
109        self.api_version = {}
110        self.stdin = sys.stdin
111        self.stdout = _stdout or sys.stdout
112        self.stderr = sys.stderr
113        self.log = _log
114
115
116class FakeOptions(object):
117    def __init__(self, **kwargs):
118        self.os_beta_command = False
119
120
121class FakeClient(object):
122
123    def __init__(self, **kwargs):
124        self.endpoint = kwargs['endpoint']
125        self.token = kwargs['token']
126
127
128class FakeClientManager(object):
129    _api_version = {
130        'image': '2',
131    }
132
133    def __init__(self):
134        self.compute = None
135        self.identity = None
136        self.image = None
137        self.object_store = None
138        self.volume = None
139        self.network = None
140        self.session = None
141        self.auth_ref = None
142        self.auth_plugin_name = None
143        self.network_endpoint_enabled = True
144        self.compute_endpoint_enabled = True
145        self.volume_endpoint_enabled = True
146
147    def get_configuration(self):
148        return {
149            'auth': {
150                'username': USERNAME,
151                'password': PASSWORD,
152                'token': AUTH_TOKEN,
153            },
154            'region': REGION_NAME,
155            'identity_api_version': VERSION,
156        }
157
158    def is_network_endpoint_enabled(self):
159        return self.network_endpoint_enabled
160
161    def is_compute_endpoint_enabled(self):
162        return self.compute_endpoint_enabled
163
164    def is_volume_endpoint_enabled(self, client):
165        return self.volume_endpoint_enabled
166
167
168class FakeModule(object):
169
170    def __init__(self, name, version):
171        self.name = name
172        self.__version__ = version
173        # Workaround for openstacksdk case
174        self.version = mock.Mock()
175        self.version.__version__ = version
176
177
178class FakeResource(object):
179
180    def __init__(self, manager=None, info=None, loaded=False, methods=None):
181        """Set attributes and methods for a resource.
182
183        :param manager:
184            The resource manager
185        :param Dictionary info:
186            A dictionary with all attributes
187        :param bool loaded:
188            True if the resource is loaded in memory
189        :param Dictionary methods:
190            A dictionary with all methods
191        """
192        info = info or {}
193        methods = methods or {}
194
195        self.__name__ = type(self).__name__
196        self.manager = manager
197        self._info = info
198        self._add_details(info)
199        self._add_methods(methods)
200        self._loaded = loaded
201
202    def _add_details(self, info):
203        for (k, v) in info.items():
204            setattr(self, k, v)
205
206    def _add_methods(self, methods):
207        """Fake methods with MagicMock objects.
208
209        For each <@key, @value> pairs in methods, add an callable MagicMock
210        object named @key as an attribute, and set the mock's return_value to
211        @value. When users access the attribute with (), @value will be
212        returned, which looks like a function call.
213        """
214        for (name, ret) in methods.items():
215            method = mock.Mock(return_value=ret)
216            setattr(self, name, method)
217
218    def __repr__(self):
219        reprkeys = sorted(k for k in self.__dict__.keys() if k[0] != '_' and
220                          k != 'manager')
221        info = ", ".join("%s=%s" % (k, getattr(self, k)) for k in reprkeys)
222        return "<%s %s>" % (self.__class__.__name__, info)
223
224    def keys(self):
225        return self._info.keys()
226
227    def to_dict(self):
228        return self._info
229
230    @property
231    def info(self):
232        return self._info
233
234    def __getitem__(self, item):
235        return self._info.get(item)
236
237    def get(self, item, default=None):
238        return self._info.get(item, default)
239
240    def pop(self, key, default_value=None):
241        return self.info.pop(key, default_value)
242
243
244class FakeResponse(requests.Response):
245
246    def __init__(self, headers=None, status_code=200,
247                 data=None, encoding=None):
248        super(FakeResponse, self).__init__()
249
250        headers = headers or {}
251
252        self.status_code = status_code
253
254        self.headers.update(headers)
255        self._content = json.dumps(data)
256        if not isinstance(self._content, six.binary_type):
257            self._content = self._content.encode()
258
259
260class FakeModel(dict):
261
262    def __getattr__(self, key):
263        try:
264            return self[key]
265        except KeyError:
266            raise AttributeError(key)
267