"Fossies" - the Fresh Open Source Software Archive

Member "openstack-cyborg-9.0.0/cyborg/tests/unit/api/base.py" (5 Oct 2022, 9667 Bytes) of package /linux/misc/openstack/openstack-cyborg-9.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "base.py": 8.0.0_vs_9.0.0.

    1 # Copyright 2017 Huawei Technologies Co.,LTD.
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 
   16 """Base classes for API tests."""
   17 
   18 from oslo_config import cfg
   19 from oslo_context import context
   20 import pecan
   21 import pecan.testing
   22 
   23 from cyborg.tests.unit.db import base
   24 
   25 cfg.CONF.import_group('keystone_authtoken', 'keystonemiddleware.auth_token')
   26 
   27 
   28 class BaseApiTest(base.DbTestCase):
   29     """Pecan controller functional testing class.
   30 
   31     Used for functional tests of Pecan controllers where you need to
   32     test your literal application and its integration with the
   33     framework.
   34     """
   35 
   36     PATH_PREFIX = ''
   37 
   38     def setUp(self):
   39         super(BaseApiTest, self).setUp()
   40         cfg.CONF.set_override("auth_version", "v3",
   41                               group='keystone_authtoken')
   42         cfg.CONF.set_override("admin_user", "admin",
   43                               group='keystone_authtoken')
   44         self.app = self._make_app()
   45 
   46         def reset_pecan():
   47             pecan.set_config({}, overwrite=True)
   48 
   49         self.addCleanup(reset_pecan)
   50 
   51     def flags(self, **kw):
   52         """Override flag variables for a test."""
   53         group = kw.pop('group', None)
   54         for k, v in kw.items():
   55             cfg.CONF.set_override(k, v, group)
   56 
   57     def _make_app(self):
   58         # Determine where we are so we can set up paths in the config
   59         root_dir = self.get_path()
   60 
   61         self.app_config = {
   62             'app': {
   63                 'root': 'cyborg.api.controllers.root.RootController',
   64                 'modules': ['cyborg.api'],
   65                 'static_root': '%s/public' % root_dir,
   66                 'template_path': '%s/api/templates' % root_dir,
   67                 'acl_public_routes': ['/', '/v1/.*'],
   68             },
   69         }
   70         return pecan.testing.load_test_app(self.app_config)
   71 
   72     def _request_json(self, path, params, expect_errors=False, headers=None,
   73                       method="post", extra_environ=None, status=None):
   74         """Sends simulated HTTP request to Pecan test app.
   75 
   76         :param path: url path of target service
   77         :param params: content for wsgi.input of request
   78         :param expect_errors: Boolean value; whether an error is expected based
   79                               on request
   80         :param headers: a dictionary of headers to send along with the request
   81         :param method: Request method type. Appropriate method function call
   82                        should be used rather than passing attribute in.
   83         :param extra_environ: a dictionary of environ variables to send along
   84                               with the request
   85         :param status: expected status code of response
   86         """
   87         response = getattr(self.app, "%s_json" % method)(
   88             str(path),
   89             params=params,
   90             headers=headers,
   91             status=status,
   92             extra_environ=extra_environ,
   93             expect_errors=expect_errors
   94         )
   95         return response
   96 
   97     def post_json(self, path, params, expect_errors=False, headers=None,
   98                   extra_environ=None, status=None):
   99         """Sends simulated HTTP POST request to Pecan test app.
  100 
  101         :param path: url path of target service
  102         :param params: content for wsgi.input of request
  103         :param expect_errors: Boolean value; whether an error is expected based
  104                               on request
  105         :param headers: a dictionary of headers to send along with the request
  106         :param extra_environ: a dictionary of environ variables to send along
  107                               with the request
  108         :param status: expected status code of response
  109         """
  110         full_path = self.PATH_PREFIX + path
  111         return self._request_json(path=full_path, params=params,
  112                                   expect_errors=expect_errors,
  113                                   headers=headers, extra_environ=extra_environ,
  114                                   status=status, method="post")
  115 
  116     def gen_context(self, value, **kwargs):
  117         ct = context.RequestContext.from_dict(value, **kwargs)
  118         return ct
  119 
  120     def gen_headers(self, context, **kw):
  121         """Generate a header for a simulated HTTP request to Pecan test app.
  122 
  123         :param context: context that store the client user information.
  124         :param kw: key word aguments, used to overwrite the context attribute.
  125 
  126         note: "is_public_api" is not in headers, it should be in environ
  127         variables to send along with the request. We can pass it by
  128         extra_environ when we call delete, get_json or other method request.
  129         """
  130         ct = context.to_dict()
  131         ct.update(kw)
  132         if ct.get("is_admin"):
  133             role = "admin"
  134         else:
  135             role = "user"
  136         headers = {
  137             'X-User-Name': ct.get("user_name") or "user",
  138             'X-User-Id':
  139                 ct.get("user_id") or "1d6d686bc2c949ddb685ffb4682e0047",
  140             'X-Project-Name': ct.get("project_name") or "no_project_name",
  141             'X-Project-Id':
  142                 ct.get("project_id") or "86f64f561b6d4f479655384572727f70",
  143             'X-User-Domain-Id':
  144                 ct.get("domain_id") or "bd5eeb7d0fb046daaf694b36f4df5518",
  145             'X-User-Domain-Name': ct.get("domain_name") or "no_domain",
  146             'X-Auth-Token':
  147                 ct.get("auth_token") or "b9764005b8c145bf972634fb16a826e8",
  148             'X-Roles': ct.get("roles") or role,
  149         }
  150         if ct.get('system_scope') == 'all':
  151             headers.update({'Openstack-System-Scope': 'all'})
  152         return headers
  153 
  154     def get_json(self, path, expect_errors=False, headers=None,
  155                  extra_environ=None, q=None, return_json=True, **params):
  156         """Sends simulated HTTP GET request to Pecan test app.
  157 
  158         :param path: url path of target service
  159         :param expect_errors: Boolean value; whether an error is expected based
  160                               on request
  161         :param headers: a dictionary of headers to send along with the request
  162         :param extra_environ: a dictionary of environ variables to send along
  163                               with the request
  164         :param q: list of queries consisting of: field, value, op, and type
  165                   keys
  166         :param path_prefix: prefix of the url path
  167         :param params: content for wsgi.input of request
  168         """
  169         full_path = self.PATH_PREFIX + path
  170         q = q or []
  171         query_params = {
  172             'q.field': [],
  173             'q.value': [],
  174             'q.op': [],
  175             }
  176         for query in q:
  177             for name in ['field', 'op', 'value']:
  178                 query_params['q.%s' % name].append(query.get(name, ''))
  179         all_params = {}
  180         all_params.update(params)
  181         if q:
  182             all_params.update(query_params)
  183         response = self.app.get(full_path,
  184                                 params=all_params,
  185                                 headers=headers,
  186                                 extra_environ=extra_environ,
  187                                 expect_errors=expect_errors)
  188         if return_json and not expect_errors:
  189             response = response.json
  190         return response
  191 
  192     def patch_json(self, path, params, expect_errors=False, headers=None,
  193                    extra_environ=None, status=None):
  194         """Sends simulated HTTP PATCH request to Pecan test app.
  195 
  196         :param path: url path of target service
  197         :param params: content for wsgi.input of request
  198         :param expect_errors: Boolean value; whether an error is expected based
  199                               on request
  200         :param headers: a dictionary of headers to send along with the request
  201         :param extra_environ: a dictionary of environ variables to send along
  202                               with the request
  203         :param status: expected status code of response
  204         """
  205         full_path = self.PATH_PREFIX + path
  206         return self._request_json(path=full_path, params=params,
  207                                   expect_errors=expect_errors,
  208                                   headers=headers, extra_environ=extra_environ,
  209                                   status=status, method="patch")
  210 
  211     def delete(self, path, expect_errors=False, headers=None,
  212                extra_environ=None, status=None):
  213         """Sends simulated HTTP DELETE request to Pecan test app.
  214 
  215         :param path: url path of target service
  216         :param expect_errors: Boolean value; whether an error is expected based
  217                               on request
  218         :param headers: a dictionary of headers to send along with the request
  219         :param extra_environ: a dictionary of environ variables to send along
  220                               with the request
  221         :param status: expected status code of response
  222         """
  223         full_path = self.PATH_PREFIX + path
  224         response = self.app.delete(full_path,
  225                                    headers=headers,
  226                                    status=status,
  227                                    extra_environ=extra_environ,
  228                                    expect_errors=expect_errors)
  229         return response