@ -14,220 +14,422 @@
# 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 TestSubunitDescribeCalls ( base . TestCase ) :
def test_return_code ( self ) :
subunit_file = os . path . join (
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__ ) ) ,
' sample_streams/calls.subunit ' )
' 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 ' , subunit_file ,
' -o ' , tempfile . mkstemp ( ) [ 1 ] ] , stdin = subprocess . PIPE )
p . communicate ( )
' 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_verbose ( self ) :
subunit_file = os . path . join (
os . path . dirname ( os . path . abspath ( __file__ ) ) ,
' sample_streams/calls.subunit ' )
def test_main_verbose ( self ) :
p = subprocess . Popen ( [
' subunit-describe-calls ' , ' -s ' , subunit_file ,
' -v ' ] , stdin = subprocess . PIPE , stdout = subprocess . PIPE )
stdout = p . communicate ( )
' 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 . assertIn ( b ' - request headers: ' , stdout [ 0 ] )
self . assertIn ( b ' - request body: ' , stdout [ 0 ] )
self . assertIn ( b ' - response headers: ' , stdout [ 0 ] )
self . assertIn ( b ' - response body: ' , stdout [ 0 ] )
self . _assert_cli_message( stdout )
self . _assert_deprecated_warning( stdout )
self . _assert_methods_details( stdout )
self . _assert_headers_and_bodies( stdout )
def test_return_code_no_output ( self ) :
subunit_file = os . path . join (
os . path . dirname ( os . path . abspath ( __file__ ) ) ,
' sample_streams/calls.subunit ' )
def test_main_all_stdout ( self ) :
p = subprocess . Popen ( [
' subunit-describe-calls ' , ' -s ' , subunit_file ] ,
stdin = subprocess . PIPE , stdout = subprocess . PIPE )
stdout = p . communicate ( )
' 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 . assertIn ( b ' foo ' , stdout [ 0 ] )
self . assertIn ( b ' - 200 POST request for Nova to v2.1/<id>/ ' ,
stdout [ 0 ] )
self . assertIn ( b ' - 200 DELETE request for Nova to v2.1/<id>/ ' ,
stdout [ 0 ] )
self . assertIn ( b ' - 200 GET request for Nova to v2.1/<id>/ ' ,
stdout [ 0 ] )
self . assertIn ( b ' - 404 DELETE request for Nova to v2.1/<id>/ ' ,
stdout [ 0 ] )
self . assertNotIn ( b ' - request headers: ' , stdout [ 0 ] )
self . assertNotIn ( b ' - request body: ' , stdout [ 0 ] )
self . assertNotIn ( b ' - response headers: ' , stdout [ 0 ] )
self . assertNotIn ( b ' - response body: ' , stdout [ 0 ] )
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 ) :
subunit_file = os . path . join (
os . path . dirname ( os . path . abspath ( __file__ ) ) ,
' sample_streams/calls.subunit ' )
parser = subunit_describe_calls . parse (
open ( subunit_file ) , " pythonlogging " , None )
expected_result = {
' bar ' : [ {
' name ' : ' AgentsAdminTestJSON:setUp ' ,
' request_body ' : ' { " agent " : { " url " : " xxx://xxxx/xxx/xxx " , '
' " hypervisor " : " common " , " md5hash " : '
' " add6bb58e139be103324d04d82d8f545 " , " version " : " 7.0 " , '
' " architecture " : " tempest-x86_64-424013832 " , " os " : " linux " }} ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_body ' : ' { " agent " : { " url " : " xxx://xxxx/xxx/xxx " , '
' " hypervisor " : " common " , " md5hash " : '
' " add6bb58e139be103324d04d82d8f545 " , " version " : " 7.0 " , '
' " architecture " : " tempest-x86_64-424013832 " , " os " : " linux " , '
' " agent_id " : 1}} ' ,
' response_headers ' : " { ' status ' : ' 200 ' , ' content-length ' : "
" ' 203 ' , ' x-compute-request-id ' : "
" ' req-25ddaae2-0ef1-40d1-8228-59bd64a7e75b ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:00 GMT ' , ' content-type ' : "
" ' application/json ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 200 ' ,
' url ' : ' v2.1/<id>/os-agents ' ,
' verb ' : ' POST ' } , {
' name ' : ' AgentsAdminTestJSON:test_create_agent ' ,
' request_body ' : ' { " agent " : { " url " : " xxx://xxxx/xxx/xxx " , '
' " hypervisor " : " kvm " , " md5hash " : '
' " add6bb58e139be103324d04d82d8f545 " , " version " : " 7.0 " , '
' " architecture " : " tempest-x86-252246646 " , " os " : " win " }} ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_body ' : ' { " agent " : { " url " : " xxx://xxxx/xxx/xxx " , '
' " hypervisor " : " kvm " , " md5hash " : '
' " add6bb58e139be103324d04d82d8f545 " , " version " : " 7.0 " , '
' " architecture " : " tempest-x86-252246646 " , " os " : " win " , '
' " agent_id " : 2}} ' ,
' response_headers ' : " { ' status ' : ' 200 ' , ' content-length ' : "
" ' 195 ' , ' x-compute-request-id ' : "
" ' req-b4136f06-c015-4e7e-995f-c43831e3ecce ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:00 GMT ' , ' content-type ' : "
" ' application/json ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 200 ' ,
' url ' : ' v2.1/<id>/os-agents ' ,
' verb ' : ' POST ' } , {
' name ' : ' AgentsAdminTestJSON:tearDown ' ,
' request_body ' : ' None ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_body ' : ' ' ,
' response_headers ' : " { ' status ' : ' 200 ' , ' content-length ' : "
" ' 0 ' , ' x-compute-request-id ' : "
" ' req-ee905fd6-a5b5-4da4-8c37-5363cb25bd9d ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:00 GMT ' , ' content-type ' : "
" ' application/json ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 200 ' ,
' url ' : ' v2.1/<id>/os-agents/1 ' ,
' verb ' : ' DELETE ' } , {
' name ' : ' AgentsAdminTestJSON:_run_cleanups ' ,
' request_body ' : ' None ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_headers ' : " { ' status ' : ' 200 ' , ' content-length ' : "
" ' 0 ' , ' x-compute-request-id ' : "
" ' req-e912cac0-63e0-4679-a68a-b6d18ddca074 ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:00 GMT ' , ' content-type ' : "
" ' application/json ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 200 ' ,
' url ' : ' v2.1/<id>/os-agents/2 ' ,
' verb ' : ' DELETE ' } ] ,
' foo ' : [ {
' name ' : ' AgentsAdminTestJSON:setUp ' ,
' request_body ' : ' { " agent " : { " url " : " xxx://xxxx/xxx/xxx " , '
' " hypervisor " : " common " , " md5hash " : '
' " add6bb58e139be103324d04d82d8f545 " , " version " : " 7.0 " , '
' " architecture " : " tempest-x86_64-948635295 " , " os " : " linux " }} ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_body ' : ' { " agent " : { " url " : " xxx://xxxx/xxx/xxx " , '
' " hypervisor " : " common " , " md5hash " : '
' " add6bb58e139be103324d04d82d8f545 " , " version " : " 7.0 " , '
' " architecture " : " tempest-x86_64-948635295 " , " os " : " linux " , '
' " agent_id " : 3}} ' ,
' response_headers ' : " { ' status ' : ' 200 ' , ' content-length ' : "
" ' 203 ' , ' x-compute-request-id ' : "
" ' req-ccd2116d-04b1-4ffe-ae32-fb623f68bf1c ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:01 GMT ' , ' content-type ' : "
" ' application/json ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 200 ' ,
' url ' : ' v2.1/<id>/os-agents ' ,
' verb ' : ' POST ' } , {
' name ' : ' AgentsAdminTestJSON:test_delete_agent ' ,
' request_body ' : ' None ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_body ' : ' ' ,
' response_headers ' : " { ' status ' : ' 200 ' , ' content-length ' : "
" ' 0 ' , ' x-compute-request-id ' : "
" ' req-6e7fa28f-ae61-4388-9a78-947c58bc0588 ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:01 GMT ' , ' content-type ' : "
" ' application/json ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 200 ' ,
' url ' : ' v2.1/<id>/os-agents/3 ' ,
' verb ' : ' DELETE ' } , {
' name ' : ' AgentsAdminTestJSON:test_delete_agent ' ,
' request_body ' : ' None ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_body ' : ' { " agents " : []} ' ,
' response_headers ' : " { ' status ' : ' 200 ' , ' content-length ' : "
" ' 14 ' , ' content-location ' : "
" ' http://23.253.76.97:8774/v2.1/ "
" cf6b1933fe5b476fbbabb876f6d1b924/os-agents ' , "
" ' x-compute-request-id ' : "
" ' req-e41aa9b4-41a6-4138-ae04-220b768eb644 ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:01 GMT ' , ' content-type ' : "
" ' application/json ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 200 ' ,
' url ' : ' v2.1/<id>/os-agents ' ,
' verb ' : ' GET ' } , {
' name ' : ' AgentsAdminTestJSON:tearDown ' ,
' request_body ' : ' None ' ,
' request_headers ' : " { ' Content-Type ' : ' application/json ' , "
" ' Accept ' : ' application/json ' , ' X-Auth-Token ' : ' <omitted> ' } " ,
' response_headers ' : " { ' status ' : ' 404 ' , ' content-length ' : "
" ' 82 ' , ' x-compute-request-id ' : "
" ' req-e297aeea-91cf-4f26-b49c-8f46b1b7a926 ' , ' vary ' : "
" ' X-OpenStack-Nova-API-Version ' , ' connection ' : ' close ' , "
" ' x-openstack-nova-api-version ' : ' 2.1 ' , ' date ' : "
" ' Tue, 02 Feb 2016 03:27:02 GMT ' , ' content-type ' : "
" ' application/json; charset=UTF-8 ' } " ,
' service ' : ' Nova ' ,
' status_code ' : ' 404 ' ,
' url ' : ' v2.1/<id>/os-agents/3 ' ,
' verb ' : ' DELETE ' } ] }
self . assertEqual ( expected_result , parser . test_logs )
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 )
Doug Schveninger
3 years ago