#!/usr/bin/env python import types try: import simplejson as json except ImportError: import json import getpass import unittest import urllib.parse from instagram import client, oauth2, InstagramAPIError TEST_AUTH = False client_id = "DEBUG" client_secret = "DEBUG" access_token = "DEBUG" redirect_uri = "http://example.com" class MockHttp(object): def __init__(self, *args, **kwargs): pass def request(self, url, method="GET", body=None, headers={}): fail_state = { 'status':'400' }, "{}" parsed = urllib.parse.urlparse(url) options = urllib.parse.parse_qs(parsed.query) fn_name = str(active_call) if fn_name == 'get_authorize_login_url': return { 'status': '200', 'content-location':'http://example.com/redirect/login' }, None if not 'access_token' in options and not 'client_id' in options: fn_name += '_unauthorized' if 'self' in url and not 'access_token' in options: fn_name += '_no_auth_user' fl = open('fixtures/%s.json' % fn_name) content = fl.read() json_content = json.loads(content) status = json_content['meta']['code'] return { 'status': status }, content oauth2.Http = MockHttp active_call = None class TestInstagramAPI(client.InstagramAPI): def __getattribute__(self, attr): global active_call actual_val = super(TestInstagramAPI, self).__getattribute__(attr) if isinstance(actual_val, types.MethodType): active_call = attr return actual_val class InstagramAuthTests(unittest.TestCase): def setUp(self): self.unauthenticated_api = TestInstagramAPI(client_id=client_id, redirect_uri=redirect_uri, client_secret=client_secret) def test_authorize_login_url(self): redirect_uri = self.unauthenticated_api.get_authorize_login_url() assert redirect_uri print(("Please visit and authorize at:\n%s" % redirect_uri)) code = input("Paste received code (blank to skip): ").strip() if not code: return access_token = self.unauthenticated_api.exchange_code_for_access_token(code) assert access_token def test_xauth_exchange(self): """ Your client ID must be authorized for xAuth access; email xauth@instagram.com for access""" username = input("Enter username for XAuth (blank to skip): ").strip() if not username: return password = getpass.getpass("Enter password for XAuth (blank to skip): ").strip() access_token = self.unauthenticated_api.exchange_xauth_login_for_access_token(username, password) assert access_token class InstagramAPITests(unittest.TestCase): def setUp(self): super(InstagramAPITests, self).setUp() self.client_only_api = TestInstagramAPI(client_id=client_id) self.api = TestInstagramAPI(access_token=access_token) def test_media_popular(self): self.api.media_popular(count=10) def test_media_search(self): self.client_only_api.media_search(lat=37.7,lng=-122.22) self.api.media_search(lat=37.7,lng=-122.22) def test_media_shortcode(self): self.client_only_api.media_shortcode('os1NQjxtvF') self.api.media_shortcode('os1NQjxtvF') def test_media_likes(self): self.client_only_api.media_likes(media_id=4) def test_like_media(self): self.api.like_media(media_id=4) self.api.unlike_media(media_id=4) """ TEMP; disabled this test while we add a proper response to create_media_comment def test_comment_media(self): comment = self.api.create_media_comment(media_id=4, text='test') self.api.delete_comment(media_id=4, comment_id=comment.id) """ def test_user_feed(self): self.api.user_media_feed(count=50) def test_generator_user_feed(self): generator = self.api.user_media_feed(as_generator=True, max_pages=3, count=2) for page in generator: str(generator) def test_user_liked_media(self): self.api.user_liked_media(count=10) def test_user_recent_media(self): media, url = self.api.user_recent_media(count=10) self.assertTrue( all( [hasattr(obj, 'type') for obj in media] ) ) image = media[0] self.assertEquals( image.get_standard_resolution_url(), "http://distillery-dev.s3.amazonaws.com/media/2011/02/02/1ce5f3f490a640ca9068e6000c91adc5_7.jpg") self.assertEquals( image.get_low_resolution_url(), "http://distillery-dev.s3.amazonaws.com/media/2011/02/02/1ce5f3f490a640ca9068e6000c91adc5_6.jpg") self.assertEquals( image.get_thumbnail_url(), "http://distillery-dev.s3.amazonaws.com/media/2011/02/02/1ce5f3f490a640ca9068e6000c91adc5_5.jpg") self.assertEquals( False, hasattr(image, 'videos') ) video = media[1] self.assertEquals( video.get_standard_resolution_url(), video.videos['standard_resolution'].url) self.assertEquals( video.get_standard_resolution_url(), "http://distilleryvesper9-13.ak.instagram.com/090d06dad9cd11e2aa0912313817975d_101.mp4") self.assertEquals( video.get_low_resolution_url(), "http://distilleryvesper9-13.ak.instagram.com/090d06dad9cd11e2aa0912313817975d_102.mp4") self.assertEquals( video.get_thumbnail_url(), "http://distilleryimage2.ak.instagram.com/11f75f1cd9cc11e2a0fd22000aa8039a_5.jpg") def test_user_search(self): self.api.user_search('mikeyk', 10) def test_user_follows(self): for page in self.api.user_followed_by(as_generator=True): str(page) def test_user_followed_by(self): for page in self.api.user_followed_by(as_generator=True): str(page) def test_other_user_followed_by(self): self.api.user_followed_by(user_id=3) def test_self_info(self): self.api.user() self.assertRaises(InstagramAPIError, self.client_only_api.user) def test_location_recent_media(self): self.api.location_recent_media(location_id=1) def test_location_search(self): self.api.location_search(lat=37.7,lng=-122.22, distance=2500) def test_location(self): self.api.location(1) def test_tag_recent_media(self): self.api.tag_recent_media(tag_name='1', count=5) def test_tag_recent_media_paginated(self): for page in self.api.tag_recent_media(tag_name='1', count=5, as_generator=True, max_pages=2): str(page) def test_tag_search(self): self.api.tag_search("coff") def test_tag(self): self.api.tag("coffee") def test_user_follows(self): self.api.user_follows() def test_user_followed_by(self): self.api.user_followed_by() def test_user_followed_by(self): self.api.user_followed_by() def test_user_requested_by(self): self.api.user_followed_by() def test_user_incoming_requests(self): self.api.user_incoming_requests() def test_change_relationship(self): self.api.change_user_relationship(user_id=10, action="follow") # test shortcuts as well self.api.follow_user(user_id='10') self.api.unfollow_user(user_id='10') def test_geography_recent_media(self): self.api.geography_recent_media(geography_id=1) if __name__ == '__main__': if not TEST_AUTH: del InstagramAuthTests unittest.main()