openstack/tacker
Code Issues Proposed changes

Merge "Remove test helpers for Python2"

Browse Source
This commit is contained in:
Zuul 2020-11-03 18:02:57 +00:00 committed by Gerrit Code Review
commit 49114276c2
5 changed files with 15 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
tacker/tests/common/helpers.py
View File

@ -1,35 +0,0 @@
# Copyright 2015 Red Hat, Inc.
# Copyright (c) 2014-2018 China Mobile (SuZhou) Software Technology Co.,Ltd.
# All Rights Reserved
#
# 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 six
import sys
import testtools
def requires_py2(testcase):
return testtools.skipUnless(six.PY2, "requires python 2.x")(testcase)
def requires_py3(testcase):
return testtools.skipUnless(six.PY3, "requires python 3.x")(testcase)
if sys.version_info < (3,):
def compact_byte(x):
return x
else:
def compact_byte(x):
return bytes(x, 'utf-8')

9
tacker/tests/unit/agent/linux/test_utils.py
View File

@ -19,7 +19,6 @@ import oslo_i18n
from tacker.agent.linux import utils
from tacker.tests import base
from tacker.tests.common import helpers
_marker = object()
@ -123,8 +122,8 @@ class AgentUtilsExecuteTest(base.BaseTestCase):
self.assertTrue(log.error.called)
def test_encode_process_input(self):
bytes_idata = helpers.compact_byte("%s\n" % self.test_file[:-1])
bytes_odata = helpers.compact_byte("%s\n" % self.test_file)
bytes_idata = bytes("%s\n" % self.test_file[:-1], 'utf-8')
bytes_odata = bytes("%s\n" % self.test_file, 'utf-8')
self.mock_popen.return_value = [bytes_odata, b'']
result = utils.execute(['cat'], process_input=bytes_idata)
self.mock_popen.assert_called_once_with(bytes_idata)
@ -144,6 +143,6 @@ class AgentUtilsExecuteEncodeTest(base.BaseTestCase):
open(self.test_file, 'w').close()
def test_decode_return_data(self):
str_data = helpers.compact_byte("%s\n" % self.test_file)
str_data = bytes("%s\n" % self.test_file, 'utf-8')
result = utils.execute(['ls', self.test_file], return_stderr=True)
self.assertEqual((str_data, helpers.compact_byte('')), result)
self.assertEqual((str_data, bytes('', 'utf-8')), result)

3
tacker/tests/unit/test_alarm_receiver.py
View File

@ -19,7 +19,6 @@ from oslo_serialization import jsonutils
from webob import Request
from tacker.alarm_receiver import AlarmReceiver
from tacker.tests.common import helpers
from tacker.tests.unit import base
@ -72,7 +71,7 @@ class TestAlarmReceiver(base.TestCase):
req.method = 'POST'
self.alarmrc.process_request(req)
self.assertEqual(helpers.compact_byte(''), req.body)
self.assertEqual(bytes('', 'utf-8'), req.body)
self.assertEqual('fake_token', req.headers['X_AUTH_TOKEN'])
self.assertIn(self.alarm_url['01_url_base'], req.environ['PATH_INFO'])
self.assertIn('triggers', req.environ['PATH_INFO'])

9
tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack.py
View File

@ -23,7 +23,6 @@ from oslo_serialization import jsonutils
from tacker import context
from tacker.db.common_services import common_services_db_plugin
from tacker.extensions import vnfm
from tacker.tests.common import helpers
from tacker.tests.unit import base
from tacker.tests.unit.db import utils
from tacker.vnfm.infra_drivers.openstack import openstack
@ -324,7 +323,7 @@ class TestOpenStack(base.TestCase):
if is_monitor:
if multi_vdus:
dvc['attributes'].update(
{'monitoring_policy': helpers.compact_byte(
{'monitoring_policy': bytes(
'{"vdus": {"VDU1": {"ping": '
'{"name": "ping", "actions": '
'{"failure": "respawn"}, '
@ -337,10 +336,10 @@ class TestOpenStack(base.TestCase):
'"parameters": {"count": 3, '
'"interval": 10}, '
'"monitoring_params": {"count": 3, '
'"interval": 10}}}}}')})
'"interval": 10}}}}}', 'utf-8')})
else:
dvc['attributes'].update(
{'monitoring_policy': helpers.compact_byte(
{'monitoring_policy': bytes(
'{"vdus": {"VDU1": {"ping": '
'{"name": "ping", "actions": '
'{"failure": "respawn"}, '
@ -348,7 +347,7 @@ class TestOpenStack(base.TestCase):
'"interval": 10}, '
'"monitoring_params": '
'{"count": 3, '
'"interval": 10}}}}}')})
'"interval": 10}}}}}', 'utf-8')})
return dvc

15
tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack_driver.py
View File

@ -31,7 +31,6 @@ from tacker import context
from tacker.db.db_sqlalchemy import models
from tacker.extensions import vnfm
from tacker import objects
from tacker.tests.common import helpers
from tacker.tests import constants
from tacker.tests.unit import base
from tacker.tests.unit.db import utils
@ -790,7 +789,7 @@ class TestOpenStack(base.FixturedTestCase):
vnf_dict, self.instance_uuid, None)
self.mock_log.debug.assert_called_with('outputs %s',
fd_utils.get_dummy_stack()['outputs'])
self.assertEqual(helpers.compact_byte('{"VDU1": "192.168.120.216"}'),
self.assertEqual(bytes('{"VDU1": "192.168.120.216"}', 'utf-8'),
vnf_dict['mgmt_ip_address'])
def test_create_wait_without_mgmt_ips(self):
@ -817,7 +816,7 @@ class TestOpenStack(base.FixturedTestCase):
vnf_dict = utils.get_dummy_vnf(scaling_group=True)
self.openstack.create_wait(None, None, vnf_dict, self.instance_uuid,
None)
self.assertEqual(helpers.compact_byte('{"vdu1": ["test1"]}'),
self.assertEqual(bytes('{"vdu1": ["test1"]}', 'utf-8'),
vnf_dict['mgmt_ip_address'])
def test_create_wait_failed_with_stack_retries_0(self):
@ -869,7 +868,7 @@ class TestOpenStack(base.FixturedTestCase):
self.openstack.update_wait(None, None, vnf_dict, None)
self.mock_log.debug.assert_called_with('outputs %s',
fd_utils.get_dummy_stack()['outputs'])
self.assertEqual(helpers.compact_byte('{"VDU1": "192.168.120.216"}'),
self.assertEqual(bytes('{"VDU1": "192.168.120.216"}', 'utf-8'),
vnf_dict['mgmt_ip_address'])
def test_heal_wait(self):
@ -880,7 +879,7 @@ class TestOpenStack(base.FixturedTestCase):
self.openstack.heal_wait(None, None, vnf_dict, None)
self.mock_log.debug.assert_called_with('outputs %s',
fd_utils.get_dummy_stack()['outputs'])
self.assertEqual(helpers.compact_byte('{"VDU1": "192.168.120.216"}'),
self.assertEqual(bytes('{"VDU1": "192.168.120.216"}', 'utf-8'),
vnf_dict['mgmt_ip_address'])
def test_heal_wait_without_mgmt_ips(self):
@ -979,8 +978,7 @@ class TestOpenStack(base.FixturedTestCase):
region_name=None,
last_event_id=uuidsentinel.
non_last_event_id)
self.assertEqual(helpers.compact_byte('{"vdu1": ["test1"]}'),
mgmt_ip)
self.assertEqual(bytes('{"vdu1": ["test1"]}', 'utf-8'), mgmt_ip)
@mock.patch.object(hc.HeatClient, "resource_event_list")
@ddt.data("SIGNAL_COMPLETE", "CREATE_COMPLETE")
@ -997,8 +995,7 @@ class TestOpenStack(base.FixturedTestCase):
policy=fd_utils.get_dummy_policy_dict(),
region_name=None,
last_event_id=fd_utils.get_dummy_event()['id'])
self.assertEqual(helpers.compact_byte('{"vdu1": ["test1"]}'),
mgmt_ip)
self.assertEqual(bytes('{"vdu1": ["test1"]}', 'utf-8'), mgmt_ip)
@mock.patch('tacker.vnfm.infra_drivers.openstack.openstack.LOG')
def test_scale_wait_failed_with_exception(self, mock_log):