Tests for embargo middleware feature
This commit is contained in:
@@ -2,10 +2,11 @@
|
||||
Django admin page for embargo models
|
||||
"""
|
||||
from django.contrib import admin
|
||||
import textwrap
|
||||
|
||||
from config_models.admin import ConfigurationModelAdmin
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPException
|
||||
from embargo.forms import EmbargoedCourseForm, EmbargoedStateForm, IPExceptionForm
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPFilter
|
||||
from embargo.forms import EmbargoedCourseForm, EmbargoedStateForm, IPFilterForm
|
||||
|
||||
|
||||
class EmbargoedCourseAdmin(admin.ModelAdmin):
|
||||
@@ -14,13 +15,15 @@ class EmbargoedCourseAdmin(admin.ModelAdmin):
|
||||
fieldsets = (
|
||||
(None, {
|
||||
'fields': ('course_id', 'embargoed'),
|
||||
'description': '''
|
||||
Enter a course id in the following box. Do not enter leading or trailing slashes. There is no need to surround the course ID with quotes.
|
||||
'description': textwrap.dedent("""Enter a course id in the following box.
|
||||
Do not enter leading or trailing slashes. There is no need to surround the
|
||||
course ID with quotes.
|
||||
Validation will be performed on the course name, and if it is invalid, an
|
||||
error message will display.
|
||||
|
||||
Validation will be performed on the course name, and if it is invalid, an error message will display.
|
||||
|
||||
To enable embargos against this course (restrict course access from embargoed states), check the "Embargoed" box, then click "Save".
|
||||
'''
|
||||
To enable embargos against this course (restrict course access from embargoed
|
||||
states), check the "Embargoed" box, then click "Save".
|
||||
""")
|
||||
}),
|
||||
)
|
||||
|
||||
@@ -31,31 +34,30 @@ class EmbargoedStateAdmin(ConfigurationModelAdmin):
|
||||
fieldsets = (
|
||||
(None, {
|
||||
'fields': ('embargoed_countries',),
|
||||
'description': '''
|
||||
Enter the two-letter ISO-3166-1 Alpha-2 code of the country or countries to embargo
|
||||
in the following box. For help, see
|
||||
<a href="http://en.wikipedia.org/wiki/ISO_3166-1#Officially_assigned_code_elements">
|
||||
this list of ISO-3166-1 country codes</a>.
|
||||
'description': textwrap.dedent("""Enter the two-letter ISO-3166-1 Alpha-2
|
||||
code of the country or countries to embargo in the following box. For help,
|
||||
see <a href="http://en.wikipedia.org/wiki/ISO_3166-1#Officially_assigned_code_elements">
|
||||
this list of ISO-3166-1 country codes</a>.
|
||||
|
||||
Enter the embargoed country codes separated by a comma. Do not surround with quotes.
|
||||
'''
|
||||
Enter the embargoed country codes separated by a comma. Do not surround with quotes.
|
||||
""")
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
class IPExceptionAdmin(ConfigurationModelAdmin):
|
||||
class IPFilterAdmin(ConfigurationModelAdmin):
|
||||
"""Admin for blacklisting/whitelisting specific IP addresses"""
|
||||
form = IPExceptionForm
|
||||
form = IPFilterForm
|
||||
fieldsets = (
|
||||
(None, {
|
||||
'fields': ('whitelist', 'blacklist'),
|
||||
'description': '''
|
||||
Enter specific IP addresses to explicitly whitelist (not block) or blacklist (block) in
|
||||
the appropriate box below. Separate IP addresses with a comma. Do not surround with quotes.
|
||||
'''
|
||||
'description': textwrap.dedent("""Enter specific IP addresses to explicitly
|
||||
whitelist (not block) or blacklist (block) in the appropriate box below.
|
||||
Separate IP addresses with a comma. Do not surround with quotes.
|
||||
""")
|
||||
}),
|
||||
)
|
||||
|
||||
admin.site.register(EmbargoedCourse, EmbargoedCourseAdmin)
|
||||
admin.site.register(EmbargoedState, EmbargoedStateAdmin)
|
||||
admin.site.register(IPException, IPExceptionAdmin)
|
||||
admin.site.register(IPFilter, IPFilterAdmin)
|
||||
|
||||
@@ -4,7 +4,7 @@ Defines forms for providing validation of embargo admin details.
|
||||
|
||||
from django import forms
|
||||
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPException
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPFilter
|
||||
from embargo.fixtures.country_codes import COUNTRY_CODES
|
||||
|
||||
import socket
|
||||
@@ -50,13 +50,16 @@ class EmbargoedStateForm(forms.ModelForm): # pylint: disable=incomplete-protoco
|
||||
|
||||
def _is_valid_code(self, code):
|
||||
"""Whether or not code is a valid country code"""
|
||||
if len(code) == 2 and code in COUNTRY_CODES:
|
||||
if code in COUNTRY_CODES:
|
||||
return True
|
||||
return False
|
||||
|
||||
def clean_embargoed_countries(self):
|
||||
"""Validate the country list"""
|
||||
embargoed_countries = self.cleaned_data["embargoed_countries"]
|
||||
if embargoed_countries == '':
|
||||
return ''
|
||||
|
||||
error_countries = []
|
||||
|
||||
for country in embargoed_countries.split(','):
|
||||
@@ -72,11 +75,11 @@ class EmbargoedStateForm(forms.ModelForm): # pylint: disable=incomplete-protoco
|
||||
return embargoed_countries
|
||||
|
||||
|
||||
class IPExceptionForm(forms.ModelForm): # pylint: disable=incomplete-protocol
|
||||
class IPFilterForm(forms.ModelForm): # pylint: disable=incomplete-protocol
|
||||
"""Form validating entry of IP addresses"""
|
||||
|
||||
class Meta: # pylint: disable=missing-docstring
|
||||
model = IPException
|
||||
model = IPFilter
|
||||
|
||||
def _is_valid_ipv4(self, address):
|
||||
"""Whether or not address is a valid ipv4 address"""
|
||||
|
||||
@@ -1,23 +1,35 @@
|
||||
"""
|
||||
Middleware for embargoing courses.
|
||||
|
||||
IMPORTANT NOTE: This code WILL NOT WORK if you have a misconfigured proxy
|
||||
server. If you are configuring embargo functionality, or if you are
|
||||
experiencing mysterious problems with embargoing, please check that your
|
||||
reverse proxy is setting any of the well known client IP address headers (ex.,
|
||||
HTTP_X_FORWARDED_FOR).
|
||||
"""
|
||||
|
||||
|
||||
from django.shortcuts import redirect
|
||||
from util.request import course_id_from_url
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPException
|
||||
from ipware.ip import get_ip
|
||||
import pygeoip
|
||||
import django.core.exceptions
|
||||
|
||||
from django.conf import settings
|
||||
from django.shortcuts import redirect
|
||||
from ipware.ip import get_ip
|
||||
from util.request import course_id_from_url
|
||||
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPFilter
|
||||
|
||||
|
||||
class EmbargoMiddleware(object):
|
||||
"""
|
||||
Middleware for embargoing courses
|
||||
|
||||
This is configured by creating ``DarkLangConfig`` rows in the database,
|
||||
using the django admin site.
|
||||
This is configured by creating ``EmbargoedCourse``, ``EmbargoedState``, and
|
||||
optionally ``IPFilter`` rows in the database, using the django admin site.
|
||||
"""
|
||||
def __init__(self):
|
||||
# If embargoing is turned off, make this middleware do nothing
|
||||
if not settings.FEATURES.get('EMBARGO', False):
|
||||
raise django.core.exceptions.MiddlewareNotUsed()
|
||||
|
||||
def process_request(self, request):
|
||||
"""
|
||||
@@ -28,15 +40,15 @@ class EmbargoMiddleware(object):
|
||||
|
||||
# If they're trying to access a course that cares about embargoes
|
||||
if EmbargoedCourse.is_embargoed(course_id):
|
||||
|
||||
# If we're having performance issues, add caching here
|
||||
ip_addr = get_ip(request)
|
||||
|
||||
# if blacklisted, immediately fail
|
||||
if ip_addr in IPException.current().blacklist_ips:
|
||||
if ip_addr in IPFilter.current().blacklist_ips:
|
||||
return redirect('embargo')
|
||||
|
||||
country_code_from_ip = pygeoip.GeoIP(settings.GEOIP_PATH).country_code_by_addr(ip_addr)
|
||||
is_embargoed = country_code_from_ip in EmbargoedState.current().embargoed_countries_list
|
||||
# Fail if country is embargoed and the ip address isn't explicitly whitelisted
|
||||
if is_embargoed and ip_addr not in IPException.current().whitelist_ips:
|
||||
if is_embargoed and ip_addr not in IPFilter.current().whitelist_ips:
|
||||
return redirect('embargo')
|
||||
|
||||
@@ -26,8 +26,8 @@ class Migration(SchemaMigration):
|
||||
))
|
||||
db.send_create_signal('embargo', ['EmbargoedState'])
|
||||
|
||||
# Adding model 'IPException'
|
||||
db.create_table('embargo_ipexception', (
|
||||
# Adding model 'IPFilter'
|
||||
db.create_table('embargo_ipfilter', (
|
||||
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
|
||||
('change_date', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, blank=True)),
|
||||
('changed_by', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['auth.User'], null=True, on_delete=models.PROTECT)),
|
||||
@@ -35,7 +35,7 @@ class Migration(SchemaMigration):
|
||||
('whitelist', self.gf('django.db.models.fields.TextField')(blank=True)),
|
||||
('blacklist', self.gf('django.db.models.fields.TextField')(blank=True)),
|
||||
))
|
||||
db.send_create_signal('embargo', ['IPException'])
|
||||
db.send_create_signal('embargo', ['IPFilter'])
|
||||
|
||||
|
||||
def backwards(self, orm):
|
||||
@@ -45,8 +45,8 @@ class Migration(SchemaMigration):
|
||||
# Deleting model 'EmbargoedState'
|
||||
db.delete_table('embargo_embargoedstate')
|
||||
|
||||
# Deleting model 'IPException'
|
||||
db.delete_table('embargo_ipexception')
|
||||
# Deleting model 'IPFilter'
|
||||
db.delete_table('embargo_ipfilter')
|
||||
|
||||
|
||||
models = {
|
||||
@@ -100,8 +100,8 @@ class Migration(SchemaMigration):
|
||||
'enabled': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'})
|
||||
},
|
||||
'embargo.ipexception': {
|
||||
'Meta': {'object_name': 'IPException'},
|
||||
'embargo.ipfilter': {
|
||||
'Meta': {'object_name': 'IPFilter'},
|
||||
'blacklist': ('django.db.models.fields.TextField', [], {'blank': 'True'}),
|
||||
'change_date': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
|
||||
'changed_by': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['auth.User']", 'null': 'True', 'on_delete': 'models.PROTECT'}),
|
||||
|
||||
@@ -60,10 +60,12 @@ class EmbargoedState(ConfigurationModel):
|
||||
"""
|
||||
Return a list of upper case country codes
|
||||
"""
|
||||
if self.embargoed_countries == '':
|
||||
return []
|
||||
return [country.strip().upper() for country in self.embargoed_countries.split(',')] # pylint: disable=no-member
|
||||
|
||||
|
||||
class IPException(ConfigurationModel):
|
||||
class IPFilter(ConfigurationModel):
|
||||
"""
|
||||
Register specific IP addresses to explicitly block or unblock.
|
||||
"""
|
||||
@@ -82,6 +84,8 @@ class IPException(ConfigurationModel):
|
||||
"""
|
||||
Return a list of valid IP addresses to whitelist
|
||||
"""
|
||||
if self.whitelist == '':
|
||||
return []
|
||||
return [addr.strip() for addr in self.whitelist.split(',')] # pylint: disable=no-member
|
||||
|
||||
@property
|
||||
@@ -89,4 +93,6 @@ class IPException(ConfigurationModel):
|
||||
"""
|
||||
Return a list of valid IP addresses to blacklist
|
||||
"""
|
||||
if self.blacklist == '':
|
||||
return []
|
||||
return [addr.strip() for addr in self.blacklist.split(',')] # pylint: disable=no-member
|
||||
|
||||
203
common/djangoapps/embargo/tests/test_forms.py
Normal file
203
common/djangoapps/embargo/tests/test_forms.py
Normal file
@@ -0,0 +1,203 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Unit tests for embargo app admin forms.
|
||||
"""
|
||||
|
||||
from django.test import TestCase
|
||||
from django.test.utils import override_settings
|
||||
|
||||
# Explicitly import the cache from ConfigurationModel so we can reset it after each test
|
||||
from config_models.models import cache
|
||||
from embargo.forms import EmbargoedCourseForm, EmbargoedStateForm, IPFilterForm
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPFilter
|
||||
|
||||
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
|
||||
from xmodule.modulestore.tests.factories import CourseFactory
|
||||
from courseware.tests.tests import TEST_DATA_MONGO_MODULESTORE
|
||||
|
||||
|
||||
@override_settings(MODULESTORE=TEST_DATA_MONGO_MODULESTORE)
|
||||
class EmbargoCourseFormTest(ModuleStoreTestCase):
|
||||
"""Test the course form properly validates course IDs"""
|
||||
|
||||
def setUp(self):
|
||||
self.course = CourseFactory.create()
|
||||
self.true_form_data = {'course_id': self.course.id, 'embargoed': True}
|
||||
self.false_form_data = {'course_id': self.course.id, 'embargoed': False}
|
||||
|
||||
def tearDown(self):
|
||||
# Delete any EmbargoedCourse record we may have created
|
||||
try:
|
||||
record = EmbargoedCourse.objects.get(course_id=self.course.id)
|
||||
record.delete()
|
||||
except EmbargoedCourse.DoesNotExist:
|
||||
return
|
||||
|
||||
def test_embargo_course(self):
|
||||
self.assertFalse(EmbargoedCourse.is_embargoed(self.course.id))
|
||||
# Test adding embargo to this course
|
||||
form = EmbargoedCourseForm(data=self.true_form_data)
|
||||
# Validation should work
|
||||
self.assertTrue(form.is_valid())
|
||||
form.save()
|
||||
# Check that this course is embargoed
|
||||
self.assertTrue(EmbargoedCourse.is_embargoed(self.course.id))
|
||||
|
||||
def test_repeat_course(self):
|
||||
# Initially course shouldn't be authorized
|
||||
self.assertFalse(EmbargoedCourse.is_embargoed(self.course.id))
|
||||
# Test authorizing the course, which should totally work
|
||||
form = EmbargoedCourseForm(data=self.true_form_data)
|
||||
# Validation should work
|
||||
self.assertTrue(form.is_valid())
|
||||
form.save()
|
||||
# Check that this course is authorized
|
||||
self.assertTrue(EmbargoedCourse.is_embargoed(self.course.id))
|
||||
|
||||
# Now make a new course authorization with the same course id that tries to turn email off
|
||||
form = EmbargoedCourseForm(data=self.false_form_data)
|
||||
# Validation should not work because course_id field is unique
|
||||
self.assertFalse(form.is_valid())
|
||||
self.assertEquals(
|
||||
"Embargoed course with this Course id already exists.",
|
||||
form._errors['course_id'][0] # pylint: disable=protected-access
|
||||
)
|
||||
with self.assertRaisesRegexp(ValueError, "The EmbargoedCourse could not be created because the data didn't validate."):
|
||||
form.save()
|
||||
|
||||
# Course should still be authorized (invalid attempt had no effect)
|
||||
self.assertTrue(EmbargoedCourse.is_embargoed(self.course.id))
|
||||
|
||||
def test_form_typo(self):
|
||||
# Munge course id
|
||||
bad_id = self.course.id + '_typo'
|
||||
|
||||
form_data = {'course_id': bad_id, 'embargoed': True}
|
||||
form = EmbargoedCourseForm(data=form_data)
|
||||
# Validation shouldn't work
|
||||
self.assertFalse(form.is_valid())
|
||||
|
||||
msg = 'COURSE NOT FOUND'
|
||||
msg += u' --- Entered course id was: "{0}". '.format(bad_id)
|
||||
msg += 'Please recheck that you have supplied a valid course id.'
|
||||
self.assertEquals(msg, form._errors['course_id'][0]) # pylint: disable=protected-access
|
||||
|
||||
with self.assertRaisesRegexp(ValueError, "The EmbargoedCourse could not be created because the data didn't validate."):
|
||||
form.save()
|
||||
|
||||
def test_invalid_location(self):
|
||||
# Munge course id
|
||||
bad_id = self.course.id.split('/')[-1]
|
||||
|
||||
form_data = {'course_id': bad_id, 'embargoed': True}
|
||||
form = EmbargoedCourseForm(data=form_data)
|
||||
# Validation shouldn't work
|
||||
self.assertFalse(form.is_valid())
|
||||
|
||||
msg = 'INVALID LOCATION'
|
||||
msg += u' --- Entered course id was: "{0}". '.format(bad_id)
|
||||
msg += 'Please recheck that you have supplied a valid course id.'
|
||||
self.assertEquals(msg, form._errors['course_id'][0]) # pylint: disable=protected-access
|
||||
|
||||
with self.assertRaisesRegexp(ValueError, "The EmbargoedCourse could not be created because the data didn't validate."):
|
||||
form.save()
|
||||
|
||||
|
||||
class EmbargoedStateFormTest(TestCase):
|
||||
"""Test form for adding new states"""
|
||||
|
||||
def tearDown(self):
|
||||
# Explicitly clear ConfigurationModel's cache so tests have a clear cache
|
||||
# and don't interfere with each other
|
||||
cache.clear()
|
||||
|
||||
def test_add_valid_states(self):
|
||||
# test adding valid two letter states
|
||||
# case and spacing should not matter
|
||||
form_data = {'embargoed_countries': 'cu, Sy , US'}
|
||||
form = EmbargoedStateForm(data=form_data)
|
||||
self.assertTrue(form.is_valid())
|
||||
form.save()
|
||||
current_embargoes = EmbargoedState.current().embargoed_countries_list
|
||||
for country in ["CU", "SY", "US"]:
|
||||
self.assertIn(country, current_embargoes)
|
||||
# Test clearing by adding an empty list is OK too
|
||||
form_data = {'embargoed_countries': ''}
|
||||
form = EmbargoedStateForm(data=form_data)
|
||||
self.assertTrue(form.is_valid())
|
||||
form.save()
|
||||
self.assertTrue(len(EmbargoedState.current().embargoed_countries_list) == 0)
|
||||
|
||||
|
||||
def test_add_invalid_states(self):
|
||||
# test adding invalid codes
|
||||
# xx is not valid
|
||||
# usa is not valid
|
||||
form_data = {'embargoed_countries': 'usa, xx'}
|
||||
form = EmbargoedStateForm(data=form_data)
|
||||
self.assertFalse(form.is_valid())
|
||||
|
||||
msg = 'COULD NOT PARSE COUNTRY CODE(S) FOR: {0}'.format([u'USA', u'XX'])
|
||||
msg += ' Please check the list of country codes and verify your entries.'
|
||||
self.assertEquals(msg, form._errors['embargoed_countries'][0]) # pylint: disable=protected-access
|
||||
|
||||
with self.assertRaisesRegexp(ValueError, "The EmbargoedState could not be created because the data didn't validate."):
|
||||
form.save()
|
||||
|
||||
self.assertFalse('USA' in EmbargoedState.current().embargoed_countries_list)
|
||||
self.assertFalse('XX' in EmbargoedState.current().embargoed_countries_list)
|
||||
|
||||
|
||||
class IPFilterFormTest(TestCase):
|
||||
"""Test form for adding [black|white]list IP addresses"""
|
||||
|
||||
def tearDown(self):
|
||||
# Explicitly clear ConfigurationModel's cache so tests have a clear cache
|
||||
# and don't interfere with each other
|
||||
cache.clear()
|
||||
|
||||
def test_add_valid_ips(self):
|
||||
# test adding valid ip addresses
|
||||
# should be able to do both ipv4 and ipv6
|
||||
# spacing should not matter
|
||||
form_data = {
|
||||
'whitelist': '127.0.0.1, 2003:dead:beef:4dad:23:46:bb:101',
|
||||
'blacklist': ' 18.244.1.5 , 2002:c0a8:101::42, 18.36.22.1'
|
||||
}
|
||||
form = IPFilterForm(data=form_data)
|
||||
self.assertTrue(form.is_valid())
|
||||
form.save()
|
||||
whitelist = IPFilter.current().whitelist_ips
|
||||
blacklist = IPFilter.current().blacklist_ips
|
||||
for addr in '127.0.0.1, 2003:dead:beef:4dad:23:46:bb:101'.split(','):
|
||||
self.assertIn(addr.strip(), whitelist)
|
||||
for addr in '18.244.1.5, 2002:c0a8:101::42, 18.36.22.1'.split(','):
|
||||
self.assertIn(addr.strip(), blacklist)
|
||||
|
||||
# Test clearing by adding an empty list is OK too
|
||||
form_data = {
|
||||
'whitelist': '',
|
||||
'blacklist': ''
|
||||
}
|
||||
form = IPFilterForm(data=form_data)
|
||||
self.assertTrue(form.is_valid())
|
||||
form.save()
|
||||
self.assertTrue(len(IPFilter.current().whitelist) == 0)
|
||||
self.assertTrue(len(IPFilter.current().blacklist) == 0)
|
||||
|
||||
def test_add_invalid_ips(self):
|
||||
# test adding invalid ip addresses
|
||||
form_data = {
|
||||
'whitelist': '.0.0.1, :dead:beef:::',
|
||||
'blacklist': ' 18.244.* , 999999:c0a8:101::42'
|
||||
}
|
||||
form = IPFilterForm(data=form_data)
|
||||
self.assertFalse(form.is_valid())
|
||||
|
||||
wmsg = "Invalid IP Address(es): [u'.0.0.1', u':dead:beef:::'] Please fix the error(s) and try again."
|
||||
self.assertEquals(wmsg, form._errors['whitelist'][0]) # pylint: disable=protected-access
|
||||
bmsg = "Invalid IP Address(es): [u'18.244.*', u'999999:c0a8:101::42'] Please fix the error(s) and try again."
|
||||
self.assertEquals(bmsg, form._errors['blacklist'][0]) # pylint: disable=protected-access
|
||||
|
||||
with self.assertRaisesRegexp(ValueError, "The IPFilter could not be created because the data didn't validate."):
|
||||
form.save()
|
||||
132
common/djangoapps/embargo/tests/test_middleware.py
Normal file
132
common/djangoapps/embargo/tests/test_middleware.py
Normal file
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
Tests for EmbargoMiddleware
|
||||
"""
|
||||
|
||||
import mock
|
||||
import pygeoip
|
||||
import unittest
|
||||
|
||||
from django.conf import settings
|
||||
from django.test import TestCase, Client
|
||||
from django.test.utils import override_settings
|
||||
from courseware.tests.tests import TEST_DATA_MONGO_MODULESTORE
|
||||
from student.models import CourseEnrollment
|
||||
from student.tests.factories import UserFactory
|
||||
from xmodule.modulestore.tests.factories import CourseFactory
|
||||
|
||||
# Explicitly import the cache from ConfigurationModel so we can reset it after each test
|
||||
from config_models.models import cache
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPFilter
|
||||
|
||||
|
||||
@override_settings(MODULESTORE=TEST_DATA_MONGO_MODULESTORE)
|
||||
class EmbargoMiddlewareTests(TestCase):
|
||||
"""
|
||||
Tests of EmbargoMiddleware
|
||||
"""
|
||||
def setUp(self):
|
||||
self.client = Client()
|
||||
self.user = UserFactory(username='fred', password='secret')
|
||||
self.client.login(username='fred', password='secret')
|
||||
self.embargo_course = CourseFactory.create()
|
||||
self.embargo_course.save()
|
||||
self.regular_course = CourseFactory.create(org="Regular")
|
||||
self.regular_course.save()
|
||||
self.embargoed_page = '/courses/' + self.embargo_course.id + '/info'
|
||||
self.regular_page = '/courses/' + self.regular_course.id + '/info'
|
||||
EmbargoedCourse(course_id=self.embargo_course.id, embargoed=True).save()
|
||||
EmbargoedState(
|
||||
embargoed_countries="cu, ir, Sy, SD",
|
||||
changed_by=self.user,
|
||||
enabled=True
|
||||
).save()
|
||||
CourseEnrollment.enroll(self.user, self.regular_course.id)
|
||||
CourseEnrollment.enroll(self.user, self.embargo_course.id)
|
||||
# Text from lms/templates/static_templates/embargo.html
|
||||
self.embargo_text = "Unfortunately, at this time edX must comply with export controls, and we cannot allow you to access this particular course."
|
||||
|
||||
def tearDown(self):
|
||||
# Explicitly clear ConfigurationModel's cache so tests have a clear cache
|
||||
# and don't interfere with each other
|
||||
cache.clear()
|
||||
|
||||
def mock_country_code_by_addr(self, ip_addr):
|
||||
"""
|
||||
Gives us a fake set of IPs
|
||||
"""
|
||||
ip_dict = {
|
||||
'1.0.0.0': 'CU',
|
||||
'2.0.0.0': 'IR',
|
||||
'3.0.0.0': 'SY',
|
||||
'4.0.0.0': 'SD',
|
||||
'5.0.0.0': 'AQ', # Antartica
|
||||
}
|
||||
return ip_dict.get(ip_addr, 'US')
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
def test_countries(self):
|
||||
with mock.patch.object(pygeoip.GeoIP, 'country_code_by_addr') as mocked_method:
|
||||
mocked_method.side_effect = self.mock_country_code_by_addr
|
||||
|
||||
# Accessing an embargoed page from a blocked IP should cause a redirect
|
||||
response = self.client.get(self.embargoed_page, HTTP_X_FORWARDED_FOR='1.0.0.0', REMOTE_ADDR='1.0.0.0')
|
||||
self.assertEqual(response.status_code, 302)
|
||||
# Following the redirect should give us the embargo page
|
||||
response = self.client.get(
|
||||
self.embargoed_page,
|
||||
HTTP_X_FORWARDED_FOR='1.0.0.0',
|
||||
REMOTE_ADDR='1.0.0.0',
|
||||
follow=True
|
||||
)
|
||||
self.assertIn(self.embargo_text, response.content)
|
||||
|
||||
# Accessing a regular page from a blocked IP should succeed
|
||||
response = self.client.get(self.regular_page, HTTP_X_FORWARDED_FOR='1.0.0.0', REMOTE_ADDR='1.0.0.0')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
# Accessing an embargoed page from a non-embargoed IP should succeed
|
||||
response = self.client.get(self.embargoed_page, HTTP_X_FORWARDED_FOR='5.0.0.0', REMOTE_ADDR='5.0.0.0')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
# Accessing a regular page from a non-embargoed IP should succeed
|
||||
response = self.client.get(self.regular_page, HTTP_X_FORWARDED_FOR='5.0.0.0', REMOTE_ADDR='5.0.0.0')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
@unittest.skipUnless(settings.ROOT_URLCONF == 'lms.urls', 'Test only valid in lms')
|
||||
def test_ip_exceptions(self):
|
||||
# Explicitly whitelist/blacklist some IPs
|
||||
IPFilter(
|
||||
whitelist='1.0.0.0',
|
||||
blacklist='5.0.0.0',
|
||||
changed_by=self.user,
|
||||
enabled=True
|
||||
).save()
|
||||
|
||||
with mock.patch.object(pygeoip.GeoIP, 'country_code_by_addr') as mocked_method:
|
||||
mocked_method.side_effect = self.mock_country_code_by_addr
|
||||
|
||||
# Accessing an embargoed page from a blocked IP that's been whitelisted
|
||||
# should succeed
|
||||
response = self.client.get(self.embargoed_page, HTTP_X_FORWARDED_FOR='1.0.0.0', REMOTE_ADDR='1.0.0.0')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
# Accessing a regular course from a blocked IP that's been whitelisted should succeed
|
||||
response = self.client.get(self.regular_page, HTTP_X_FORWARDED_FOR='1.0.0.0', REMOTE_ADDR='1.0.0.0')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
# Accessing an embargoed course from non-embargoed IP that's been blacklisted
|
||||
# should cause a redirect
|
||||
response = self.client.get(self.embargoed_page, HTTP_X_FORWARDED_FOR='5.0.0.0', REMOTE_ADDR='5.0.0.0')
|
||||
self.assertEqual(response.status_code, 302)
|
||||
# Following the redirect should give us the embargo page
|
||||
response = self.client.get(
|
||||
self.embargoed_page,
|
||||
HTTP_X_FORWARDED_FOR='5.0.0.0',
|
||||
REMOTE_ADDR='1.0.0.0',
|
||||
follow=True
|
||||
)
|
||||
self.assertIn(self.embargo_text, response.content)
|
||||
|
||||
# Accessing a regular course from a non-embargoed IP that's been blacklisted should succeed
|
||||
response = self.client.get(self.regular_page, HTTP_X_FORWARDED_FOR='5.0.0.0', REMOTE_ADDR='5.0.0.0')
|
||||
self.assertEqual(response.status_code, 200)
|
||||
80
common/djangoapps/embargo/tests/test_models.py
Normal file
80
common/djangoapps/embargo/tests/test_models.py
Normal file
@@ -0,0 +1,80 @@
|
||||
"""Test of models for embargo middleware app"""
|
||||
from django.test import TestCase
|
||||
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState, IPFilter
|
||||
|
||||
|
||||
class EmbargoModelsTest(TestCase):
|
||||
"""Test each of the 3 models in embargo.models"""
|
||||
def test_course_embargo(self):
|
||||
course_id = 'abc/123/doremi'
|
||||
# Test that course is not authorized by default
|
||||
self.assertFalse(EmbargoedCourse.is_embargoed(course_id))
|
||||
|
||||
# Authorize
|
||||
cauth = EmbargoedCourse(course_id=course_id, embargoed=True)
|
||||
cauth.save()
|
||||
|
||||
# Now, course should be embargoed
|
||||
self.assertTrue(EmbargoedCourse.is_embargoed(course_id))
|
||||
self.assertEquals(
|
||||
cauth.__unicode__(),
|
||||
"Course 'abc/123/doremi' is Embargoed"
|
||||
)
|
||||
|
||||
# Unauthorize by explicitly setting email_enabled to False
|
||||
cauth.embargoed = False
|
||||
cauth.save()
|
||||
# Test that course is now unauthorized
|
||||
self.assertFalse(EmbargoedCourse.is_embargoed(course_id))
|
||||
self.assertEquals(
|
||||
cauth.__unicode__(),
|
||||
"Course 'abc/123/doremi' is Not Embargoed"
|
||||
)
|
||||
|
||||
def test_state_embargo(self):
|
||||
# Azerbaijan and France should not be blocked
|
||||
good_states = ['AZ', 'FR']
|
||||
# Gah block USA and Antartica
|
||||
blocked_states = ['US', 'AQ']
|
||||
currently_blocked = EmbargoedState.current().embargoed_countries_list
|
||||
|
||||
for state in blocked_states + good_states:
|
||||
self.assertFalse(state in currently_blocked)
|
||||
|
||||
# Block
|
||||
cauth = EmbargoedState(embargoed_countries='US, AQ')
|
||||
cauth.save()
|
||||
currently_blocked = EmbargoedState.current().embargoed_countries_list
|
||||
|
||||
for state in good_states:
|
||||
self.assertFalse(state in currently_blocked)
|
||||
for state in blocked_states:
|
||||
self.assertTrue(state in currently_blocked)
|
||||
|
||||
# Change embargo - block Isle of Man too
|
||||
blocked_states.append('IM')
|
||||
cauth.embargoed_countries = 'US, AQ, IM'
|
||||
cauth.save()
|
||||
currently_blocked = EmbargoedState.current().embargoed_countries_list
|
||||
|
||||
for state in good_states:
|
||||
self.assertFalse(state in currently_blocked)
|
||||
for state in blocked_states:
|
||||
self.assertTrue(state in currently_blocked)
|
||||
|
||||
def test_ip_blocking(self):
|
||||
whitelist = '127.0.0.1'
|
||||
blacklist = '18.244.51.3'
|
||||
|
||||
cwhitelist = IPFilter.current().whitelist_ips
|
||||
self.assertFalse(whitelist in cwhitelist)
|
||||
cblacklist = IPFilter.current().blacklist_ips
|
||||
self.assertFalse(blacklist in cblacklist)
|
||||
|
||||
IPFilter(whitelist=whitelist, blacklist=blacklist).save()
|
||||
|
||||
cwhitelist = IPFilter.current().whitelist_ips
|
||||
self.assertTrue(whitelist in cwhitelist)
|
||||
cblacklist = IPFilter.current().blacklist_ips
|
||||
self.assertTrue(blacklist in cblacklist)
|
||||
@@ -1,71 +0,0 @@
|
||||
"""
|
||||
Tests for EmbargoMiddleware
|
||||
"""
|
||||
|
||||
from xmodule.modulestore.tests.factories import CourseFactory
|
||||
from django.test import TestCase
|
||||
from django.test.utils import override_settings
|
||||
from courseware.tests.tests import TEST_DATA_MONGO_MODULESTORE
|
||||
from embargo.models import EmbargoedCourse, EmbargoedState
|
||||
from django.test import Client
|
||||
from student.models import CourseEnrollment
|
||||
from student.tests.factories import UserFactory
|
||||
import mock
|
||||
import pygeoip
|
||||
|
||||
|
||||
@override_settings(MODULESTORE=TEST_DATA_MONGO_MODULESTORE)
|
||||
class EmbargoMiddlewareTests(TestCase):
|
||||
"""
|
||||
Tests of EmbargoMiddleware
|
||||
"""
|
||||
def setUp(self):
|
||||
self.client = Client()
|
||||
self.user = UserFactory(username='fred', password='secret')
|
||||
self.client.login(username='fred', password='secret')
|
||||
self.embargo_course = CourseFactory.create()
|
||||
self.embargo_course.save()
|
||||
self.regular_course = CourseFactory.create(org="Regular")
|
||||
self.regular_course.save()
|
||||
self.embargoed_page = '/courses/' + self.embargo_course.id + '/info'
|
||||
self.regular_page = '/courses/' + self.regular_course.id + '/info'
|
||||
EmbargoedCourse(course_id=self.embargo_course.id, embargoed=True).save()
|
||||
EmbargoedState(
|
||||
embargoed_countries="cu, ir, Sy, SD",
|
||||
changed_by=self.user,
|
||||
enabled=True
|
||||
).save()
|
||||
# TODO need to set up & test whitelist/blacklist IPs (IPException model)
|
||||
CourseEnrollment.enroll(self.user, self.regular_course.id)
|
||||
CourseEnrollment.enroll(self.user, self.embargo_course.id)
|
||||
|
||||
def test_countries(self):
|
||||
def mock_country_code_by_addr(ip_addr):
|
||||
"""
|
||||
Gives us a fake set of IPs
|
||||
"""
|
||||
ip_dict = {
|
||||
'1.0.0.0': 'CU',
|
||||
'2.0.0.0': 'IR',
|
||||
'3.0.0.0': 'SY',
|
||||
'4.0.0.0': 'SD',
|
||||
}
|
||||
return ip_dict.get(ip_addr, 'US')
|
||||
|
||||
with mock.patch.object(pygeoip.GeoIP, 'country_code_by_addr') as mocked_method:
|
||||
mocked_method.side_effect = mock_country_code_by_addr
|
||||
|
||||
# Accessing an embargoed page from a blocked IP should cause a redirect
|
||||
response = self.client.get(self.embargoed_page, HTTP_X_FORWARDED_FOR='1.0.0.0', REMOTE_ADDR='1.0.0.0')
|
||||
self.assertEqual(response.status_code, 302)
|
||||
|
||||
# Accessing a regular course from a blocked IP should succeed
|
||||
response = self.client.get(self.regular_page, HTTP_X_FORWARDED_FOR='1.0.0.0', REMOTE_ADDR='1.0.0.0')
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
# Accessing any course from non-embaroged IPs should succeed
|
||||
response = self.client.get(self.regular_page, HTTP_X_FORWARDED_FOR='5.0.0.0', REMOTE_ADDR='5.0.0.0')
|
||||
self.assertEqual(response.status_code, 404)
|
||||
|
||||
response = self.client.get(self.embargoed_page, HTTP_X_FORWARDED_FOR='5.0.0.0', REMOTE_ADDR='5.0.0.0')
|
||||
self.assertEqual(response.status_code, 404)
|
||||
Reference in New Issue
Block a user