Source Code for Module cm_api_tests.test_impala

 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  import datetime 
18  import json 
19  import unittest 
20  from cm_api.endpoints.clusters import * 
21  from cm_api.endpoints.services import * 
22  from cm_api.endpoints.types import * 
23  from cm_api_tests import utils 
24   
25 -class TestImpala(unittest.TestCase): 
26   
27 -  def test_get_queries(self): 
28      resource = utils.MockResource(self) 
29      service = ApiService(resource, name="bar") 
30   
31      time = datetime.datetime.now() 
32   
33      resource.expect("GET", "/cm/service/impalaQueries", 
34          retdata={ 'queries': [], 'warnings' : [] }, 
35          params={ 'from':time.isoformat(), 'to':time.isoformat(), \ 
36              'filter':'', 'limit':100, 'offset':0 }) 
37      resp = service.get_impala_queries(time, time) 
38      self.assertEquals(0, len(resp.queries)) 
39   
40 -  def test_get_details(self): 
41      resource = utils.MockResource(self) 
42      service = ApiService(resource, name="bar") 
43   
44      resource.expect("GET", "/cm/service/impalaQueries/randomId", 
45          retdata={ 'details': '' }, 
46          params={ 'format':'text' } ) 
47      resp = service.get_query_details('randomId') 
48      self.assertEquals('', resp.details) 
49   
50 -  def test_cancel_query(self): 
51      resource = utils.MockResource(self) 
52      service = ApiService(resource, name="bar") 
53   
54      resource.expect("POST", "/cm/service/impalaQueries/randomId/cancel", 
55          retdata={ 'warning' : 'test' }) 
56      resp = service.cancel_impala_query('randomId') 
57      self.assertEquals('test', resp.warning) 
58