Skip to content

E
edx-platform
Commit db7308ad by David Ormsbee Committed by Diana Huang
Options

Remove unused parts of LinkedIn API 

Fix whitelist logic to handle empty lists.
parent ed648176
Showing with 23 additions and 664 deletions
lms/djangoapps/linkedin/management/commands/__init__.py
"""
Class for accessing LinkedIn's API.
"""
import json
import urllib2
import urlparse
import uuid
from django.conf import settings
from django.core.management.base import CommandError
import requests
from ...models import LinkedInToken
class LinkedInError(Exception):
pass
class LinkedInAPI(object):
"""
Encapsulates the LinkedIn API.
"""
def __init__(self, command):
config = getattr(settings, "LINKEDIN_API", None)
if not config:
raise CommandError("LINKEDIN_API is not configured")
self.config = config
try:
self.token = LinkedInToken.objects.get()
except LinkedInToken.DoesNotExist:
self.token = None
self.command = command
self.state = str(uuid.uuid4())
def http_error(self, error, message):
"""
Handle an unexpected HTTP response.
"""
stderr = self.command.stderr
stderr.write("!!ERROR!!")
stderr.write(error)
stderr.write(error.read())
raise CommandError(message)
def authorization_url(self):
"""
Synthesize a URL for beginning the authorization flow.
"""
config = self.config
return ("https://www.linkedin.com/uas/oauth2/authorization"
"?response_type=code"
"&client_id=%s&state=%s&redirect_uri=%s" % (
config['CLIENT_ID'], self.state, config['REDIRECT_URI']))
def get_authorization_code(self, redirect):
"""
Extract the authorization code from the redirect URL at the end of
the authorization flow.
"""
query = urlparse.parse_qs(urlparse.urlparse(redirect).query)
assert query['state'][0] == self.state, (query['state'][0], self.state)
return query['code'][0]
def access_token_url(self, code):
"""
Construct URL for retreiving access token, given authorization code.
"""
config = self.config
return ("https://www.linkedin.com/uas/oauth2/accessToken"
"?grant_type=authorization_code"
"&code=%s&redirect_uri=%s&client_id=%s&client_secret=%s" % (
code, config['REDIRECT_URI'], config['CLIENT_ID'],
config['CLIENT_SECRET']))
def call_json_api(self, url):
"""
Make an HTTP call to the LinkedIn JSON API.
"""
if settings.LINKEDIN_API.get('TEST_MODE'):
raise LinkedInError(
"Attempting to make real API call while in test mode - "
"Mock LinkedInAPI.call_json_api instead."
)
try:
request = urllib2.Request(url, headers={'x-li-format': 'json'})
response = urllib2.urlopen(request, timeout=5).read()
return json.loads(response)
except urllib2.HTTPError, error:
self.http_error(error, "Error calling LinkedIn API")
def get_access_token(self, code):
"""
Given an authorization code, get an access token.
"""
response = self.call_json_api(self.access_token_url(code))
access_token = response['access_token']
try:
token = LinkedInToken.objects.get()
token.access_token = access_token
except LinkedInToken.DoesNotExist:
token = LinkedInToken(access_token=access_token)
token.save()
self.token = token
return access_token
def require_token(self):
"""
Raise CommandError if user has not yet obtained an access token.
"""
if self.token is None:
raise CommandError(
"You must log in to LinkedIn in order to use this script. "
"Please use the 'login' command to log in to LinkedIn.")
def batch_url(self, emails):
"""
Construct URL for querying a batch of email addresses.
"""
self.require_token()
queries = ','.join(("email=" + email for email in emails))
url = "https://api.linkedin.com/v1/people::(%s):(id)" % queries
url += "?oauth2_access_token=%s" % self.token.access_token
return url
def batch(self, emails):
"""
Get the LinkedIn status for a batch of emails.
"""
emails = list(emails) # realize generator since we traverse twice
response = self.call_json_api(self.batch_url(emails))
accounts = set(value['_key'][6:] for value in response['values'])
return (email in accounts for email in emails)
lms/djangoapps/linkedin/management/commands/linkedin_findusers.py deleted 100644 → 0
"""
Provides a command to use with Django's `manage.py` that uses LinkedIn's API to
find edX users that are also users on LinkedIn.
"""
import datetime
import pytz
import time
from django.contrib.auth.models import User
from django.core.management.base import BaseCommand, CommandError
from django.utils import timezone
from optparse import make_option
from util.query import use_read_replica_if_available
from linkedin.models import LinkedIn
from . import LinkedInAPI
FRIDAY = 4
def get_call_limits(force_unlimited=False):
"""
Returns a tuple of: (max_checks, checks_per_call, time_between_calls)
Here are the parameters provided by LinkedIn:
Please note: in order to ensure a successful call, please run the calls
between Friday 6pm PST and Monday 5am PST.
During the week, calls are limited to very low volume (500 profiles/day)
and must be run after 6pm and before 5am. This should only be used to do
subsequent trigger emails. Please contact the developer support alias for
more information.
Use 80 emails per API call and 1 call per second.
"""
now = timezone.now().astimezone(pytz.timezone('US/Pacific'))
lastfriday = now
while lastfriday.weekday() != FRIDAY:
lastfriday -= datetime.timedelta(days=1)
safeharbor_begin = lastfriday.replace(hour=18, minute=0)
safeharbor_end = safeharbor_begin + datetime.timedelta(days=2, hours=11)
if force_unlimited or (safeharbor_begin < now < safeharbor_end):
return -1, 80, 1
elif now.hour >= 18 or now.hour < 5:
return 500, 80, 1
else:
return 0, 0, 0
class Command(BaseCommand):
"""
Provides a command to use with Django's `manage.py` that uses LinkedIn's
API to find edX users that are also users on LinkedIn.
"""
args = ''
help = 'Checks LinkedIn for students that are on LinkedIn'
option_list = BaseCommand.option_list + (
make_option(
'--recheck',
action='store_true',
dest='recheck',
default=False,
help='Check users that have been checked in the past to see if '
'they have joined or left LinkedIn since the last check'
),
make_option(
'--force',
action='store_true',
dest='force',
default=False,
help='Disregard the parameters provided by LinkedIn about when it '
'is appropriate to make API calls.'
)
)
def handle(self, *args, **options):
"""
Check users.
"""
api = LinkedInAPI(self)
recheck = options.get('recheck', False)
force = options.get('force', False)
max_checks, checks_per_call, time_between_calls = get_call_limits(force)
if not max_checks:
raise CommandError("No checks allowed during this time.")
def user_batches_to_check():
"""Generate batches of users we should query against LinkedIn."""
count = 0
batch = []
users = use_read_replica_if_available(
None
)
for user in User.objects.all():
if not hasattr(user, 'linkedin'):
LinkedIn(user=user).save()
checked = user.linkedin.has_linkedin_account is not None
if recheck or not checked:
batch.append(user)
if len(batch) == checks_per_call:
yield batch
batch = []
count += 1
if max_checks != -1 and count >= max_checks:
self.stderr.write(
"WARNING: limited to checking only %d users today."
% max_checks)
break
if batch:
yield batch
def update_linkedin_account_status(users):
"""
Given a an iterable of User objects, check their email addresses
to see if they have LinkedIn email addresses and save that
information to our database.
"""
emails = (u.email for u in users)
for user, has_account in zip(users, api.batch(emails)):
linkedin = user.linkedin
if linkedin.has_linkedin_account != has_account:
linkedin.has_linkedin_account = has_account
linkedin.save()
for i, user_batch in enumerate(user_batches_to_check()):
if i > 0:
# Sleep between LinkedIn API web service calls
time.sleep(time_between_calls)
update_linkedin_account_status(user_batch)
lms/djangoapps/linkedin/management/commands/linkedin_login.py deleted 100644 → 0
"""
Log into LinkedIn API.
"""
from django.core.management.base import BaseCommand
from . import LinkedInAPI
class Command(BaseCommand):
"""
Can take a sysadmin through steps to log into LinkedIn API so that the
findusers script can work.
"""
args = ''
help = ('Takes a user through the steps to log in to LinkedIn as a user '
'with API access in order to gain an access token for use by the '
'findusers script.')
def handle(self, *args, **options):
"""
"""
api = LinkedInAPI(self)
print "Let's log into your LinkedIn account."
print "Start by visiting this url:"
print api.authorization_url()
print
print "Within 30 seconds of logging in, enter the full URL of the "
print "webpage you were redirected to: "
redirect = raw_input()
code = api.get_authorization_code(redirect)
api.get_access_token(code)
lms/djangoapps/linkedin/management/commands/linkedin_mailusers.py
......@@ -20,7 +20,6 @@ from certificates.models import GeneratedCertificate
from courseware.courses import get_course_by_id, course_image_url
from ...models import LinkedIn
from . import LinkedInAPI
class Command(BaseCommand):
......@@ -47,15 +46,14 @@ class Command(BaseCommand):
def __init__(self):
super(Command, self).__init__()
self.api = LinkedInAPI(self)
def handle(self, *args, **options):
whitelist = self.api.config.get('EMAIL_WHITELIST')
whitelist = settings.LINKEDIN_API['EMAIL_WHITELIST']
grandfather = options.get('grandfather', False)
accounts = LinkedIn.objects.filter(has_linkedin_account=True)
for account in accounts:
user = account.user
if whitelist is not None and user.email not in whitelist:
if whitelist and user.email not in whitelist:
# Whitelist only certain addresses for testing purposes
continue
emailed = json.loads(account.emailed_courses)
......@@ -63,6 +61,7 @@ class Command(BaseCommand):
certificates = certificates.filter(status='downloadable')
certificates = [cert for cert in certificates
if cert.course_id not in emailed]
if not certificates:
continue
if grandfather:
......@@ -90,7 +89,7 @@ class Command(BaseCommand):
query = [
('pfCertificationName', certificate.name),
('pfAuthorityName', settings.PLATFORM_NAME),
('pfAuthorityId', self.api.config['COMPANY_ID']),
('pfAuthorityId', settings.LINKEDIN_API['COMPANY_ID']),
('pfCertificationUrl', certificate.download_url),
('pfLicenseNo', certificate.course_id),
('pfCertStartDate', course.start.strftime('%Y%m')),
......
lms/djangoapps/linkedin/management/commands/tests/test_api.py deleted 100644 → 0
import mock
import StringIO
from django.core.management.base import CommandError
from django.test import TestCase
from linkedin.management.commands import LinkedInAPI
from linkedin.models import LinkedInToken
class LinkedInAPITests(TestCase):
def setUp(self):
patcher = mock.patch('linkedin.management.commands.uuid.uuid4')
uuid4 = patcher.start()
uuid4.return_value = '0000-0000'
self.addCleanup(patcher.stop)
def make_one(self):
return LinkedInAPI(DummyCommand())
@mock.patch('django.conf.settings.LINKEDIN_API', None)
def test_ctor_no_api_config(self):
with self.assertRaises(CommandError):
self.make_one()
def test_ctor_no_token(self):
api = self.make_one()
self.assertEqual(api.token, None)
def test_ctor_with_token(self):
token = LinkedInToken()
token.save()
api = self.make_one()
self.assertEqual(api.token, token)
def test_http_error(self):
api = self.make_one()
with self.assertRaises(CommandError):
api.http_error(DummyHTTPError(), "That didn't work")
self.assertEqual(
api.command.stderr.getvalue(),
"!!ERROR!!"
"HTTPError OMG!"
"OMG OHNOES!")
def test_authorization_url(self):
api = self.make_one()
self.assertEqual(
api.authorization_url(),
'https://www.linkedin.com/uas/oauth2/authorization?'
'response_type=code&client_id=12345&state=0000-0000&'
'redirect_uri=http://bar.foo')
def test_get_authorization_code(self):
fut = self.make_one().get_authorization_code
self.assertEqual(
fut('http://foo.bar/?state=0000-0000&code=54321'), '54321')
def test_access_token_url(self):
fut = self.make_one().access_token_url
self.assertEqual(
fut('54321'),
'https://www.linkedin.com/uas/oauth2/accessToken?'
'grant_type=authorization_code&code=54321&'
'redirect_uri=http://bar.foo&client_id=12345&client_secret=SECRET')
def test_get_access_token(self):
api = self.make_one()
api.call_json_api = mock.Mock(return_value={'access_token': '777'})
self.assertEqual(api.get_access_token('54321'), '777')
token = LinkedInToken.objects.get()
self.assertEqual(token.access_token, '777')
def test_get_access_token_overwrite_previous(self):
LinkedInToken(access_token='888').save()
api = self.make_one()
api.call_json_api = mock.Mock(return_value={'access_token': '777'})
self.assertEqual(api.get_access_token('54321'), '777')
token = LinkedInToken.objects.get()
self.assertEqual(token.access_token, '777')
def test_require_token_no_token(self):
fut = self.make_one().require_token
with self.assertRaises(CommandError):
fut()
def test_require_token(self):
LinkedInToken().save()
fut = self.make_one().require_token
fut()
def test_batch_url(self):
LinkedInToken(access_token='777').save()
fut = self.make_one().batch_url
emails = ['foo@bar', 'bar@foo']
self.assertEquals(
fut(emails),
'https://api.linkedin.com/v1/people::(email=foo@bar,email=bar@foo):'
'(id)?oauth2_access_token=777')
def test_batch(self):
LinkedInToken(access_token='777').save()
api = self.make_one()
api.call_json_api = mock.Mock(return_value={
'values': [{'_key': 'email=bar@foo'}]})
emails = ['foo@bar', 'bar@foo']
self.assertEqual(list(api.batch(emails)), [False, True])
class DummyCommand(object):
def __init__(self):
self.stderr = StringIO.StringIO()
class DummyHTTPError(object):
def __str__(self):
return 'HTTPError OMG!'
def read(self):
return 'OMG OHNOES!'
lms/djangoapps/linkedin/management/commands/tests/test_findusers.py deleted 100644 → 0
"""
Tests for the findusers script.
"""
import datetime
import mock
import pytz
import StringIO
from django.test import TestCase
from linkedin.management.commands import linkedin_findusers as findusers
MODULE = 'linkedin.management.commands.linkedin_findusers.'
class FindUsersTests(TestCase):
"""
Tests for the findusers script.
"""
@mock.patch(MODULE + 'timezone')
def test_get_call_limits_in_safe_harbor(self, timezone):
"""
We should be able to perform unlimited API calls during "safe harbor".
"""
fut = findusers.get_call_limits
tzinfo = pytz.timezone('US/Eastern')
timezone.now.return_value = datetime.datetime(
2013, 12, 14, 0, 0, tzinfo=tzinfo)
self.assertEqual(fut(), (-1, 80, 1))
timezone.now.return_value = datetime.datetime(
2013, 12, 13, 21, 1, tzinfo=tzinfo)
self.assertEqual(fut(), (-1, 80, 1))
timezone.now.return_value = datetime.datetime(
2013, 12, 15, 7, 59, tzinfo=tzinfo)
self.assertEqual(fut(), (-1, 80, 1))
@mock.patch(MODULE + 'timezone')
def test_get_call_limits_in_business_hours(self, timezone):
"""
During business hours we shouldn't be able to make any API calls.
"""
fut = findusers.get_call_limits
tzinfo = pytz.timezone('US/Eastern')
timezone.now.return_value = datetime.datetime(
2013, 12, 11, 11, 3, tzinfo=tzinfo)
self.assertEqual(fut(), (0, 0, 0))
timezone.now.return_value = datetime.datetime(
2013, 12, 13, 20, 59, tzinfo=tzinfo)
self.assertEqual(fut(), (0, 0, 0))
timezone.now.return_value = datetime.datetime(
2013, 12, 16, 8, 1, tzinfo=tzinfo)
self.assertEqual(fut(), (0, 0, 0))
@mock.patch(MODULE + 'timezone')
def test_get_call_limits_on_weeknights(self, timezone):
"""
On weeknights outside of "safe harbor" we can only make limited API
calls.
"""
fut = findusers.get_call_limits
tzinfo = pytz.timezone('US/Eastern')
timezone.now.return_value = datetime.datetime(
2013, 12, 11, 21, 3, tzinfo=tzinfo)
self.assertEqual(fut(), (500, 80, 1))
timezone.now.return_value = datetime.datetime(
2013, 12, 11, 7, 59, tzinfo=tzinfo)
self.assertEqual(fut(), (500, 80, 1))
@mock.patch(MODULE + 'time')
@mock.patch(MODULE + 'User')
@mock.patch(MODULE + 'LinkedInAPI')
@mock.patch(MODULE + 'get_call_limits')
def test_command_success_recheck_no_limits(self, get_call_limits, apicls,
usercls, time):
"""
Test rechecking all users with no API limits.
"""
fut = findusers.Command().handle
get_call_limits.return_value = (-1, 6, 42)
api = apicls.return_value
users = [mock.Mock(email=i) for i in xrange(10)]
usercls.objects.all.return_value = users
def dummy_batch(emails):
"Mock LinkedIn API."
return [email % 2 == 0 for email in emails]
api.batch = dummy_batch
fut(recheck=True)
time.sleep.assert_called_once_with(42)
self.assertEqual([u.linkedin.has_linkedin_account for u in users],
[i % 2 == 0 for i in xrange(10)])
@mock.patch(MODULE + 'time')
@mock.patch(MODULE + 'User')
@mock.patch(MODULE + 'LinkedInAPI')
@mock.patch(MODULE + 'get_call_limits')
def test_command_success_no_recheck_no_limits(self, get_call_limits, apicls,
usercls, time):
"""
Test checking only unchecked users, with no API limits.
"""
fut = findusers.Command().handle
get_call_limits.return_value = (-1, 6, 42)
api = apicls.return_value
users = [mock.Mock(email=i) for i in xrange(10)]
for user in users[:6]:
user.linkedin.has_linkedin_account = user.email % 2 == 0
for user in users[6:]:
user.linkedin.has_linkedin_account = None
usercls.objects.all.return_value = users
def dummy_batch(emails):
"Mock LinkedIn API."
emails = list(emails)
self.assertEqual(len(emails), 4)
return [email % 2 == 0 for email in emails]
api.batch = dummy_batch
fut()
time.sleep.assert_not_called()
self.assertEqual([u.linkedin.has_linkedin_account for u in users],
[i % 2 == 0 for i in xrange(10)])
@mock.patch(MODULE + 'time')
@mock.patch(MODULE + 'User')
@mock.patch(MODULE + 'LinkedInAPI')
@mock.patch(MODULE + 'get_call_limits')
def test_command_success_no_recheck_no_users(self, get_call_limits, apicls,
usercls, time):
"""
Test no users to check.
"""
fut = findusers.Command().handle
get_call_limits.return_value = (-1, 6, 42)
api = apicls.return_value
users = [mock.Mock(email=i) for i in xrange(10)]
for user in users:
user.linkedin.has_linkedin_account = user.email % 2 == 0
usercls.objects.all.return_value = users
def dummy_batch(_):
"Mock LinkedIn API."
self.assertTrue(False) # shouldn't be called
api.batch = dummy_batch
fut()
time.sleep.assert_not_called()
self.assertEqual([u.linkedin.has_linkedin_account for u in users],
[i % 2 == 0 for i in xrange(10)])
@mock.patch(MODULE + 'time')
@mock.patch(MODULE + 'User')
@mock.patch(MODULE + 'LinkedInAPI')
@mock.patch(MODULE + 'get_call_limits')
def test_command_success_recheck_with_limit(self, get_call_limits, apicls,
usercls, time):
"""
Test recheck all users with API limit.
"""
command = findusers.Command()
command.stderr = StringIO.StringIO()
fut = command.handle
get_call_limits.return_value = (9, 6, 42)
api = apicls.return_value
users = [mock.Mock(email=i) for i in xrange(10)]
for user in users:
user.linkedin.has_linkedin_account = None
usercls.objects.all.return_value = users
def dummy_batch(emails):
"Mock LinkedIn API."
return [email % 2 == 0 for email in emails]
api.batch = dummy_batch
fut()
time.sleep.assert_called_once_with(42)
self.assertEqual([u.linkedin.has_linkedin_account for u in users[:9]],
[i % 2 == 0 for i in xrange(9)])
self.assertEqual(users[9].linkedin.has_linkedin_account, None)
self.assertTrue(command.stderr.getvalue().startswith("WARNING"))
@mock.patch(MODULE + 'User')
@mock.patch(MODULE + 'LinkedInAPI')
@mock.patch(MODULE + 'get_call_limits')
def test_command_success_recheck_with_force(self, get_call_limits, apicls,
usercls):
"""
Test recheck all users with API limit.
"""
command = findusers.Command()
command.stderr = StringIO.StringIO()
fut = command.handle
get_call_limits.return_value = (9, 6, 42)
api = apicls.return_value
users = [mock.Mock(email=i) for i in xrange(10)]
for user in users:
user.linkedin.has_linkedin_account = None
usercls.objects.all.return_value = users
def dummy_batch(emails):
"Mock LinkedIn API."
return [email % 2 == 0 for email in emails]
api.batch = dummy_batch
get_call_limits.return_value = (-1, 80, 1)
fut(force=True)
self.assertEqual([u.linkedin.has_linkedin_account for u in users],
[i % 2 == 0 for i in xrange(10)])
@mock.patch(MODULE + 'get_call_limits')
def test_command_no_api_calls(self, get_call_limits):
"""
Test rechecking all users with no API limits.
"""
from django.core.management.base import CommandError
fut = findusers.Command().handle
get_call_limits.return_value = (0, 0, 0)
with self.assertRaises(CommandError):
fut(recheck=True)
lms/djangoapps/linkedin/management/commands/tests/test_mailusers.py
......@@ -15,8 +15,8 @@ from django.test import TestCase
from xmodule.modulestore.tests.factories import CourseFactory
from student.models import UserProfile
from linkedin.models import LinkedIn
from xmodule.modulestore.tests.django_utils import mixed_store_config
from linkedin.models import LinkedIn
from linkedin.management.commands import linkedin_mailusers as mailusers
MODULE = 'linkedin.management.commands.linkedin_mailusers.'
......
lms/djangoapps/linkedin/models.py
......@@ -12,11 +12,3 @@ class LinkedIn(models.Model):
user = models.OneToOneField(User, primary_key=True)
has_linkedin_account = models.NullBooleanField(default=None)
emailed_courses = models.TextField(default="[]") # JSON list of course ids
class LinkedInToken(models.Model):
"""
For storing access token and authorization code after logging in to
LinkedIn.
"""
access_token = models.CharField(max_length=255)
lms/envs/common.py
......@@ -1143,3 +1143,18 @@ GRADES_DOWNLOAD = {
'BUCKET': 'edx-grades',
'ROOT_PATH': '/tmp/edx-s3/grades',
}
##################### LinkedIn #####################
INSTALLED_APPS += ('django_openid_auth',)
LINKEDIN_API = {
'COMPANY_NAME': 'edX',
}
############################ LinkedIn Integration #############################
INSTALLED_APPS += ('linkedin',)
LINKEDIN_API = {
'EMAIL_WHITELIST': [],
'COMPANY_ID': '2746406',
}
lms/envs/test.py
......@@ -257,18 +257,6 @@ XQUEUE_PORT = 8040
YOUTUBE_PORT = 8031
LTI_PORT = 8765
############################ LinkedIn Integration #############################
INSTALLED_APPS += ('linkedin',)
LINKEDIN_API = {
'CLIENT_ID': '12345',
'CLIENT_SECRET': 'SECRET',
'REDIRECT_URI': 'http://bar.foo',
'COMPANY_NAME': 'edX',
'COMPANY_ID': '0000000',
'EMAIL_FROM': 'The Team <team@test.foo>',
'TEST_MODE': True
}
################### Make tests faster
#http://slacy.com/blog/2012/04/make-your-tests-faster-in-django-1-4/
......@@ -318,3 +306,6 @@ if len(MICROSITE_CONFIGURATION.keys()) > 0:
VIRTUAL_UNIVERSITIES,
microsites_root=ENV_ROOT / 'edx-platform' / 'test_microsites'
)
######### LinkedIn ########
LINKEDIN_API['COMPANY_ID'] = '0000000'
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment