Source Code for Module cm_api_tests.utils

 1  # Licensed to Cloudera, Inc. under one 
 2  # or more contributor license agreements.  See the NOTICE file 
 3  # distributed with this work for additional information 
 4  # regarding copyright ownership.  Cloudera, Inc. licenses this file 
 5  # to you under the Apache License, Version 2.0 (the 
 6  # "License"); you may not use this file except in compliance 
 7  # with the License.  You may obtain a copy of the License at 
 8  # 
 9  #     http://www.apache.org/licenses/LICENSE-2.0 
10  # 
11  # Unless required by applicable law or agreed to in writing, software 
12  # distributed under the License is distributed on an "AS IS" BASIS, 
13  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
14  # See the License for the specific language governing permissions and 
15  # limitations under the License. 
16   
17  from cm_api import api_client 
18  from cm_api.resource import Resource 
19   
20  try: 
21    import json 
22  except ImportError: 
23    import simplejson as json 
24   
25 -class MockResource(Resource): 
26    """ 
27    Allows code to control the behavior of a resource's "invoke" method for 
28    unit testing. 
29    """ 
30   
31 -  def __init__(self, test, version=api_client.API_CURRENT_VERSION): 
32      self._next_expect = None 
33      self.test = test 
34      self.version = version 
35   
36    @property 
37 -  def base_url(self): 
38      return "" 
39   
40 -  def invoke(self, method, relpath=None, params=None, data=None, headers=None): 
41      """ 
42      Checks the expected input data and returns the appropriate data to the caller. 
43      """ 
44      exp_method, exp_path, exp_params, exp_data, exp_headers, retdata = self._next_expect 
45      self._next_expect = None 
46   
47      if exp_method is not None: 
48        self.test.assertEquals(exp_method, method) 
49      if exp_path is not None: 
50        self.test.assertEquals(exp_path, relpath) 
51      if exp_params is not None: 
52        self.test.assertEquals(exp_params, params) 
53      if exp_data is not None: 
54        self.test.assertEquals(exp_data, data) 
55      if exp_headers is not None: 
56        self.test.assertEquals(exp_headers, headers) 
57      return retdata 
58   
59 -  def expect(self, method, reqpath, params=None, data=None, headers=None, 
60        retdata=None): 
61      """ 
62      Sets the data to expect in the next call to invoke(). 
63   
64      @param method: method to expect, or None for any. 
65      @param reqpath: request path, or None for any. 
66      @param params: query parameters, or None for any. 
67      @param data: request body, or None for any. 
68      @param headers: request headers, or None for any. 
69      @param retdata: data to return from the invoke call. 
70      """ 
71      self._next_expect = (method, reqpath, params, data, headers, retdata) 
72   
73 -def deserialize(raw_data, cls): 
74    """ 
75    Deserializes raw JSON data into an instance of cls. 
76   
77    The data is deserialized, serialized again using the class's to_json_dict() 
78    implementation, and deserialized again, to make sure both from_json_dict() 
79    and to_json_dict() are working. 
80    """ 
81    instance = cls.from_json_dict(json.loads(raw_data), None) 
82    return cls.from_json_dict(instance.to_json_dict(preserve_ro=True), None) 
83