+++ /dev/null
-[DEFAULT]
-test_command=OS_STDOUT_CAPTURE=${OS_STDOUT_CAPTURE:-1} \
- OS_STDERR_CAPTURE=${OS_STDERR_CAPTURE:-1} \
- OS_LOG_CAPTURE=${OS_LOG_CAPTURE:-1} \
- OS_TEST_TIMEOUT=${OS_TEST_TIMEOUT:-160} \
- ${PYTHON:-python} -m subunit.run discover -t ./ ${OS_TEST_PATH:-./conductor/tests/unit} $LISTOPT $IDOPTION
-test_id_option=--load-list $IDFILE
-test_list_option=--list
f_unit = None
f_value = None
- for key, value in attrib_value.iteritems():
+ for key, value in attrib_value.items():
if key == 'value':
f_value = value
elif key == 'unit':
req_attr_key, flavor_cfa)
if not flavor_feature_attr:
flavor_flag = False
- elif not self._compare_attribute(flavor_feature_attr[0],
+ elif not self._compare_attribute(list(flavor_feature_attr)[0],
req_feature_attr):
flavor_flag = False
if not flavor_flag:
extra_environment = {
'AUTH_TYPE': 'Basic',
- 'HTTP_AUTHORIZATION': 'Basic {}'.format(base64.encodestring('admin:default').strip())}
+ 'HTTP_AUTHORIZATION': 'Basic {}'.format(base64.encodestring('admin:default'.encode()).decode().strip())}
def setUp(self):
print("setup called ... ")
self.assertEqual(200, actual_response.status_int)
self.assertJsonEqual(expected_response,
- json.loads(actual_response.body))
+ json.loads(actual_response.body.decode()))
actual_response = self.app.get(url=url, expect_errors=True, extra_environ=self.extra_environment)
self.assertEqual(200, actual_response.status_int)
self.assertJsonEqual(expected_response,
- json.loads(actual_response.body))
+ json.loads(actual_response.body.decode()))
@mock.patch('conductor.api.adapters.aaf.aaf_authentication.authenticate')
@mock.patch('conductor.common.music.messaging.component.RPCClient.call')
a_arg.append(k)
for key in sorted(self.plan_expected):
b_arg.append(key)
- self.assertEquals(rtn.get('error'), False)
- self.assertEquals(a_arg, b_arg)
+ self.assertEqual(rtn.get('error'), False)
+ self.assertEqual(a_arg, b_arg)
for k in sorted(rtn.get('response').get('plan')):
a_arg.append(k)
for key in sorted(self.plan_expected.get('plan')):
b_arg.append(key)
- self.assertEquals(a_arg, b_arg)
+ self.assertEqual(a_arg, b_arg)
@patch('conductor.common.music.model.search.Query.all')
def test_plan_get_same_schema(self, mock_query):
mock_query.return_value = self.plan_mock
rtn_get = self.r.plans_get(self._cvx, _id)
plans = rtn_get.get('response').get('plans')
- self.assertEquals(plans, self.the_plan_expected)
+ self.assertEqual(plans, self.the_plan_expected)
self.assertFalse(rtn_get.get('error'))
@patch('conductor.common.music.model.search.Query.all')
_id = {}
mock_call.return_value = self.plan_mock
rtn = self.r.plans_delete(self._cvx, _id)
- self.assertEquals(rtn.get('response'), {})
+ self.assertEqual(rtn.get('response'), {})
self.assertFalse(rtn.get('error'))
def tearDown(self):
self.Translator._locations, "locations")
location = {'customer_loc': {
'latitude': 32.89748, 'longitude': -97.040443}}
- self.assertEquals(rtn, location)
+ self.assertEqual(rtn, location)
@patch('conductor.common.music.messaging.component.RPCClient.call')
def test_parse_locations(self, mock_call):
}
mock_call.return_value = {'resolved_location': {
'latitude': 32.89748, 'longitude': -97.040443}}
- self.assertEquals(
+ self.assertEqual(
self.Translator.parse_locations(locations), locations)
def test_parse_error_format_demands(self):
'inventory_type': 'service',
'service_type': '5G'}]}}
- self.assertEquals(self.Translator.parse_demands(demands), rtn)
+ self.assertEqual(self.Translator.parse_demands(demands), rtn)
@patch('conductor.common.music.messaging.component.RPCClient.call')
def test_parse_demands_inventory_type_vfmodule(self, mock_call):
}
}
- self.assertEquals(self.Translator.parse_demands(demands), rtn)
+ self.assertEqual(self.Translator.parse_demands(demands), rtn)
@patch('conductor.common.music.messaging.component.RPCClient.call')
def test_parse_demands_without_candidate(self, mock_call):
}],
'service_type': '5G'}]}}
- self.assertEquals(self.Translator.parse_demands(demands), rtn)
+ self.assertEqual(self.Translator.parse_demands(demands), rtn)
def test_parse_constraints(self):
constraints = {'constraint_loc': {
'value': 100.0},
'location': 'custom_loc'},
'type': 'distance_to_location'}}
- self.assertEquals(self.Translator.parse_constraints(constraints), rtn)
+ self.assertEqual(self.Translator.parse_constraints(constraints), rtn)
def test_parse_hpa_constraints(self):
hpa_constraint = {
}
}
- self.assertEquals(self.Translator.parse_constraints(hpa_constraint),
+ self.assertEqual(self.Translator.parse_constraints(hpa_constraint),
rtn)
hpa_constraint_2 = {
}
}
- self.assertEquals(self.Translator.parse_constraints(hpa_constraint_2),
+ self.assertEqual(self.Translator.parse_constraints(hpa_constraint_2),
rtn_2)
def test_parse_hpa_constraints_format_validation(self):
}
}
self.maxDiff = None
- self.assertEquals(expected_response, self.Translator.parse_constraints(
+ self.assertEqual(expected_response, self.Translator.parse_constraints(
vim_fit_constraint))
self.assertRaises(TranslatorException,
self.Translator.parse_constraints,
self.Translator._locations = {'vG': '',
'vGMuxInfra': '',
'customer_loc': ''}
- self.assertEquals(
+ self.assertEqual(
self.Translator.parse_optimization(
opt), expected_parse)
'customer_loc': ''}
self.Translator._constraints = hpa_json["HAS_Template"]["constraints"]
self.maxDiff = None
- self.assertEquals(
+ self.assertEqual(
self.Translator.parse_optimization(
opt), expected_parse)
}
}
}
- self.assertEquals(
+ self.assertEqual(
self.Translator.parse_reservations(resv), expected_resv)
@patch('conductor.controller.translator.Translator.parse_constraints')
mock_opt.return_value = {}
mock_cons.return_value = {}
self.Translator.do_translation()
- self.assertEquals(self.Translator._translation, expected_format)
+ self.assertEqual(self.Translator._translation, expected_format)
@patch('conductor.controller.translator.Translator.create_components')
@patch('conductor.controller.translator.Translator.validate_components')
def test_translate(self, mock_parse, mock_do_trans,
mock_valid, mock_create):
self.Translator.translate()
- self.assertEquals(self.Translator._ok, True)
+ self.assertEqual(self.Translator._ok, True)
def tearDown(self):
patch.stopall()
mock_ok.return_value = True
mock_ok_func.return_value = True
self.translator_svc.translate(self.mock_plan)
- self.assertEquals(self.mock_plan.status, 'translated')
+ self.assertEqual(self.mock_plan.status, 'translated')
@patch('conductor.controller.translator.Translator.translate')
mock_ok.return_value = False
mock_error.return_value = 'error'
self.translator_svc.translate(self.mock_plan)
- self.assertEquals(self.mock_plan.status, 'error')
+ self.assertEqual(self.mock_plan.status, 'error')
def test_millisec_to_sec(self):
- self.assertEquals(self.translator_svc.millisec_to_sec(1000), 1)
+ self.assertEqual(self.translator_svc.millisec_to_sec(1000), 1)
def test_current_time_seconds(self):
- self.assertEquals(self.translator_svc.current_time_seconds(),
+ self.assertEqual(self.translator_svc.current_time_seconds(),
int(round(time.time())))
@patch('conductor.common.music.model.base.Base.insert')
req_json = yaml.safe_load(open(req_json_file).read())
candidate_list = req_json['candidate_list']
(constraint_id, constraint_info) = \
- hpa_json["conductor_solver"]["constraints"][0].items()[0]
+ list(hpa_json["conductor_solver"]["constraints"][0].items())[0]
hpa_constraint = constraint_info['properties']
flavorProperties = hpa_constraint['evaluate'][0]['flavorProperties']
id = hpa_constraint['evaluate'][0]['id']
candidate_list = req_json['candidate_list']
ext_mock1.return_value = ['MultiCloud']
(constraint_id, constraint_info) = \
- hpa_json["conductor_solver"]["constraints"][2].items()[0]
+ list(hpa_json["conductor_solver"]["constraints"][2].items())[0]
vim_request = constraint_info['properties']['request']
ctxt = {}
candidate_list_copy = list(copy.deepcopy(candidate_list))
response.ok = True
response.text = '12345678'
rest_mock.return_value = response
- self.assertEquals('12345678', self.music_api._lock_id_create('temp'))
+ self.assertEqual('12345678', self.music_api._lock_id_create('temp'))
self.mock_lock_id.start()
@mock.patch('conductor.common.rest.REST.request')
response.ok = True
response.text = 'true'
rest_mock.return_value = response
- self.assertEquals(True, self.music_api._lock_id_acquire('12345678'))
+ self.assertEqual(True, self.music_api._lock_id_acquire('12345678'))
self.mock_lock_acquire.start()
@mock.patch('conductor.common.rest.REST.request')
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api._lock_id_release('12345678'))
+ self.assertEqual(True, self.music_api._lock_id_release('12345678'))
self.mock_lock_release.start()
def test_lock_name_generate(self):
def test_lock_create(self):
expected = 'keyspace.votecount.pk_value'
- self.assertEquals(expected, self.music_api.lock_create('keyspace',
+ self.assertEqual(expected, self.music_api.lock_create('keyspace',
'votecount',
'pk_value'))
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.lock_release('test-lock-name'))
+ self.assertEqual(True, self.music_api.lock_release('test-lock-name'))
self.mock_lock_release.start()
@mock.patch('conductor.common.rest.REST.request')
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.lock_delete('test-lock-name'))
+ self.assertEqual(True, self.music_api.lock_delete('test-lock-name'))
@mock.patch('conductor.common.rest.REST.request')
def test_keyspace_create(self, rest_mock):
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.keyspace_create('keyspace'))
+ self.assertEqual(True, self.music_api.keyspace_create('keyspace'))
@mock.patch('conductor.common.rest.REST.request')
def test_keyspace_delete(self, rest_mock):
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.keyspace_delete('keyspace'))
+ self.assertEqual(True, self.music_api.keyspace_delete('keyspace'))
@mock.patch('conductor.common.rest.REST.request')
def test_row_create(self, rest_mock):
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.row_create(**kwargs))
+ self.assertEqual(True, self.music_api.row_create(**kwargs))
@mock.patch('conductor.common.rest.REST.request')
# Following changes made by 'ikram'.
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.row_update(**kwargs))
+ self.assertEqual(True, self.music_api.row_update(**kwargs))
@mock.patch('conductor.common.rest.REST.request')
def test_row_read(self, rest_mock):
response.status_code = 200
response.json.return_value = {'row 1': {'count': 2}}
rest_mock.return_value = response
- self.assertEquals({'row 1': {'count': 2}},
+ self.assertEqual({'row 1': {'count': 2}},
self.music_api.row_read(**kwargs))
@mock.patch('conductor.common.rest.REST.request')
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.row_delete(**kwargs))
+ self.assertEqual(True, self.music_api.row_delete(**kwargs))
def test_table_path_generate(self):
keyspace = 'test-keyspace'
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.table_create(**kwargs))
+ self.assertEqual(True, self.music_api.table_create(**kwargs))
@mock.patch('conductor.common.rest.REST.request')
def test_table_delete(self, rest_mock):
response.status_code = 200
response.ok = True
rest_mock.return_value = response
- self.assertEquals(True, self.music_api.table_delete(**kwargs))
+ self.assertEqual(True, self.music_api.table_delete(**kwargs))
def test_version(self):
with mock.patch.object(rest.REST, 'request',
response.status = 200
response.text = 'MUSIC:2.2.14'
rest_mock.return_value = response
- self.assertEquals('MUSIC:2.2.14', self.music_api.version())
+ self.assertEqual('MUSIC:2.2.14', self.music_api.version())
def test_row_url_path(self):
keyspace = 'test-keyspace'
req_json = yaml.safe_load(open(req_json_file).read())
(constraint_id, constraint_info) = \
- hpa_json["conductor_solver"]["constraints"][0].items()[0]
+ list(hpa_json["conductor_solver"]["constraints"][0].items())[0]
c_property = constraint_info['properties']
constraint_type = constraint_info['properties']
constraint_demands = list()
plans=plans,
order_lock=order_lock_inst)
- self.assertEquals(actual_response, {plan_id: OrderLock.COMPLETED})
+ self.assertEqual(actual_response, {plan_id: OrderLock.COMPLETED})
def test_update_order_status_and_get_effected_plans(self):
actual_response = self.order_lock_svc.update_order_status_and_get_effected_plans(rehome_status=rehome_status,
service_resource_id='resource-id')
- self.assertEquals(actual_response, plans)
+ self.assertEqual(actual_response, plans)
def test_rehomes_for_service_resource(self):
actual_response = self.order_lock_svc.rehomes_for_service_resource(rehome_status, 'resource-id', list())
expect_response = [{'plan_id': plan_id, 'should_rehome': True}]
- self.assertEquals(actual_response, expect_response)
+ self.assertEqual(actual_response, expect_response)
def test_get_plans_by_id(self):
self.order_lock_svc.OrderLock.query.get_plan_by_col = mock.MagicMock(return_value=order_locks)
actual_response = self.order_lock_svc._get_plans_by_id('order_id')
- self.assertEquals(actual_response, plans)
+ self.assertEqual(actual_response, plans)
if __name__ == '__main__':
req_json = yaml.safe_load(open(req_json_file).read())
(constraint_id, constraint_info) = \
- hpa_json["conductor_solver"]["constraints"][2].items()[0]
+ list(hpa_json["conductor_solver"]["constraints"][2].items())[0]
c_property = constraint_info['properties']
constraint_type = constraint_info['properties']
constraint_demands = list()
a.sort()
b.sort()
- self.assertEquals(a, b)
+ self.assertEqual(a, b)
def tearDown(self):
pass
# process, which may cause wedges in the gate later.
# Hacking already pins down pep8, pyflakes and flake8
-hacking<0.11,>=0.10.0
+hacking>=2.0.0
# bandit>=1.1.0 # Apache-2.0
+pytest>=5.4.1
coverage>=3.6 # Apache-2.0
fixtures>=3.0.0 # Apache-2.0/BSD
kombu>=3.0.25 # BSD
+++ /dev/null
-#!/usr/bin/env bash
-#
-# -------------------------------------------------------------------------
-# Copyright (c) 2015-2017 AT&T Intellectual Property
-#
-# 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.
-#
-# -------------------------------------------------------------------------
-#
-
-
-set -o pipefail
-
-TESTRARGS=$1
-
-# --until-failure is not compatible with --subunit see:
-#
-# https://bugs.launchpad.net/testrepository/+bug/1411804
-#
-# this work around exists until that is addressed
-if [[ "$TESTARGS" =~ "until-failure" ]]; then
- python setup.py testr --slowest --testr-args="$TESTRARGS" | subunit2junitxml --output-to=xunit-results.xml
-else
- python setup.py testr --slowest --testr-args="--subunit $TESTRARGS" | subunit-1to2 | subunit2junitxml --no-passthrough --forward --output-to=xunit-results.xml | subunit-trace -f
-fi
minversion = 1.6
skipsdist = True
#envlist = py35,py27,functional,pep8
-envlist = py27,pep8
+envlist = py3,pep8
[testenv]
+basepython=python3
deps = -r{toxinidir}/requirements.txt
-r{toxinidir}/test-requirements.txt
install_command = pip install -U {opts} {packages}
OS_TEST_PATH=conductor/tests/unit
passenv = OS_TEST_TIMEOUT OS_STDOUT_CAPTURE OS_STDERR_CAPTURE OS_LOG_CAPTURE
commands =
- {toxinidir}/tools/pretty_tox.sh "{posargs}"
+ coverage run --module pytest --junitxml xunit-results.xml
oslo-config-generator --config-file=etc/conductor/conductor-config-generator.conf
whitelist_externals = bash
find
coverage erase
find . -type f -name "*.pyc" -delete
find . -type f -name ".coverage.*" -delete
- {toxinidir}/tools/pretty_tox.sh "{posargs}"
- coverage combine
+ coverage run --module pytest --junitxml xunit-results.xml
coverage html -d cover
coverage xml -o cover/coverage.xml
- coverage report --omit=".tox/py27/*","conductor/tests/*"
+ coverage report --omit=".tox/*","conductor/tests/*"
[testenv:pep8]
-deps = hacking<0.12,>=0.11.0
+deps = hacking>=2.0.0
commands =
- flake8
+ flake8 --exit-zero
[testenv:genconfig]
commands = oslo-config-generator --config-file=etc/conductor/conductor-config-generator.conf
Code Review / optf / has.git / commitdiff