| # Copyright 2016 Rackspace |
| # |
| # 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 argparse |
| from io import StringIO |
| import os |
| import shutil |
| import subprocess |
| import sys |
| import tempfile |
| from unittest import mock |
| from unittest.mock import patch |
| |
| |
| from oslo_serialization import jsonutils as json |
| from tempest.cmd import subunit_describe_calls |
| from tempest.tests import base |
| |
| |
| class TestArgumentParser(base.TestCase): |
| def test_init(self): |
| test_object = subunit_describe_calls.ArgumentParser() |
| self.assertEqual("subunit-describe-calls", test_object.prog) |
| self.assertEqual(subunit_describe_calls.DESCRIPTION, |
| test_object.description) |
| |
| |
| class TestUrlParser(base.TestCase): |
| services_custom_ports = { |
| "18776": "Block Storage", |
| "18774": "Nova", |
| "18773": "Nova-API", |
| "18386": "Sahara", |
| "35358": "Keystone", |
| "19292": "Glance", |
| "19696": "Neutron", |
| "16000": "Swift", |
| "18004": "Heat", |
| "18777": "Ceilometer", |
| "10080": "Horizon", |
| "18080": "Swift", |
| "1873": "rsync", |
| "13260": "iSCSI", |
| "13306": "MySQL", |
| "15672": "AMQP", |
| "18082": "murano"} |
| |
| def setUp(self): |
| super(TestUrlParser, self).setUp() |
| self.test_object = subunit_describe_calls.UrlParser() |
| |
| def test_get_service_default_ports(self): |
| base_url = "http://site.something.com:" |
| for port in self.test_object.services: |
| url = base_url + port + "/v2/action" |
| service = self.test_object.services[port] |
| self.assertEqual(service, self.test_object.get_service(url)) |
| |
| def test_get_service_custom_ports(self): |
| self.test_object = subunit_describe_calls.\ |
| UrlParser(services=self.services_custom_ports) |
| base_url = "http://site.something.com:" |
| for port in self.services_custom_ports: |
| url = base_url + port + "/v2/action" |
| service = self.services_custom_ports[port] |
| self.assertEqual(service, self.test_object.get_service(url)) |
| |
| def test_get_service_port_not_found(self): |
| url = "https://site.somewhere.com:1234/v2/action" |
| self.assertEqual("Unknown", self.test_object.get_service(url)) |
| self.assertEqual("Unknown", self.test_object.get_service("")) |
| |
| def test_parse_details_none(self): |
| self.assertIsNone(self.test_object.parse_details(None)) |
| |
| def test_url_path_ports(self): |
| uuid_sample1 = "3715e0bb-b1b3-4291-aa13-2c86c3b9ec93" |
| uuid_sample2 = "2715e0bb-b1b4-4291-aa13-2c86c3b9ec88" |
| |
| # test http url |
| host = "http://host.company.com" |
| url = host + ":8776/v3/" + uuid_sample1 + "/types/" + \ |
| uuid_sample2 + "/extra_specs" |
| self.assertEqual("v3/<uuid>/types/<uuid>/extra_specs", |
| self.test_object.url_path(url)) |
| url = host + ":8774/v2.1/servers/" + uuid_sample1 |
| self.assertEqual("v2.1/servers/<uuid>", |
| self.test_object.url_path(url)) |
| # test https url |
| host = "https://host.company.com" |
| url = host + ":8776/v3/" + uuid_sample1 + "/types/" + \ |
| uuid_sample2 + "/extra_specs" |
| self.assertEqual("v3/<uuid>/types/<uuid>/extra_specs", |
| self.test_object.url_path(url)) |
| url = host + ":8774/v2.1/servers/" + uuid_sample1 |
| self.assertEqual("v2.1/servers/<uuid>", |
| self.test_object.url_path(url)) |
| |
| def test_url_path_no_match(self): |
| host_port = 'https://host.company.com:1234/' |
| url = 'v2/action/no/special/data' |
| self.assertEqual(url, self.test_object.url_path(host_port + url)) |
| url = 'data' |
| self.assertEqual(url, self.test_object.url_path(url)) |
| |
| |
| class TestCliBase(base.TestCase): |
| """Base class for share code on all CLI sub-process testing""" |
| |
| def setUp(self): |
| super(TestCliBase, self).setUp() |
| self._subunit_file = os.path.join( |
| os.path.dirname(os.path.abspath(__file__)), |
| 'subunit_describe_calls_data', 'calls.subunit') |
| |
| def _bytes_to_string(self, data): |
| if isinstance(data, (bytes, bytearray)): |
| data = str(data, 'utf-8') |
| return data |
| |
| def _assert_cli_message(self, data): |
| data = self._bytes_to_string(data) |
| self.assertIn("Running subunit_describe_calls ...", data) |
| |
| def _assert_deprecated_warning(self, stdout): |
| self.assertIn( |
| b"Use of: 'subunit-describe-calls' is deprecated, " |
| b"please use: 'tempest subunit-describe-calls'", stdout) |
| |
| def _assert_expect_json(self, json_data): |
| expected_file_name = os.path.join( |
| os.path.dirname(os.path.abspath(__file__)), |
| 'subunit_describe_calls_data', 'calls_subunit_expected.json') |
| with open(expected_file_name, "rb") as read_file: |
| expected_result = json.load(read_file) |
| self.assertDictEqual(expected_result, json_data) |
| |
| def _assert_headers_and_bodies(self, data): |
| data = self._bytes_to_string(data) |
| self.assertIn('- request headers:', data) |
| self.assertIn('- request body:', data) |
| self.assertIn('- response headers:', data) |
| self.assertIn('- response body:', data) |
| |
| def _assert_methods_details(self, data): |
| data = self._bytes_to_string(data) |
| self.assertIn('foo', data) |
| self.assertIn('- 200 POST request for Nova to v2.1/<id>/', |
| data) |
| self.assertIn('- 200 DELETE request for Nova to v2.1/<id>/', |
| data) |
| self.assertIn('- 200 GET request for Nova to v2.1/<id>/', |
| data) |
| self.assertIn('- 404 DELETE request for Nova to v2.1/<id>/', |
| data) |
| |
| def _assert_mutual_exclusive_message(self, stderr): |
| self.assertIn(b"usage: subunit-describe-calls " |
| b"[-h] [-s [<subunit file>]]", stderr) |
| self.assertIn(b"[-n <non subunit name>] [-o <output file>]", |
| stderr) |
| self.assertIn(b"[-p <ports file>] [-v | -a]", stderr) |
| self.assertIn( |
| b"subunit-describe-calls: error: argument -v/--verbose: " |
| b"not allowed with argument -a/--all-stdout", stderr) |
| |
| def _assert_no_headers_and_bodies(self, data): |
| data = self._bytes_to_string(data) |
| self.assertNotIn('- request headers:', data) |
| self.assertNotIn('- request body:', data) |
| self.assertNotIn('- response headers:', data) |
| self.assertNotIn('- response body:', data) |
| |
| |
| class TestMainCli(TestCliBase): |
| """Test cases that use subunit_describe_calls module main interface |
| |
| via subprocess calls to make sure the total user experience |
| is well defined and tested. This interface is deprecated. |
| Note: these test do not affect code coverage percentages. |
| """ |
| |
| def test_main_output_file(self): |
| temp_file = tempfile.mkstemp()[1] |
| p = subprocess.Popen([ |
| 'subunit-describe-calls', '-s', self._subunit_file, |
| '-o', temp_file], stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_deprecated_warning(stdout) |
| with open(temp_file, 'r') as file: |
| data = json.loads(file.read()) |
| self._assert_expect_json(data) |
| |
| def test_main_verbose(self): |
| p = subprocess.Popen([ |
| 'subunit-describe-calls', '-s', self._subunit_file, |
| '-v'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_deprecated_warning(stdout) |
| self._assert_methods_details(stdout) |
| self._assert_headers_and_bodies(stdout) |
| |
| def test_main_all_stdout(self): |
| p = subprocess.Popen([ |
| 'subunit-describe-calls', '-s', self._subunit_file, |
| '--all-stdout'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_deprecated_warning(stdout) |
| self._assert_methods_details(stdout) |
| self._assert_headers_and_bodies(stdout) |
| |
| def test_main(self): |
| p = subprocess.Popen([ |
| 'subunit-describe-calls', '-s', self._subunit_file], |
| stdin=subprocess.PIPE, stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_deprecated_warning(stdout) |
| self._assert_methods_details(stdout) |
| self._assert_no_headers_and_bodies(stdout) |
| |
| def test_main_verbose_and_all_stdout(self): |
| p = subprocess.Popen([ |
| 'subunit-describe-calls', '-s', self._subunit_file, |
| '-a', '-v'], |
| stdin=subprocess.PIPE, stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(2, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_deprecated_warning(stdout) |
| self._assert_mutual_exclusive_message(stderr) |
| |
| |
| class TestCli(TestCliBase): |
| """Test cases that use tempest subunit_describe_calls cliff interface |
| |
| via subprocess calls to make sure the total user experience |
| is well defined and tested. |
| Note: these test do not affect code coverage percentages. |
| """ |
| |
| def _assert_cliff_verbose(self, stdout): |
| self.assertIn(b'tempest initialize_app', stdout) |
| self.assertIn(b'prepare_to_run_command TempestSubunitDescribeCalls', |
| stdout) |
| self.assertIn(b'tempest clean_up TempestSubunitDescribeCalls', |
| stdout) |
| |
| def test_run_all_stdout(self): |
| p = subprocess.Popen(['tempest', 'subunit-describe-calls', |
| '-s', self._subunit_file, '-a'], |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_methods_details(stdout) |
| self._assert_headers_and_bodies(stdout) |
| |
| def test_run_verbose(self): |
| p = subprocess.Popen(['tempest', 'subunit-describe-calls', |
| '-s', self._subunit_file, '-v'], |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_methods_details(stdout) |
| self._assert_no_headers_and_bodies(stdout) |
| self._assert_cliff_verbose(stderr) |
| |
| def test_run_min(self): |
| p = subprocess.Popen(['tempest', 'subunit-describe-calls', |
| '-s', self._subunit_file], |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_methods_details(stdout) |
| self._assert_no_headers_and_bodies(stdout) |
| |
| def test_run_verbose_all_stdout(self): |
| """Test Cliff -v argument |
| |
| Since Cliff framework has a argument at the |
| abstract command level the -v or --verbose for |
| this command is not processed as a boolean. |
| So the use of verbose only exists for the |
| deprecated main CLI interface. When the |
| main is deleted this test would not be needed. |
| """ |
| p = subprocess.Popen(['tempest', 'subunit-describe-calls', |
| '-s', self._subunit_file, '-a', '-v'], |
| stdin=subprocess.PIPE, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| stdout, stderr = p.communicate() |
| self.assertEqual(0, p.returncode) |
| self._assert_cli_message(stdout) |
| self._assert_cliff_verbose(stderr) |
| self._assert_methods_details(stdout) |
| |
| |
| class TestSubunitDescribeCalls(TestCliBase): |
| """Test cases use the subunit_describe_calls module interface |
| |
| and effect code coverage reporting |
| """ |
| |
| def setUp(self): |
| super(TestSubunitDescribeCalls, self).setUp() |
| self.test_object = subunit_describe_calls.TempestSubunitDescribeCalls( |
| app=mock.Mock(), |
| app_args=mock.Mock(spec=argparse.Namespace)) |
| |
| def test_parse(self): |
| with open(self._subunit_file, 'r') as read_file: |
| parser = subunit_describe_calls.parse( |
| read_file, "pythonlogging", None) |
| self._assert_expect_json(parser.test_logs) |
| |
| def test_get_description(self): |
| self.assertEqual(subunit_describe_calls.DESCRIPTION, |
| self.test_object.get_description()) |
| |
| def test_get_parser_default_min(self): |
| parser = self.test_object.get_parser('NAME') |
| parsed_args = parser.parse_args([]) |
| self.assertIsNone(parsed_args.output_file) |
| self.assertIsNone(parsed_args.ports) |
| self.assertFalse(parsed_args.all_stdout) |
| self.assertEqual(parsed_args.subunit, sys.stdin) |
| |
| def test_get_parser_default_max(self): |
| temp_dir = tempfile.mkdtemp(prefix="parser") |
| self.addCleanup(shutil.rmtree, temp_dir, ignore_errors=True) |
| outfile_name = os.path.join(temp_dir, 'output.json') |
| open(outfile_name, 'a').close() |
| portfile_name = os.path.join(temp_dir, 'ports.json') |
| open(portfile_name, 'a').close() |
| |
| parser = self.test_object.get_parser('NAME') |
| parsed_args = parser.parse_args(["-a", "-o " + outfile_name, |
| "-p " + portfile_name]) |
| |
| self.assertIsNotNone(parsed_args.output_file) |
| self.assertIsNotNone(parsed_args.ports) |
| self.assertTrue(parsed_args.all_stdout) |
| self.assertEqual(parsed_args.subunit, sys.stdin) |
| |
| def test_take_action_min(self): |
| parser = self.test_object.get_parser('NAME') |
| parsed_args = parser.parse_args(["-s" + self._subunit_file],) |
| with patch('sys.stdout', new=StringIO()) as mock_stdout: |
| self.test_object.take_action(parsed_args) |
| |
| stdout_data = mock_stdout.getvalue() |
| self._assert_methods_details(stdout_data) |
| self._assert_no_headers_and_bodies(stdout_data) |
| |
| def test_take_action_all_stdout(self): |
| parser = self.test_object.get_parser('NAME') |
| parsed_args = parser.parse_args(["-as" + self._subunit_file],) |
| with patch('sys.stdout', new=StringIO()) as mock_stdout: |
| self.test_object.take_action(parsed_args) |
| |
| stdout_data = mock_stdout.getvalue() |
| self._assert_methods_details(stdout_data) |
| self._assert_headers_and_bodies(stdout_data) |
| |
| def test_take_action_outfile_files(self): |
| temp_file = tempfile.mkstemp()[1] |
| parser = self.test_object.get_parser('NAME') |
| parsed_args = parser.parse_args( |
| ["-as" + self._subunit_file, '-o', temp_file], ) |
| with patch('sys.stdout', new=StringIO()) as mock_stdout: |
| self.test_object.take_action(parsed_args) |
| stdout_data = mock_stdout.getvalue() |
| self._assert_cli_message(stdout_data) |
| with open(temp_file, 'r') as file: |
| data = json.loads(file.read()) |
| self._assert_expect_json(data) |
| |
| def test_take_action_no_items(self): |
| temp_file = tempfile.mkstemp()[1] |
| parser = self.test_object.get_parser('NAME') |
| parsed_args = parser.parse_args( |
| ["-as" + temp_file], ) |
| with patch('sys.stdout', new=StringIO()) as mock_stdout: |
| self.test_object.take_action(parsed_args) |
| stdout_data = mock_stdout.getvalue() |
| self._assert_cli_message(stdout_data) |
| |
| def test_take_action_exception(self): |
| parser = self.test_object.get_parser('NAME') |
| parsed_args = parser.parse_args(["-s" + self._subunit_file],) |
| with patch('sys.stderr', new=StringIO()) as mock_stderr: |
| with patch('tempest.cmd.subunit_describe_calls.entry_point') \ |
| as mock_method: |
| mock_method.side_effect = OSError() |
| self.assertRaises(OSError, self.test_object.take_action, |
| parsed_args) |
| stderr_data = mock_stderr.getvalue() |
| |
| self.assertIn("Traceback (most recent call last):", stderr_data) |
| self.assertIn("entry_point(parsed_args)", stderr_data) |