FM / unit

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit

Name	Size	Mode	Actions
accelerator/	-	0755	rm
baremetal/	-	0755	rm
baremetal_introspection/	-	0755	rm
block_storage/	-	0755	rm
cloud/	-	0755	rm
clustering/	-	0755	rm
compute/	-	0755	rm
config/	-	0755	rm
database/	-	0755	rm
dns/	-	0755	rm
fake/	-	0755	rm
fixtures/	-	0755	rm
identity/	-	0755	rm
image/	-	0755	rm
instance_ha/	-	0755	rm
key_manager/	-	0755	rm
load_balancer/	-	0755	rm
message/	-	0755	rm
network/	-	0755	rm
object_store/	-	0755	rm
orchestration/	-	0755	rm
workflow/	-	0755	rm
__pycache__/	-	0755	rm
base.py	32954	0644	edit dl rm
fakes.py	1273	0644	edit dl rm
test_connection.py	17450	0644	edit dl rm
test_exceptions.py	7885	0644	edit dl rm
test_format.py	1768	0644	edit dl rm
test_hacking.py	2729	0644	edit dl rm
test_microversions.py	3755	0644	edit dl rm
test_missing_version.py	1557	0644	edit dl rm
test_placement_rest.py	3718	0644	edit dl rm
test_proxy.py	22980	0644	edit dl rm
test_proxy_base.py	11982	0644	edit dl rm
test_proxy_base2.py	10883	0644	edit dl rm
test_resource.py	107968	0644	edit dl rm
test_stats.py	9248	0644	edit dl rm
test_utils.py	8464	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/test_placement_rest.py (3718B)

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import ddt
from keystoneauth1 import exceptions

from openstack.tests.unit import base

@ddt.ddt
class TestPlacementRest(base.TestCase):

def setUp(self):
        super(TestPlacementRest, self).setUp()
        self.use_placement()

def _register_uris(self, status_code=None):
        uri = dict(
            method='GET',
            uri=self.get_mock_url(
                'placement', 'public', append=['allocation_candidates']),
            json={})
        if status_code is not None:
            uri['status_code'] = status_code
        self.register_uris([uri])

def _validate_resp(self, resp, status_code):
        self.assertEqual(status_code, resp.status_code)
        self.assertEqual(
            'https://placement.example.com/allocation_candidates',
            resp.url)
        self.assert_calls()

@ddt.data({}, {'raise_exc': False}, {'raise_exc': True})
    def test_discovery(self, get_kwargs):
        self._register_uris()
        # Regardless of raise_exc, a <400 response doesn't raise
        rs = self.cloud.placement.get('/allocation_candidates', **get_kwargs)
        self._validate_resp(rs, 200)

@ddt.data({}, {'raise_exc': False})
    def test_discovery_err(self, get_kwargs):
        self._register_uris(status_code=500)
        # >=400 doesn't raise by default or with explicit raise_exc=False
        rs = self.cloud.placement.get('/allocation_candidates', **get_kwargs)
        self._validate_resp(rs, 500)

def test_discovery_exc(self):
        self._register_uris(status_code=500)
        # raise_exc=True raises a ksa exception appropriate to the status code
        ex = self.assertRaises(
            exceptions.InternalServerError,
            self.cloud.placement.get, '/allocation_candidates', raise_exc=True)
        self._validate_resp(ex.response, 500)

def test_microversion_discovery(self):
        self.assertEqual(
            (1, 17),
            self.cloud.placement.get_endpoint_data().max_microversion)
        self.assert_calls()

class TestBadPlacementRest(base.TestCase):

def setUp(self):
        super(TestBadPlacementRest, self).setUp()
        # The bad-placement.json is for older placement that was
        # missing the status field from its discovery doc. This
        # lets us show that we can talk to such a placement.
        self.use_placement(discovery_fixture='bad-placement.json')

def test_discovery(self):
        self._register_uris()
        rs = self.cloud.placement.get('/allocation_candidates')
        self._validate_resp(rs, 200)

# Copyright (c) 2018 Red Hat, Inc. # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import ddt from keystoneauth1 import exceptions from openstack.tests.unit import base @ddt.ddt class TestPlacementRest(base.TestCase): def setUp(self): super(TestPlacementRest, self).setUp() self.use_placement() def _register_uris(self, status_code=None): uri = dict( method='GET', uri=self.get_mock_url( 'placement', 'public', append=['allocation_candidates']), json={}) if status_code is not None: uri['status_code'] = status_code self.register_uris([uri]) def _validate_resp(self, resp, status_code): self.assertEqual(status_code, resp.status_code) self.assertEqual( 'https://placement.example.com/allocation_candidates', resp.url) self.assert_calls() @ddt.data({}, {'raise_exc': False}, {'raise_exc': True}) def test_discovery(self, get_kwargs): self._register_uris() # Regardless of raise_exc, a <400 response doesn't raise rs = self.cloud.placement.get('/allocation_candidates', **get_kwargs) self._validate_resp(rs, 200) @ddt.data({}, {'raise_exc': False}) def test_discovery_err(self, get_kwargs): self._register_uris(status_code=500) # >=400 doesn't raise by default or with explicit raise_exc=False rs = self.cloud.placement.get('/allocation_candidates', **get_kwargs) self._validate_resp(rs, 500) def test_discovery_exc(self): self._register_uris(status_code=500) # raise_exc=True raises a ksa exception appropriate to the status code ex = self.assertRaises( exceptions.InternalServerError, self.cloud.placement.get, '/allocation_candidates', raise_exc=True) self._validate_resp(ex.response, 500) def test_microversion_discovery(self): self.assertEqual( (1, 17), self.cloud.placement.get_endpoint_data().max_microversion) self.assert_calls() class TestBadPlacementRest(base.TestCase): def setUp(self): super(TestBadPlacementRest, self).setUp() # The bad-placement.json is for older placement that was # missing the status field from its discovery doc. This # lets us show that we can talk to such a placement. self.use_placement(discovery_fixture='bad-placement.json') def _register_uris(self, status_code=None): uri = dict( method='GET', uri=self.get_mock_url( 'placement', 'public', append=['allocation_candidates']), json={}) if status_code is not None: uri['status_code'] = status_code self.register_uris([uri]) def _validate_resp(self, resp, status_code): self.assertEqual(status_code, resp.status_code) self.assertEqual( 'https://placement.example.com/allocation_candidates', resp.url) self.assert_calls() def test_discovery(self): self._register_uris() rs = self.cloud.placement.get('/allocation_candidates') self._validate_resp(rs, 200)