tests.py

#! /usr/bin/python

import types
import sys
import simplejson
import time
import getpass
import unittest
import urlparse
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 = urlparse.urlparse(url)
        options = urlparse.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 = simplejson.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 = raw_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 = raw_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_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):
        self.api.user_recent_media(count=10)

    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()