scapi.tests.scapi

1 from __future__ import with_statement 2 3 import os 4 import urllib2 5 import itertools 6 from textwrap import dedent 7 import pkg_resources 8 import logging 9 import webbrowser 10 from unittest import TestCase 11 12 from configobj import ConfigObj 13 from validate import Validator 14 15 16 import scapi 17 import scapi.authentication 18 19 logger = logging.getLogger("scapi.tests") 20 21 api_logger = logging.getLogger("scapi")

22 23 24 -class SCAPITests(TestCase):

25 26 CONFIG_NAME = "test.ini" 27 TOKEN = None 28 SECRET = None 29 CONSUMER = None 30 CONSUMER_SECRET = None 31 API_HOST = None 32 USER = None 33 PASSWORD = None 34 AUTHENTICATOR = None 35 RUN_INTERACTIVE_TESTS = False 36 37

38 - def setUp(self):

39 self._load_config() 40 assert pkg_resources.resource_exists("scapi.tests.test_connect", "knaster.mp3") 41 self.data = pkg_resources.resource_stream("scapi.tests.test_connect", "knaster.mp3") 42 self.artwork_data = pkg_resources.resource_stream("scapi.tests.test_connect", "spam.jpg")

43 44 CONFIGSPEC=dedent(""" 45 [api] 46 token=string 47 secret=string 48 consumer=string 49 consumer_secret=string 50 api_host=string 51 user=string 52 password=string 53 authenticator=option('oauth', 'base', default='oauth') 54 55 [proxy] 56 use_proxy=boolean(default=false) 57 proxy=string(default=http://127.0.0.1:10000/) 58 59 [logging] 60 test_logger=string(default=ERROR) 61 api_logger=string(default=ERROR) 62 63 [test] 64 run_interactive_tests=boolean(default=false) 65 """) 66 67

68 - def _load_config(self):

69 """ 70 Loads the configuration by looking from 71 72 - the environment variable SCAPI_CONFIG 73 - the installation location upwards until it finds test.ini 74 - the current working directory upwards until it finds test.ini 75 76 Raises an error if there is no config found 77 """ 78 config_name = self.CONFIG_NAME 79 80 name = None 81 82 if "SCAPI_CONFIG" in os.environ: 83 if os.path.exists(os.environ["SCAPI_CONFIG"]): 84 name = os.environ["SCAPI_CONFIG"] 85 86 def search_for_config(current): 87 while current: 88 name = os.path.join(current, config_name) 89 if os.path.exists(name): 90 return name 91 new_current = os.path.dirname(current) 92 if new_current == current: 93 return 94 current = new_current

95 96 if name is None: 97 name = search_for_config(os.path.dirname(__file__)) 98 if name is None: 99 name = search_for_config(os.getcwd()) 100 101 if not name: 102 raise Exception("No test configuration file found!") 103 104 parser = ConfigObj(name, configspec=self.CONFIGSPEC.split("\n")) 105 val = Validator() 106 if not parser.validate(val): 107 raise Exception("Config file validation error") 108 109 api = parser['api'] 110 self.TOKEN = api.get('token') 111 self.SECRET = api.get('secret') 112 self.CONSUMER = api.get('consumer') 113 self.CONSUMER_SECRET = api.get('consumer_secret') 114 self.API_HOST = api.get('api_host') 115 self.USER = api.get('user', None) 116 self.PASSWORD = api.get('password', None) 117 self.AUTHENTICATOR = api.get("authenticator") 118 119 # reset the hard-coded values in the api 120 if self.API_HOST: 121 scapi.AUTHORIZATION_URL = "http://%s/oauth/authorize" % self.API_HOST 122 scapi.REQUEST_TOKEN_URL = 'http://%s/oauth/request_token' % self.API_HOST 123 scapi.ACCESS_TOKEN_URL = 'http://%s/oauth/access_token' % self.API_HOST 124 125 if "proxy" in parser and parser["proxy"]["use_proxy"]: 126 scapi.USE_PROXY = True 127 scapi.PROXY = parser["proxy"]["proxy"] 128 129 if "logging" in parser: 130 logger.setLevel(getattr(logging, parser["logging"]["test_logger"])) 131 api_logger.setLevel(getattr(logging, parser["logging"]["api_logger"])) 132 133 self.RUN_INTERACTIVE_TESTS = parser["test"]["run_interactive_tests"]

134 135 136 @property

137 - def root(self):

138 """ 139 Return the properly configured root-scope. 140 """ 141 if self.AUTHENTICATOR == "oauth": 142 authenticator = scapi.authentication.OAuthAuthenticator(self.CONSUMER, 143 self.CONSUMER_SECRET, 144 self.TOKEN, 145 self.SECRET) 146 elif self.AUTHENTICATOR == "base": 147 authenticator = scapi.authentication.BasicAuthenticator(self.USER, self.PASSWORD, self.CONSUMER, self.CONSUMER_SECRET) 148 else: 149 raise Exception("Unknown authenticator setting: %s", self.AUTHENTICATOR) 150 151 connector = scapi.ApiConnector(host=self.API_HOST, 152 authenticator=authenticator) 153 154 logger.debug("RootScope: %s authenticator: %s", self.API_HOST, self.AUTHENTICATOR) 155 return scapi.Scope(connector)

156 157

158 - def test_connect(self):

159 """ 160 test_connect 161 162 Tries to connect & performs some read-only operations. 163 """ 164 sca = self.root 165 # quite_a_few_users = list(itertools.islice(sca.users(), 0, 127)) 166 167 # logger.debug(quite_a_few_users) 168 # assert isinstance(quite_a_few_users, list) and isinstance(quite_a_few_users[0], scapi.User) 169 user = sca.me() 170 logger.debug(user) 171 assert isinstance(user, scapi.User) 172 contacts = list(user.contacts()) 173 assert isinstance(contacts, list) 174 if contacts: 175 assert isinstance(contacts[0], scapi.User) 176 logger.debug(contacts) 177 tracks = list(user.tracks()) 178 assert isinstance(tracks, list) 179 if tracks: 180 assert isinstance(tracks[0], scapi.Track) 181 logger.debug(tracks)

182 183

184 - def test_access_token_acquisition(self):

185 """ 186 This test is commented out because it needs user-interaction. 187 """ 188 if not self.RUN_INTERACTIVE_TESTS: 189 return 190 oauth_authenticator = scapi.authentication.OAuthAuthenticator(self.CONSUMER, 191 self.CONSUMER_SECRET, 192 None, 193 None) 194 195 sca = scapi.ApiConnector(host=self.API_HOST, authenticator=oauth_authenticator) 196 token, secret = sca.fetch_request_token() 197 authorization_url = sca.get_request_token_authorization_url(token) 198 webbrowser.open(authorization_url) 199 oauth_verifier = raw_input("please enter verifier code as seen in the browser:") 200 201 oauth_authenticator = scapi.authentication.OAuthAuthenticator(self.CONSUMER, 202 self.CONSUMER_SECRET, 203 token, 204 secret) 205 206 sca = scapi.ApiConnector(self.API_HOST, authenticator=oauth_authenticator) 207 token, secret = sca.fetch_access_token(oauth_verifier) 208 logger.info("Access token: '%s'", token) 209 logger.info("Access token secret: '%s'", secret) 210 # force oauth-authentication with the new parameters, and 211 # then invoke some simple test 212 self.AUTHENTICATOR = "oauth" 213 self.TOKEN = token 214 self.SECRET = secret 215 self.test_connect()

216 217

218 - def test_track_creation(self):

219 sca = self.root 220 track = sca.Track.new(title='bar', asset_data=self.data) 221 assert isinstance(track, scapi.Track)

222 223

224 - def test_track_update(self):

225 sca = self.root 226 track = sca.Track.new(title='bar', asset_data=self.data) 227 assert isinstance(track, scapi.Track) 228 track.title='baz' 229 track = sca.Track.get(track.id) 230 assert track.title == "baz"

231 232

233 - def test_scoped_track_creation(self):

234 sca = self.root 235 user = sca.me() 236 track = user.tracks.new(title="bar", asset_data=self.data) 237 assert isinstance(track, scapi.Track)

238 239

240 - def test_upload(self):

241 sca = self.root 242 sca = self.root 243 track = sca.Track.new(title='bar', asset_data=self.data) 244 assert isinstance(track, scapi.Track)

245 246

247 - def test_contact_list(self):

248 sca = self.root 249 user = sca.me() 250 contacts = list(user.contacts()) 251 assert isinstance(contacts, list) 252 if contacts: 253 assert isinstance(contacts[0], scapi.User)

254 255

256 - def test_permissions(self):

257 sca = self.root 258 user = sca.me() 259 tracks = itertools.islice(user.tracks(), 1) 260 for track in tracks: 261 permissions = list(track.permissions()) 262 logger.debug(permissions) 263 assert isinstance(permissions, list) 264 if permissions: 265 assert isinstance(permissions[0], scapi.User)

266 267

268 - def test_setting_permissions(self):

269 sca = self.root 270 me = sca.me() 271 track = sca.Track.new(title='bar', sharing="private", asset_data=self.data) 272 assert track.sharing == "private" 273 users = itertools.islice(sca.users(), 10) 274 users_to_set = [user for user in users if user != me] 275 assert users_to_set, "Didn't find any suitable users" 276 track.permissions = users_to_set 277 assert set(track.permissions()) == set(users_to_set)

278 279

280 - def test_setting_comments(self):

281 sca = self.root 282 user = sca.me() 283 track = sca.Track.new(title='bar', sharing="private", asset_data=self.data) 284 comment = sca.Comment.create(body="This is the body of my comment", timestamp=10) 285 track.comments = comment 286 assert track.comments().next().body == comment.body

287 288

289 - def test_setting_comments_the_way_shawn_says_its_correct(self):

290 sca = self.root 291 track = sca.Track.new(title='bar', sharing="private", asset_data=self.data) 292 cbody = "This is the body of my comment" 293 track.comments.new(body=cbody, timestamp=10) 294 assert list(track.comments())[0].body == cbody

295 296

297 - def test_contact_add_and_removal(self):

298 sca = self.root 299 me = sca.me() 300 for user in sca.users(): 301 if user != me: 302 user_to_set = user 303 break 304 305 contacts = list(me.contacts()) 306 if user_to_set in contacts: 307 me.contacts.remove(user_to_set) 308 309 me.contacts.append(user_to_set) 310 311 contacts = list(me.contacts() ) 312 assert user_to_set.id in [c.id for c in contacts] 313 314 me.contacts.remove(user_to_set) 315 316 contacts = list(me.contacts() ) 317 assert user_to_set not in contacts

318 319

320 - def test_favorites(self):

321 sca = self.root 322 me = sca.me() 323 324 favorites = list(me.favorites()) 325 assert favorites == [] or isinstance(favorites[0], scapi.Track) 326 327 track = None 328 for user in sca.users(): 329 if user == me: 330 continue 331 for track in user.tracks(): 332 break 333 if track is not None: 334 break 335 336 me.favorites.append(track) 337 338 favorites = list(me.favorites()) 339 assert track in favorites 340 341 me.favorites.remove(track) 342 343 favorites = list(me.favorites()) 344 assert track not in favorites

345 346

347 - def test_large_list(self):

348 sca = self.root 349 350 tracks = list(sca.tracks()) 351 if len(tracks) < scapi.ApiConnector.LIST_LIMIT: 352 for i in xrange(scapi.ApiConnector.LIST_LIMIT): 353 sca.Track.new(title='test_track_%i' % i, asset_data=self.data) 354 all_tracks = sca.tracks() 355 assert not isinstance(all_tracks, list) 356 all_tracks = list(all_tracks) 357 assert len(all_tracks) > scapi.ApiConnector.LIST_LIMIT

358 359 360

361 - def test_filtered_list(self):

362 sca = self.root 363 364 tracks = list(sca.tracks(params={ 365 "bpm[from]" : "180", 366 })) 367 if len(tracks) < scapi.ApiConnector.LIST_LIMIT: 368 for i in xrange(scapi.ApiConnector.LIST_LIMIT): 369 sca.Track.new(title='test_track_%i' % i, asset_data=self.data) 370 all_tracks = sca.tracks() 371 assert not isinstance(all_tracks, list) 372 all_tracks = list(all_tracks) 373 assert len(all_tracks) > scapi.ApiConnector.LIST_LIMIT

374 375

376 - def test_events(self):

377 events = list(self.root.events()) 378 assert isinstance(events, list) 379 assert isinstance(events[0], scapi.Event)

380 381

382 - def test_me_having_stress(self):

383 sca = self.root 384 for _ in xrange(20): 385 self.setUp() 386 sca.me()

387 388

389 - def test_non_global_api(self):

390 root = self.root 391 me = root.me() 392 assert isinstance(me, scapi.User) 393 394 # now get something *from* that user 395 list(me.favorites())

396 397

398 - def test_playlists(self):

399 sca = self.root 400 playlists = list(itertools.islice(sca.playlists(), 0, 127)) 401 for playlist in playlists: 402 tracks = playlist.tracks 403 if not isinstance(tracks, list): 404 tracks = [tracks] 405 for trackdata in tracks: 406 print trackdata 407 #user = trackdata.user 408 #print user 409 #print user.tracks() 410 print playlist.user 411 break

412 413 414 415

416 - def test_playlist_creation(self):

417 sca = self.root 418 sca.Playlist.new(title="I'm so happy, happy, happy, happy!")

419 420 421

422 - def test_groups(self):

423 sca = self.root 424 groups = list(itertools.islice(sca.groups(), 0, 127)) 425 for group in groups: 426 users = group.users() 427 for user in users: 428 pass

429 430

431 - def test_track_creation_with_email_sharers(self):

432 sca = self.root 433 emails = [dict(address="deets@web.de"), dict(address="hannes@soundcloud.com")] 434 track = sca.Track.new(title='bar', asset_data=self.data, 435 shared_to=dict(emails=emails) 436 ) 437 assert isinstance(track, scapi.Track)

438 439 440

441 - def test_track_creation_with_artwork(self):

442 sca = self.root 443 track = sca.Track.new(title='bar', 444 asset_data=self.data, 445 artwork_data=self.artwork_data, 446 ) 447 assert isinstance(track, scapi.Track) 448 449 track.title = "foobarbaz"

450 451 452

453 - def test_oauth_get_signing(self):

454 sca = self.root 455 456 url = "http://api.soundcloud.dev/oauth/test_request" 457 params = dict(foo="bar", 458 baz="padamm", 459 ) 460 url += sca._create_query_string(params) 461 signed_url = sca.oauth_sign_get_request(url) 462 463 464 res = urllib2.urlopen(signed_url).read() 465 assert "oauth_nonce" in res

466 467

468 - def test_streaming(self):

469 sca = self.root 470 471 track = sca.tracks(params={ 472 "filter" : "streamable", 473 }).next() 474 475 476 assert isinstance(track, scapi.Track) 477 478 stream_url = track.stream_url 479 480 signed_url = track.oauth_sign_get_request(stream_url)

481 482

483 - def test_downloadable(self):

484 sca = self.root 485 486 track = sca.tracks(params={ 487 "filter" : "downloadable", 488 }).next() 489 490 491 assert isinstance(track, scapi.Track) 492 493 download_url = track.download_url 494 495 signed_url = track.oauth_sign_get_request(download_url) 496 497 data = urllib2.urlopen(signed_url).read() 498 assert data

499 500 501

502 - def test_modifying_playlists(self):

503 sca = self.root 504 505 me = sca.me() 506 my_tracks = list(me.tracks()) 507 508 assert my_tracks 509 510 playlist = me.playlists().next() 511 playlist = sca.Playlist.get(playlist.id) 512 513 assert isinstance(playlist, scapi.Playlist) 514 515 pl_tracks = playlist.tracks 516 517 playlist.title = "foobarbaz"

518 519 520

521 - def test_track_deletion(self):

522 sca = self.root 523 track = sca.Track.new(title='bar', asset_data=self.data, 524 ) 525 526 sca.tracks.remove(track)

527 528 529

530 - def test_track_creation_with_updated_artwork(self):

531 sca = self.root 532 track = sca.Track.new(title='bar', 533 asset_data=self.data, 534 ) 535 assert isinstance(track, scapi.Track) 536 537 track.artwork_data = self.artwork_data

538

Source Code for Module scapi.tests.scapi_tests