<s> import exceptions <EOL> from nose . tools import assert_raises , assert_equal <EOL> from nose . plugins . skip import SkipTest <EOL> from pylons import config <EOL> from ckan . tests import CkanServerCase <EOL> from ckanclient import CkanClient , CkanApiError <EOL> config_path = config [ '<STR_LIT>' ] <EOL> class TestCkanClient ( CkanServerCase ) : <EOL> @ classmethod <EOL> def setup_class ( self ) : <EOL> self . pid = self . _start_ckan_server ( ) <EOL> self . test_base_location = '<STR_LIT>' <EOL> self . _wait_for_url ( url = self . test_base_location ) <EOL> self . _recreate_ckan_server_testdata ( config_path ) <EOL> test_api_key = '<STR_LIT>' <EOL> test_api_key2 = '<STR_LIT>' <EOL> self . c = CkanClient ( <EOL> base_location = self . test_base_location , <EOL> api_key = test_api_key , <EOL> is_verbose = True , <EOL> ) <EOL> self . c2 = CkanClient ( <EOL> base_location = self . test_base_location , <EOL> api_key = test_api_key2 , <EOL> is_verbose = True , <EOL> ) <EOL> @ classmethod <EOL> def teardown_class ( self ) : <EOL> self . _stop_ckan_server ( self . pid ) <EOL> def delete_relationships ( self ) : <EOL> res = self . c . package_relationship_register_get ( '<STR_LIT>' ) <EOL> if self . c . last_status == <NUM_LIT:200> : <EOL> if self . c . last_message : <EOL> for rel_dict in self . c . last_message : <EOL> self . c . package_relationship_entity_delete ( rel_dict [ '<STR_LIT>' ] , <EOL> rel_dict [ '<STR_LIT:type>' ] , <EOL> rel_dict [ '<STR_LIT:object>' ] ) <EOL> def test_01_get_locations ( self ) : <EOL> rest_base = self . test_base_location + '<STR_LIT>' <EOL> search_base = self . test_base_location + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' ) <EOL> assert url == self . test_base_location , url <EOL> url = self . c . get_location ( '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert url == rest_base + '<STR_LIT>' <EOL> url = self . c . get_location ( '<STR_LIT>' ) <EOL> assert url == search_base + '<STR_LIT>' <EOL> def test_02_get_api_version ( self ) : <EOL> version = self . c . api_version_get ( ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> <EOL> body = self . c . last_body <EOL> assert '<STR_LIT:version>' in body , body <EOL> assert int ( version ) > <NUM_LIT:0> , version <EOL> def test_03_package_register_get ( self ) : <EOL> self . c . package_register_get ( ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> <EOL> body = self . c . last_body <EOL> assert '<STR_LIT>' in body , body <EOL> assert type ( self . c . last_message ) == list <EOL> assert '<STR_LIT>' in self . c . last_message <EOL> def test_04_package_entity_get ( self ) : <EOL> self . c . package_entity_get ( '<STR_LIT>' ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> , status <EOL> body = self . c . last_body <EOL> assert '<STR_LIT>' in body <EOL> assert self . c . last_message <EOL> message = self . c . last_message <EOL> assert type ( message ) == dict <EOL> assert message [ '<STR_LIT:name>' ] == u'<STR_LIT>' <EOL> assert message [ '<STR_LIT:title>' ] == u'<STR_LIT>' <EOL> def test_05_package_entity_get_404 ( self ) : <EOL> assert_raises ( CkanApiError , <EOL> self . c . package_entity_get , <EOL> '<STR_LIT>' ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT> , status <EOL> @ classmethod <EOL> def _generate_pkg_name ( self ) : <EOL> pkg_name = '<STR_LIT>' <EOL> import time <EOL> timestr = str ( time . time ( ) ) . replace ( '<STR_LIT:.>' , '<STR_LIT>' ) <EOL> pkg_name += timestr <EOL> return pkg_name <EOL> def test_06_package_register_post ( self ) : <EOL> pkg_name = self . _generate_pkg_name ( ) <EOL> assert_raises ( CkanApiError , <EOL> self . c . package_entity_get , pkg_name ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT> , status <EOL> package = { <EOL> '<STR_LIT:name>' : pkg_name , <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> } <EOL> self . c . package_register_post ( package ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT> , status <EOL> self . c . package_entity_get ( pkg_name ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> , status <EOL> message = self . c . last_message <EOL> assert message <EOL> assert '<STR_LIT:name>' in message , repr ( message ) <EOL> name = message [ '<STR_LIT:name>' ] <EOL> assert name == pkg_name <EOL> url = message [ '<STR_LIT:url>' ] <EOL> assert url == '<STR_LIT>' <EOL> download_url = message [ '<STR_LIT>' ] <EOL> assert download_url == '<STR_LIT>' <EOL> tags = message [ '<STR_LIT>' ] <EOL> assert set ( tags ) == set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , tags <EOL> extras = message [ '<STR_LIT>' ] <EOL> assert extras == package [ '<STR_LIT>' ] <EOL> def test_07_package_entity_put ( self ) : <EOL> pkg_name_test_07 = self . _generate_pkg_name ( ) <EOL> package = { <EOL> '<STR_LIT:name>' : pkg_name_test_07 , <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> } <EOL> self . c . package_register_post ( package ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT> , status <EOL> mytag = '<STR_LIT>' + pkg_name_test_07 <EOL> package = { <EOL> '<STR_LIT:name>' : pkg_name_test_07 , <EOL> '<STR_LIT:url>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , mytag ] , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> } <EOL> self . c . package_entity_put ( package ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> <EOL> self . c . package_entity_get ( pkg_name_test_07 ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> , status <EOL> message = self . c . last_message <EOL> name = message [ '<STR_LIT:name>' ] <EOL> assert name == pkg_name_test_07 <EOL> url = message [ '<STR_LIT:url>' ] <EOL> assert url == '<STR_LIT>' <EOL> download_url = message [ '<STR_LIT>' ] <EOL> assert download_url == '<STR_LIT>' <EOL> tags = message [ '<STR_LIT>' ] <EOL> assert set ( tags ) == set ( [ '<STR_LIT>' , '<STR_LIT>' , mytag ] ) , tags <EOL> extras = message [ '<STR_LIT>' ] <EOL> assert extras == package [ '<STR_LIT>' ] <EOL> def test_08_package_entity_delete ( self ) : <EOL> pkg_name = self . _generate_pkg_name ( ) <EOL> self . c . package_register_post ( { '<STR_LIT:name>' : pkg_name } ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT> , status <EOL> self . c . package_entity_get ( pkg_name ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> self . c . package_entity_delete ( pkg_name ) <EOL> assert_raises ( CkanApiError , <EOL> self . c2 . package_entity_get , pkg_name ) <EOL> assert self . c2 . last_status == <NUM_LIT> , self . c . last_status <EOL> self . c . package_entity_get ( pkg_name ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> def test_09_tag_register_get ( self ) : <EOL> self . c . tag_register_get ( ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> <EOL> body = self . c . last_body <EOL> assert '<STR_LIT>' in body <EOL> assert type ( self . c . last_message ) == list <EOL> assert '<STR_LIT>' in self . c . last_message <EOL> def test_10_pkg_search_basic ( self ) : <EOL> res = self . c . package_search ( '<STR_LIT>' ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> , status <EOL> assert_equal ( list ( res [ '<STR_LIT>' ] ) , [ u'<STR_LIT>' ] ) <EOL> assert_equal ( res [ '<STR_LIT:count>' ] , <NUM_LIT:1> ) <EOL> def test_10_pkg_search_paged ( self ) : <EOL> res = self . c . package_search ( '<STR_LIT>' , search_options = { '<STR_LIT>' : <NUM_LIT:1> } ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> , status <EOL> all_results = list ( res [ '<STR_LIT>' ] ) <EOL> assert set ( all_results ) >= set ( [ u'<STR_LIT>' , u'<STR_LIT>' ] ) , all_results <EOL> assert res [ '<STR_LIT:count>' ] >= <NUM_LIT:2> , '<STR_LIT>' % ( res , all_results ) <EOL> def test_10_pkg_search_options ( self ) : <EOL> res = self . c . package_search ( None , search_options = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> , status <EOL> assert_equal ( list ( res [ '<STR_LIT>' ] ) , [ u'<STR_LIT>' ] ) <EOL> assert_equal ( res [ '<STR_LIT:count>' ] , <NUM_LIT:1> ) <EOL> def test_10_pkg_search_options_all_fields ( self ) : <EOL> res = self . c . package_search ( None , search_options = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True } ) <EOL> status = self . c . last_status <EOL> assert status == <NUM_LIT:200> , status <EOL> assert_equal ( res [ '<STR_LIT:count>' ] , <NUM_LIT:1> ) <EOL> assert_equal ( list ( res [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] , u'<STR_LIT>' ) <EOL> def test_11_package_relationship_post ( self ) : <EOL> res = self . c . package_relationship_register_get ( '<STR_LIT>' ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> assert not self . c . last_message , self . c . last_body <EOL> res = self . c . package_relationship_entity_post ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> assert self . c . last_status == <NUM_LIT> , self . c . last_status <EOL> finally : <EOL> self . delete_relationships ( ) <EOL> def test_12_package_relationship_get ( self ) : <EOL> res = self . c . package_relationship_entity_post ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> res = self . c . package_relationship_register_get ( '<STR_LIT>' ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> rels = self . c . last_message <EOL> assert len ( rels ) == <NUM_LIT:1> , rels <EOL> assert rels [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' , rels [ <NUM_LIT:0> ] <EOL> assert rels [ <NUM_LIT:0> ] [ '<STR_LIT:object>' ] == '<STR_LIT>' , rels [ <NUM_LIT:0> ] <EOL> assert rels [ <NUM_LIT:0> ] [ '<STR_LIT:type>' ] == '<STR_LIT>' , rels [ <NUM_LIT:0> ] <EOL> assert rels [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' , rels [ <NUM_LIT:0> ] <EOL> finally : <EOL> self . delete_relationships ( ) <EOL> def test_13_package_relationship_put ( self ) : <EOL> res = self . c . package_relationship_entity_post ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> res = self . c . package_relationship_entity_put ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> res = self . c . package_relationship_register_get ( '<STR_LIT>' ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> rels = self . c . last_message <EOL> assert len ( rels ) == <NUM_LIT:1> , rels <EOL> assert rels [ <NUM_LIT:0> ] [ '<STR_LIT>' ] == '<STR_LIT>' , rels [ <NUM_LIT:0> ] <EOL> finally : <EOL> self . delete_relationships ( ) <EOL> def test_14_package_relationship_delete ( self ) : <EOL> res = self . c . package_relationship_entity_post ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> self . c . package_relationship_entity_delete ( '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert_raises ( CkanApiError , <EOL> self . c . package_relationship_register_get , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert self . c . last_status == <NUM_LIT> , self . c . last_status <EOL> res = self . c . package_relationship_register_get ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> assert not res , res <EOL> finally : <EOL> self . delete_relationships ( ) <EOL> def test_15_package_edit_form_get ( self ) : <EOL> try : <EOL> import ckanext . dgu <EOL> except exceptions . ImportError , e : <EOL> raise SkipTest ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' not in config . get ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> raise SkipTest ( '<STR_LIT>' ) <EOL> res = self . c . package_edit_form_get ( '<STR_LIT>' ) <EOL> assert self . c . last_status == <NUM_LIT:200> , self . c . last_status <EOL> assert res , res <EOL> def test_16_group_get ( self ) : <EOL> groups = self . c . group_register_get ( ) <EOL> assert '<STR_LIT>' in groups , groups <EOL> assert '<STR_LIT>' in groups <EOL> david = self . c . group_entity_get ( '<STR_LIT>' ) <EOL> for expected_key in ( '<STR_LIT:name>' , '<STR_LIT:id>' , '<STR_LIT:title>' , '<STR_LIT>' , '<STR_LIT:description>' ) : <EOL> assert expected_key in david , david <EOL> assert set ( david [ '<STR_LIT>' ] ) == set ( ( u'<STR_LIT>' , u'<STR_LIT>' ) ) , david <EOL> roger = self . c . group_entity_get ( '<STR_LIT>' ) <EOL> assert roger [ '<STR_LIT>' ] == [ u'<STR_LIT>' ] , roger </s>
