<s> import collections <EOL> import itertools <EOL> import mock <EOL> from sahara . service . validations . edp import job as j <EOL> from sahara . service . validations . edp import job_execution_schema as j_e_schema <EOL> from sahara . service . validations . edp import job_interface as j_i <EOL> from sahara . service . validations . edp import job_schema as j_schema <EOL> from sahara . tests . unit . service . validation import utils as u <EOL> from sahara . utils import edp <EOL> def _configs ( ** kwargs ) : <EOL> arg = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:location>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT:default>" : "<STR_LIT:1>" <EOL> } <EOL> arg . update ( kwargs ) <EOL> return arg <EOL> def _params ( ** kwargs ) : <EOL> arg = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:location>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT:default>" : "<STR_LIT>" <EOL> } <EOL> arg . update ( kwargs ) <EOL> return arg <EOL> def _args ( ** kwargs ) : <EOL> arg = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:args>" , <EOL> "<STR_LIT:location>" : "<STR_LIT:0>" , <EOL> "<STR_LIT>" : "<STR_LIT:string>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT:default>" : "<STR_LIT>" <EOL> } <EOL> arg . update ( kwargs ) <EOL> return arg <EOL> _mapping_types = { "<STR_LIT>" , "<STR_LIT:args>" , "<STR_LIT>" } <EOL> def _job ( job_type , interface ) : <EOL> return { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT:type>" : job_type , "<STR_LIT>" : interface } <EOL> _job_types = [ <EOL> _job ( edp . JOB_TYPE_HIVE , [ _configs ( ) , _params ( ) ] ) , <EOL> _job ( edp . JOB_TYPE_PIG , [ _configs ( ) , _params ( ) , _args ( ) ] ) , <EOL> _job ( edp . JOB_TYPE_MAPREDUCE , [ _configs ( ) ] ) , <EOL> _job ( edp . JOB_TYPE_MAPREDUCE_STREAMING , [ _configs ( ) ] ) , <EOL> _job ( edp . JOB_TYPE_JAVA , [ _configs ( ) , _args ( ) ] ) , <EOL> _job ( edp . JOB_TYPE_SHELL , [ _configs ( ) , _params ( ) , _args ( ) ] ) , <EOL> _job ( edp . JOB_TYPE_SPARK , [ _configs ( ) , _args ( ) ] ) , <EOL> _job ( edp . JOB_TYPE_STORM , [ _args ( ) ] ) <EOL> ] <EOL> class TestJobInterfaceValidation ( u . ValidationTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestJobInterfaceValidation , self ) . setUp ( ) <EOL> self . _create_object_fun = j . check_interface <EOL> self . scheme = j_schema . JOB_SCHEMA <EOL> def test_interface ( self ) : <EOL> for job in _job_types : <EOL> self . _assert_create_object_validation ( data = job ) <EOL> def test_no_interface ( self ) : <EOL> job = _job ( edp . JOB_TYPE_PIG , None ) <EOL> self . _assert_create_object_validation ( data = job ) <EOL> def test_overlapping_names ( self ) : <EOL> job = _job ( edp . JOB_TYPE_PIG , <EOL> [ _configs ( ) , _configs ( location = "<STR_LIT>" ) ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> def test_unacceptable_types ( self ) : <EOL> for job in _job_types : <EOL> acceptable_types = [ arg [ '<STR_LIT>' ] <EOL> for arg in job [ '<STR_LIT>' ] ] <EOL> unacceptable_types = _mapping_types - set ( acceptable_types ) <EOL> unacceptable_args = ( globals ( ) [ "<STR_LIT:_>" + m_type ] ( ) <EOL> for m_type in unacceptable_types ) <EOL> bad_job = job . copy ( ) <EOL> for arg in unacceptable_args : <EOL> bad_job [ '<STR_LIT>' ] = [ arg ] <EOL> permutations = itertools . permutations ( acceptable_types ) <EOL> msgs = [ "<STR_LIT>" % <EOL> ( list ( permutation ) , bad_job [ '<STR_LIT:type>' ] ) <EOL> for permutation in permutations ] <EOL> self . _assert_create_object_validation ( <EOL> data = bad_job , bad_req_i = ( <NUM_LIT:1> , "<STR_LIT>" , msgs ) ) <EOL> def test_bad_positional_arg_locations ( self ) : <EOL> job = _job ( edp . JOB_TYPE_PIG , [ _args ( location = "<STR_LIT:1>" ) ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> job = _job ( edp . JOB_TYPE_PIG , [ _args ( location = "<STR_LIT>" ) ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> job = _job ( edp . JOB_TYPE_PIG , <EOL> [ _args ( ) , _args ( location = "<STR_LIT:2>" , name = "<STR_LIT>" ) ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_required_positional_arg_without_default ( self ) : <EOL> arg = _args ( required = False ) <EOL> del arg [ '<STR_LIT:default>' ] <EOL> job = _job ( edp . JOB_TYPE_PIG , [ arg ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_overlapping_locations ( self ) : <EOL> job = _job ( edp . JOB_TYPE_PIG , <EOL> [ _configs ( ) , _configs ( name = "<STR_LIT>" ) ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_number_values ( self ) : <EOL> job = _job ( edp . JOB_TYPE_PIG , [ _configs ( default = "<STR_LIT>" ) ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_data_source_values ( self , data_source ) : <EOL> data_source . return_value = True <EOL> job = _job ( edp . JOB_TYPE_PIG , <EOL> [ _params ( default = "<STR_LIT>" ) ] ) <EOL> self . _assert_create_object_validation ( data = job ) <EOL> data_source . return_value = False <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> job = _job ( edp . JOB_TYPE_PIG , [ _params ( default = "<STR_LIT>" ) ] ) <EOL> self . _assert_create_object_validation ( <EOL> data = job , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_default_data_type ( self ) : <EOL> param = _params ( ) <EOL> del param [ '<STR_LIT>' ] <EOL> job = _job ( edp . JOB_TYPE_PIG , [ param ] ) <EOL> self . _assert_create_object_validation ( data = job ) <EOL> assert param [ '<STR_LIT>' ] == j_i . DEFAULT_DATA_TYPE <EOL> int_arg = collections . namedtuple ( "<STR_LIT>" , <EOL> [ "<STR_LIT:name>" , "<STR_LIT>" , "<STR_LIT:location>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT:default>" ] ) <EOL> def j_e_i_wrapper ( data ) : <EOL> job = mock . Mock ( <EOL> interface = [ int_arg ( ** _configs ( ) ) , <EOL> int_arg ( ** _args ( ) ) , <EOL> int_arg ( ** _params ( ) ) ] <EOL> ) <EOL> j_i . check_execution_interface ( data , job ) <EOL> class TestJobExecutionInterfaceValidation ( u . ValidationTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestJobExecutionInterfaceValidation , self ) . setUp ( ) <EOL> self . _create_object_fun = j_e_i_wrapper <EOL> self . scheme = j_e_schema . JOB_EXEC_SCHEMA <EOL> def test_valid_execution ( self ) : <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT:2>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:value>" } } <EOL> self . _assert_create_object_validation ( data = data ) <EOL> def test_no_execution_interface ( self ) : <EOL> data = { "<STR_LIT>" : "<STR_LIT>" } <EOL> self . _assert_create_object_validation ( data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_bad_argument_name ( self ) : <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT:2>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> self . _assert_create_object_validation ( data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_required_argument_missing ( self ) : <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> self . _assert_create_object_validation ( data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_bad_values ( self , data_source ) : <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT>" } } <EOL> self . _assert_create_object_validation ( data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT:2>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> self . _assert_create_object_validation ( data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT:2>" , <EOL> "<STR_LIT>" : <NUM_LIT:2> } } <EOL> self . _assert_create_object_validation ( data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> data_source . return_value = False <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <EOL> { "<STR_LIT>" : "<STR_LIT:2>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } } <EOL> self . _assert_create_object_validation ( <EOL> data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> def test_overlapping_data ( self ) : <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT:2>" } , <EOL> "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT:2>" } } } <EOL> self . _assert_create_object_validation ( data = data ) <EOL> data = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT:2>" } , <EOL> "<STR_LIT>" : { "<STR_LIT>" : { "<STR_LIT>" : "<STR_LIT:3>" } } } <EOL> self . _assert_create_object_validation ( data = data , bad_req_i = ( <EOL> <NUM_LIT:1> , "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) </s>
