Commit a5d1cb35 by Chris Rossi Committed by Diana Huang

pep8 and pylint

parent 64887c68
"""
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
......@@ -8,9 +12,9 @@ from django.utils import timezone
from optparse import make_option
FRIDAY = 4
def get_call_limits():
"""
Returns a tuple of: (max_checks, checks_per_call, time_between_calls)
......@@ -42,18 +46,25 @@ def get_call_limits():
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',
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'),
)
'they have joined or left LinkedIn since the last check'),)
def handle(self, *args, **options):
"""
Check users.
"""
api = LinkedinAPI()
recheck = options.pop('recheck', False)
max_checks, checks_per_call, time_between_calls = get_call_limits()
......@@ -76,6 +87,7 @@ class Command(BaseCommand):
for i in xrange(0, len(check_users), checks_per_call)]
def do_batch(batch):
"Process a batch of users."
emails = [u.email for u in batch]
for user, has_account in zip(batch, api.batch(emails)):
user.linkedin.has_linkedin_account = has_account
......@@ -88,7 +100,12 @@ class Command(BaseCommand):
class LinkedinAPI(object):
"""
Encapsulates the LinkedIn API.
"""
def batch(self, emails):
"""
Get the LinkedIn status for a batch of emails.
"""
pass
"""
Tests for the findusers script.
"""
import datetime
import mock
import pytz
......@@ -9,59 +12,76 @@ from linkedin.management.commands import findusers
class FindUsersTests(unittest.TestCase):
"""
Tests for the findusers script.
"""
@mock.patch('linkedin.management.commands.findusers.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
tz = pytz.timezone('US/Eastern')
tzinfo = pytz.timezone('US/Eastern')
timezone.now.return_value = datetime.datetime(
2013, 12, 14, 0, 0, tzinfo=tz)
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=tz)
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=tz)
2013, 12, 15, 7, 59, tzinfo=tzinfo)
self.assertEqual(fut(), (-1, 80, 1))
@mock.patch('linkedin.management.commands.findusers.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
tz = pytz.timezone('US/Eastern')
tzinfo = pytz.timezone('US/Eastern')
timezone.now.return_value = datetime.datetime(
2013, 12, 11, 11, 3, tzinfo=tz)
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=tz)
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=tz)
2013, 12, 16, 8, 1, tzinfo=tzinfo)
self.assertEqual(fut(), (0, 0, 0))
@mock.patch('linkedin.management.commands.findusers.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
tz = pytz.timezone('US/Eastern')
tzinfo = pytz.timezone('US/Eastern')
timezone.now.return_value = datetime.datetime(
2013, 12, 11, 21, 3, tzinfo=tz)
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=tz)
2013, 12, 11, 7, 59, tzinfo=tzinfo)
self.assertEqual(fut(), (500, 80, 1))
@mock.patch('linkedin.management.commands.findusers.time')
@mock.patch('linkedin.management.commands.findusers.User')
@mock.patch('linkedin.management.commands.findusers.LinkedinAPI')
@mock.patch('linkedin.management.commands.findusers.get_call_limits')
def test_command_success_recheck_no_limits(
self, get_call_limits, API, User, time):
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 = API.return_value
api = apicls.return_value
users = [mock.Mock(email=i) for i in xrange(10)]
User.objects.all.return_value = users
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)
......@@ -73,19 +93,23 @@ class FindUsersTests(unittest.TestCase):
@mock.patch('linkedin.management.commands.findusers.User')
@mock.patch('linkedin.management.commands.findusers.LinkedinAPI')
@mock.patch('linkedin.management.commands.findusers.get_call_limits')
def test_command_success_no_recheck_no_limits(
self, get_call_limits, API, User, time):
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 = API.return_value
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
User.objects.all.return_value = users
usercls.objects.all.return_value = users
def dummy_batch(emails):
"Mock LinkedIn API."
self.assertEqual(len(emails), 4)
return [email % 2 == 0 for email in emails]
api.batch = dummy_batch
......@@ -98,17 +122,21 @@ class FindUsersTests(unittest.TestCase):
@mock.patch('linkedin.management.commands.findusers.User')
@mock.patch('linkedin.management.commands.findusers.LinkedinAPI')
@mock.patch('linkedin.management.commands.findusers.get_call_limits')
def test_command_success_no_recheck_no_users(
self, get_call_limits, API, User, time):
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 = API.return_value
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
User.objects.all.return_value = users
def dummy_batch(emails):
usercls.objects.all.return_value = users
def dummy_batch(_):
"Mock LinkedIn API."
self.assertTrue(False) # shouldn't be called
api.batch = dummy_batch
fut()
......@@ -120,19 +148,23 @@ class FindUsersTests(unittest.TestCase):
@mock.patch('linkedin.management.commands.findusers.User')
@mock.patch('linkedin.management.commands.findusers.LinkedinAPI')
@mock.patch('linkedin.management.commands.findusers.get_call_limits')
def test_command_success_recheck_with_limit(
self, get_call_limits, API, User, time):
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 = API.return_value
api = apicls.return_value
users = [mock.Mock(email=i) for i in xrange(10)]
for user in users:
user.linkedin.has_linkedin_account = None
User.objects.all.return_value = users
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()
......
"""
Models for LinkedIn integration app.
"""
from django.contrib.auth.models import User
from django.db import models
class LinkedIn(models.Model):
"""
Defines a table for storing a users's LinkedIn status.
"""
user = models.OneToOneField(User, primary_key=True)
has_linkedin_account = models.NullBooleanField(default=None)
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