refactor: Ran pyupgrade on openedx/core/djangoapps/embargo (#26911)
This commit is contained in:
@@ -84,9 +84,9 @@ def check_course_access(course_key, user=None, ip_address=None, url=None):
|
||||
if not CountryAccessRule.check_country_access(course_key, user_country_from_ip):
|
||||
log.info(
|
||||
(
|
||||
u"Blocking user %s from accessing course %s at %s "
|
||||
u"because the user's IP address %s appears to be "
|
||||
u"located in %s."
|
||||
"Blocking user %s from accessing course %s at %s "
|
||||
"because the user's IP address %s appears to be "
|
||||
"located in %s."
|
||||
),
|
||||
getattr(user, 'id', '<Not Authenticated>'),
|
||||
course_key,
|
||||
@@ -104,8 +104,8 @@ def check_course_access(course_key, user=None, ip_address=None, url=None):
|
||||
if not CountryAccessRule.check_country_access(course_key, user_country_from_profile):
|
||||
log.info(
|
||||
(
|
||||
u"Blocking user %s from accessing course %s at %s "
|
||||
u"because the user's profile country is %s."
|
||||
"Blocking user %s from accessing course %s at %s "
|
||||
"because the user's profile country is %s."
|
||||
),
|
||||
user.id, course_key, url, user_country_from_profile
|
||||
)
|
||||
@@ -146,7 +146,7 @@ def _get_user_country_from_profile(user):
|
||||
user country from profile.
|
||||
|
||||
"""
|
||||
cache_key = u'user.{user_id}.profile.country'.format(user_id=user.id)
|
||||
cache_key = f'user.{user.id}.profile.country'
|
||||
profile_country = cache.get(cache_key)
|
||||
if profile_country is None:
|
||||
profile = getattr(user, 'profile', None)
|
||||
@@ -203,7 +203,7 @@ def get_embargo_response(request, course_id, user):
|
||||
status=status.HTTP_403_FORBIDDEN,
|
||||
data={
|
||||
"message": (
|
||||
u"Users from this location cannot access the course '{course_id}'."
|
||||
"Users from this location cannot access the course '{course_id}'."
|
||||
).format(course_id=course_id),
|
||||
"user_message_url": request.build_absolute_uri(redirect_url)
|
||||
}
|
||||
|
||||
@@ -6,6 +6,6 @@ class InvalidAccessPoint(Exception):
|
||||
|
||||
def __init__(self, access_point, *args, **kwargs):
|
||||
msg = (
|
||||
u"Access point '{access_point}' should be either 'enrollment' or 'courseware'"
|
||||
"Access point '{access_point}' should be either 'enrollment' or 'courseware'"
|
||||
).format(access_point=access_point)
|
||||
super(InvalidAccessPoint, self).__init__(msg, *args, **kwargs) # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().__init__(msg, *args, **kwargs)
|
||||
|
||||
@@ -3,7 +3,7 @@ List of valid ISO 3166-1 Alpha-2 country codes, used for
|
||||
validating entries on entered country codes on django-admin page.
|
||||
"""
|
||||
|
||||
COUNTRY_CODES = set([
|
||||
COUNTRY_CODES = {
|
||||
"AC", "AD", "AE", "AF", "AG", "AI", "AL", "AM", "AN", "AO", "AQ", "AR", "AS", "AT",
|
||||
"AU", "AW", "AX", "AZ", "BA", "BB", "BD", "BE", "BF", "BG", "BH", "BI", "BJ", "BM",
|
||||
"BN", "BO", "BR", "BS", "BT", "BV", "BW", "BY", "BZ", "CA", "CC", "CD", "CF", "CG",
|
||||
@@ -22,4 +22,4 @@ COUNTRY_CODES = set([
|
||||
"TF", "TG", "TH", "TJ", "TK", "TL", "TM", "TN", "TO", "TR", "TT", "TV", "TW", "TZ",
|
||||
"UA", "UG", "UM", "US", "UY", "UZ", "VA", "VC", "VE", "VG", "VI", "VN", "VU", "WF",
|
||||
"WS", "YE", "YT", "ZA", "ZM", "ZW"
|
||||
])
|
||||
}
|
||||
|
||||
@@ -25,7 +25,7 @@ class RestrictedCourseForm(forms.ModelForm):
|
||||
error message instead.
|
||||
|
||||
"""
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
model = RestrictedCourse
|
||||
fields = '__all__'
|
||||
|
||||
@@ -59,7 +59,7 @@ class RestrictedCourseForm(forms.ModelForm):
|
||||
class IPFilterForm(forms.ModelForm):
|
||||
"""Form validating entry of IP addresses"""
|
||||
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
model = IPFilter
|
||||
fields = '__all__'
|
||||
|
||||
@@ -86,7 +86,7 @@ class IPFilterForm(forms.ModelForm):
|
||||
if not self._is_valid_ip(address):
|
||||
error_addresses.append(address)
|
||||
if error_addresses:
|
||||
msg = u'Invalid IP Address(es): {0}'.format(error_addresses)
|
||||
msg = f'Invalid IP Address(es): {error_addresses}'
|
||||
msg += ' Please fix the error(s) and try again.'
|
||||
raise forms.ValidationError(msg)
|
||||
|
||||
|
||||
@@ -62,7 +62,7 @@ class EmbargoMiddleware(MiddlewareMixin):
|
||||
# If embargoing is turned off, make this middleware do nothing
|
||||
if not settings.FEATURES.get('EMBARGO'):
|
||||
raise MiddlewareNotUsed()
|
||||
super(EmbargoMiddleware, self).__init__(*args, **kwargs) # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def process_request(self, request):
|
||||
"""Block requests based on embargo rules.
|
||||
@@ -89,8 +89,8 @@ class EmbargoMiddleware(MiddlewareMixin):
|
||||
if ip_filter.enabled and ip_address in ip_filter.blacklist_ips:
|
||||
log.info(
|
||||
(
|
||||
u"User %s was blocked from accessing %s "
|
||||
u"because IP address %s is blacklisted."
|
||||
"User %s was blocked from accessing %s "
|
||||
"because IP address %s is blacklisted."
|
||||
), request.user.id, request.path, ip_address
|
||||
)
|
||||
|
||||
@@ -108,8 +108,8 @@ class EmbargoMiddleware(MiddlewareMixin):
|
||||
elif ip_filter.enabled and ip_address in ip_filter.whitelist_ips:
|
||||
log.info(
|
||||
(
|
||||
u"User %s was allowed access to %s because "
|
||||
u"IP address %s is whitelisted."
|
||||
"User %s was allowed access to %s because "
|
||||
"IP address %s is whitelisted."
|
||||
),
|
||||
request.user.id, request.path, ip_address
|
||||
)
|
||||
|
||||
@@ -1,6 +1,3 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
|
||||
from django.db import migrations, models
|
||||
import django_countries.fields
|
||||
import django.db.models.deletion
|
||||
@@ -29,7 +26,7 @@ class Migration(migrations.Migration):
|
||||
name='CountryAccessRule',
|
||||
fields=[
|
||||
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
|
||||
('rule_type', models.CharField(default=u'blacklist', help_text='Whether to include or exclude the given course. If whitelist countries are specified, then ONLY users from whitelisted countries will be able to access the course. If blacklist countries are specified, then users from blacklisted countries will NOT be able to access the course.', max_length=255, choices=[(u'whitelist', u'Whitelist (allow only these countries)'), (u'blacklist', u'Blacklist (block these countries)')])),
|
||||
('rule_type', models.CharField(default='blacklist', help_text='Whether to include or exclude the given course. If whitelist countries are specified, then ONLY users from whitelisted countries will be able to access the course. If blacklist countries are specified, then users from blacklisted countries will NOT be able to access the course.', max_length=255, choices=[('whitelist', 'Whitelist (allow only these countries)'), ('blacklist', 'Blacklist (block these countries)')])),
|
||||
('country', models.ForeignKey(help_text='The country to which this rule applies.', to='embargo.Country', on_delete=models.CASCADE)),
|
||||
],
|
||||
),
|
||||
@@ -59,7 +56,7 @@ class Migration(migrations.Migration):
|
||||
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
|
||||
('change_date', models.DateTimeField(auto_now_add=True, verbose_name='Change date')),
|
||||
('enabled', models.BooleanField(default=False, verbose_name='Enabled')),
|
||||
('embargoed_countries', models.TextField(help_text=u'A comma-separated list of country codes that fall under U.S. embargo restrictions', blank=True)),
|
||||
('embargoed_countries', models.TextField(help_text='A comma-separated list of country codes that fall under U.S. embargo restrictions', blank=True)),
|
||||
('changed_by', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, editable=False, to=settings.AUTH_USER_MODEL, null=True, verbose_name='Changed by')),
|
||||
],
|
||||
options={
|
||||
@@ -73,8 +70,8 @@ class Migration(migrations.Migration):
|
||||
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
|
||||
('change_date', models.DateTimeField(auto_now_add=True, verbose_name='Change date')),
|
||||
('enabled', models.BooleanField(default=False, verbose_name='Enabled')),
|
||||
('whitelist', models.TextField(help_text=u'A comma-separated list of IP addresses that should not fall under embargo restrictions.', blank=True)),
|
||||
('blacklist', models.TextField(help_text=u'A comma-separated list of IP addresses that should fall under embargo restrictions.', blank=True)),
|
||||
('whitelist', models.TextField(help_text='A comma-separated list of IP addresses that should not fall under embargo restrictions.', blank=True)),
|
||||
('blacklist', models.TextField(help_text='A comma-separated list of IP addresses that should fall under embargo restrictions.', blank=True)),
|
||||
('changed_by', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, editable=False, to=settings.AUTH_USER_MODEL, null=True, verbose_name='Changed by')),
|
||||
],
|
||||
options={
|
||||
@@ -87,8 +84,8 @@ class Migration(migrations.Migration):
|
||||
fields=[
|
||||
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
|
||||
('course_key', CourseKeyField(help_text='The course key for the restricted course.', unique=True, max_length=255, db_index=True)),
|
||||
('enroll_msg_key', models.CharField(default=u'default', help_text=u'The message to show when a user is blocked from enrollment.', max_length=255, choices=[(u'default', u'Default'), (u'embargo', u'Embargo')])),
|
||||
('access_msg_key', models.CharField(default=u'default', help_text=u'The message to show when a user is blocked from accessing a course.', max_length=255, choices=[(u'default', u'Default'), (u'embargo', u'Embargo')])),
|
||||
('enroll_msg_key', models.CharField(default='default', help_text='The message to show when a user is blocked from enrollment.', max_length=255, choices=[('default', 'Default'), ('embargo', 'Embargo')])),
|
||||
('access_msg_key', models.CharField(default='default', help_text='The message to show when a user is blocked from accessing a course.', max_length=255, choices=[('default', 'Default'), ('embargo', 'Embargo')])),
|
||||
('disable_access_check', models.BooleanField(default=False, help_text='Allow users who enrolled in an allowed country to access restricted courses from excluded countries.')),
|
||||
],
|
||||
),
|
||||
@@ -99,6 +96,6 @@ class Migration(migrations.Migration):
|
||||
),
|
||||
migrations.AlterUniqueTogether(
|
||||
name='countryaccessrule',
|
||||
unique_together=set([('restricted_course', 'country')]),
|
||||
unique_together={('restricted_course', 'country')},
|
||||
),
|
||||
]
|
||||
|
||||
@@ -1,6 +1,3 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
|
||||
# Converted from the original South migration 0003_add_countries.py
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
@@ -16,7 +16,6 @@ import ipaddress
|
||||
import json
|
||||
import logging
|
||||
|
||||
import six
|
||||
from django.utils.encoding import python_2_unicode_compatible
|
||||
from config_models.models import ConfigurationModel
|
||||
from django.core.cache import cache
|
||||
@@ -28,7 +27,6 @@ from django.utils.translation import ugettext_lazy
|
||||
from django_countries import countries
|
||||
from django_countries.fields import CountryField
|
||||
from opaque_keys.edx.django.models import CourseKeyField
|
||||
from six import text_type
|
||||
|
||||
from openedx.core.djangoapps.xmodule_django.models import NoneToEmptyManager
|
||||
|
||||
@@ -72,7 +70,7 @@ class EmbargoedCourse(models.Model):
|
||||
not_em = "Not "
|
||||
if self.embargoed:
|
||||
not_em = ""
|
||||
return u"Course '{}' is {}Embargoed".format(text_type(self.course_id), not_em)
|
||||
return "Course '{}' is {}Embargoed".format(str(self.course_id), not_em)
|
||||
|
||||
|
||||
@python_2_unicode_compatible
|
||||
@@ -87,7 +85,7 @@ class EmbargoedState(ConfigurationModel):
|
||||
# The countries to embargo
|
||||
embargoed_countries = models.TextField(
|
||||
blank=True,
|
||||
help_text=u"A comma-separated list of country codes that fall under U.S. embargo restrictions"
|
||||
help_text="A comma-separated list of country codes that fall under U.S. embargo restrictions"
|
||||
)
|
||||
|
||||
@property
|
||||
@@ -128,37 +126,37 @@ class RestrictedCourse(models.Model):
|
||||
|
||||
ENROLL_MSG_KEY_CHOICES = tuple(sorted([
|
||||
(msg_key, msg.description)
|
||||
for msg_key, msg in six.iteritems(ENROLL_MESSAGES)
|
||||
for msg_key, msg in ENROLL_MESSAGES.items()
|
||||
]))
|
||||
|
||||
COURSEWARE_MSG_KEY_CHOICES = tuple(sorted([
|
||||
(msg_key, msg.description)
|
||||
for msg_key, msg in six.iteritems(COURSEWARE_MESSAGES)
|
||||
for msg_key, msg in COURSEWARE_MESSAGES.items()
|
||||
]))
|
||||
|
||||
course_key = CourseKeyField(
|
||||
max_length=255, db_index=True, unique=True,
|
||||
help_text=ugettext_lazy(u"The course key for the restricted course.")
|
||||
help_text=ugettext_lazy("The course key for the restricted course.")
|
||||
)
|
||||
|
||||
enroll_msg_key = models.CharField(
|
||||
max_length=255,
|
||||
choices=ENROLL_MSG_KEY_CHOICES,
|
||||
default=u'default',
|
||||
help_text=ugettext_lazy(u"The message to show when a user is blocked from enrollment.")
|
||||
default='default',
|
||||
help_text=ugettext_lazy("The message to show when a user is blocked from enrollment.")
|
||||
)
|
||||
|
||||
access_msg_key = models.CharField(
|
||||
max_length=255,
|
||||
choices=COURSEWARE_MSG_KEY_CHOICES,
|
||||
default=u'default',
|
||||
help_text=ugettext_lazy(u"The message to show when a user is blocked from accessing a course.")
|
||||
default='default',
|
||||
help_text=ugettext_lazy("The message to show when a user is blocked from accessing a course.")
|
||||
)
|
||||
|
||||
disable_access_check = models.BooleanField(
|
||||
default=False,
|
||||
help_text=ugettext_lazy(
|
||||
u"Allow users who enrolled in an allowed country "
|
||||
"Allow users who enrolled in an allowed country "
|
||||
"to access restricted courses from excluded countries."
|
||||
)
|
||||
)
|
||||
@@ -175,7 +173,7 @@ class RestrictedCourse(models.Model):
|
||||
Boolean
|
||||
True if course is in restricted course list.
|
||||
"""
|
||||
return six.text_type(course_id) in cls._get_restricted_courses_from_cache()
|
||||
return str(course_id) in cls._get_restricted_courses_from_cache()
|
||||
|
||||
@classmethod
|
||||
def is_disabled_access_check(cls, course_id):
|
||||
@@ -193,8 +191,8 @@ class RestrictedCourse(models.Model):
|
||||
# checking is_restricted_course method also here to make sure course exists in the list otherwise in case of
|
||||
# no course found it will throw the key not found error on 'disable_access_check'
|
||||
return (
|
||||
cls.is_restricted_course(six.text_type(course_id))
|
||||
and cls._get_restricted_courses_from_cache().get(six.text_type(course_id))["disable_access_check"]
|
||||
cls.is_restricted_course(str(course_id))
|
||||
and cls._get_restricted_courses_from_cache().get(str(course_id))["disable_access_check"]
|
||||
)
|
||||
|
||||
@classmethod
|
||||
@@ -205,7 +203,7 @@ class RestrictedCourse(models.Model):
|
||||
restricted_courses = cache.get(cls.COURSE_LIST_CACHE_KEY)
|
||||
if restricted_courses is None:
|
||||
restricted_courses = {
|
||||
six.text_type(course.course_key): {
|
||||
str(course.course_key): {
|
||||
'disable_access_check': course.disable_access_check
|
||||
}
|
||||
for course in RestrictedCourse.objects.all()
|
||||
@@ -243,7 +241,7 @@ class RestrictedCourse(models.Model):
|
||||
'access_msg': self.access_msg_key,
|
||||
'country_rules': [
|
||||
{
|
||||
'country': six.text_type(rule.country.country),
|
||||
'country': str(rule.country.country),
|
||||
'rule_type': rule.rule_type
|
||||
}
|
||||
for rule in country_rules_for_course
|
||||
@@ -271,7 +269,7 @@ class RestrictedCourse(models.Model):
|
||||
return self.access_msg_key
|
||||
|
||||
def __str__(self):
|
||||
return six.text_type(self.course_key)
|
||||
return str(self.course_key)
|
||||
|
||||
@classmethod
|
||||
def message_url_path(cls, course_key, access_point):
|
||||
@@ -387,16 +385,16 @@ class Country(models.Model):
|
||||
"""
|
||||
country = CountryField(
|
||||
db_index=True, unique=True,
|
||||
help_text=ugettext_lazy(u"Two character ISO country code.")
|
||||
help_text=ugettext_lazy("Two character ISO country code.")
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return u"{name} ({code})".format(
|
||||
name=six.text_type(self.country.name),
|
||||
code=six.text_type(self.country)
|
||||
return "{name} ({code})".format(
|
||||
name=str(self.country.name),
|
||||
code=str(self.country)
|
||||
)
|
||||
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
"""Default ordering is ascending by country code """
|
||||
ordering = ['country']
|
||||
|
||||
@@ -421,12 +419,12 @@ class CountryAccessRule(models.Model):
|
||||
.. no_pii:
|
||||
"""
|
||||
|
||||
WHITELIST_RULE = u'whitelist'
|
||||
BLACKLIST_RULE = u'blacklist'
|
||||
WHITELIST_RULE = 'whitelist'
|
||||
BLACKLIST_RULE = 'blacklist'
|
||||
|
||||
RULE_TYPE_CHOICES = (
|
||||
(WHITELIST_RULE, u'Whitelist (allow only these countries)'),
|
||||
(BLACKLIST_RULE, u'Blacklist (block these countries)'),
|
||||
(WHITELIST_RULE, 'Whitelist (allow only these countries)'),
|
||||
(BLACKLIST_RULE, 'Blacklist (block these countries)'),
|
||||
)
|
||||
|
||||
rule_type = models.CharField(
|
||||
@@ -434,28 +432,28 @@ class CountryAccessRule(models.Model):
|
||||
choices=RULE_TYPE_CHOICES,
|
||||
default=BLACKLIST_RULE,
|
||||
help_text=ugettext_lazy(
|
||||
u"Whether to include or exclude the given course. "
|
||||
u"If whitelist countries are specified, then ONLY users from whitelisted countries "
|
||||
u"will be able to access the course. If blacklist countries are specified, then "
|
||||
u"users from blacklisted countries will NOT be able to access the course."
|
||||
"Whether to include or exclude the given course. "
|
||||
"If whitelist countries are specified, then ONLY users from whitelisted countries "
|
||||
"will be able to access the course. If blacklist countries are specified, then "
|
||||
"users from blacklisted countries will NOT be able to access the course."
|
||||
)
|
||||
)
|
||||
|
||||
restricted_course = models.ForeignKey(
|
||||
"RestrictedCourse",
|
||||
help_text=ugettext_lazy(u"The course to which this rule applies."),
|
||||
help_text=ugettext_lazy("The course to which this rule applies."),
|
||||
on_delete=models.CASCADE,
|
||||
)
|
||||
|
||||
country = models.ForeignKey(
|
||||
"Country",
|
||||
help_text=ugettext_lazy(u"The country to which this rule applies."),
|
||||
help_text=ugettext_lazy("The country to which this rule applies."),
|
||||
on_delete=models.CASCADE,
|
||||
)
|
||||
|
||||
CACHE_KEY = u"embargo.allowed_countries.{course_key}"
|
||||
CACHE_KEY = "embargo.allowed_countries.{course_key}"
|
||||
|
||||
ALL_COUNTRIES = set(code[0] for code in list(countries))
|
||||
ALL_COUNTRIES = {code[0] for code in list(countries)}
|
||||
|
||||
@classmethod
|
||||
def check_country_access(cls, course_id, country):
|
||||
@@ -526,14 +524,14 @@ class CountryAccessRule(models.Model):
|
||||
|
||||
def __str__(self):
|
||||
if self.rule_type == self.WHITELIST_RULE:
|
||||
return _(u"Whitelist {country} for {course}").format(
|
||||
course=six.text_type(self.restricted_course.course_key),
|
||||
country=six.text_type(self.country),
|
||||
return _("Whitelist {country} for {course}").format(
|
||||
course=str(self.restricted_course.course_key),
|
||||
country=str(self.country),
|
||||
)
|
||||
elif self.rule_type == self.BLACKLIST_RULE:
|
||||
return _(u"Blacklist {country} for {course}").format(
|
||||
course=six.text_type(self.restricted_course.course_key),
|
||||
country=six.text_type(self.country),
|
||||
return _("Blacklist {country} for {course}").format(
|
||||
course=str(self.restricted_course.course_key),
|
||||
country=str(self.country),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
@@ -541,9 +539,9 @@ class CountryAccessRule(models.Model):
|
||||
"""Invalidate the cache. """
|
||||
cache_key = cls.CACHE_KEY.format(course_key=course_key)
|
||||
cache.delete(cache_key)
|
||||
log.info(u"Invalidated country access list for course %s", course_key)
|
||||
log.info("Invalidated country access list for course %s", course_key)
|
||||
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
"""a course can be added with either black or white list. """
|
||||
unique_together = (
|
||||
# This restriction ensures that a country is on
|
||||
@@ -672,7 +670,7 @@ class CourseAccessRuleHistory(models.Model):
|
||||
else:
|
||||
CourseAccessRuleHistory.save_snapshot(restricted_course)
|
||||
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
get_latest_by = 'timestamp'
|
||||
|
||||
|
||||
@@ -693,15 +691,15 @@ class IPFilter(ConfigurationModel):
|
||||
"""
|
||||
whitelist = models.TextField(
|
||||
blank=True,
|
||||
help_text=u"A comma-separated list of IP addresses that should not fall under embargo restrictions."
|
||||
help_text="A comma-separated list of IP addresses that should not fall under embargo restrictions."
|
||||
)
|
||||
|
||||
blacklist = models.TextField(
|
||||
blank=True,
|
||||
help_text=u"A comma-separated list of IP addresses that should fall under embargo restrictions."
|
||||
help_text="A comma-separated list of IP addresses that should fall under embargo restrictions."
|
||||
)
|
||||
|
||||
class IPFilterList(object):
|
||||
class IPFilterList:
|
||||
"""
|
||||
Represent a list of IP addresses with support of networks.
|
||||
"""
|
||||
@@ -710,8 +708,7 @@ class IPFilter(ConfigurationModel):
|
||||
self.networks = [ipaddress.ip_network(ip) for ip in ips]
|
||||
|
||||
def __iter__(self):
|
||||
for network in self.networks:
|
||||
yield network
|
||||
yield from self.networks
|
||||
|
||||
def __contains__(self, ip_addr):
|
||||
try:
|
||||
@@ -744,4 +741,4 @@ class IPFilter(ConfigurationModel):
|
||||
return self.IPFilterList([addr.strip() for addr in self.blacklist.split(',')])
|
||||
|
||||
def __str__(self):
|
||||
return "Whitelist: {} - Blacklist: {}".format(self.whitelist_ips, self.blacklist_ips)
|
||||
return f"Whitelist: {self.whitelist_ips} - Blacklist: {self.blacklist_ips}"
|
||||
|
||||
@@ -1,13 +1,13 @@
|
||||
"""Utilities for writing unit tests that involve course embargos. """
|
||||
|
||||
import contextlib
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import maxminddb
|
||||
from django.core.cache import cache
|
||||
from django.urls import reverse
|
||||
|
||||
import geoip2.database
|
||||
from mock import MagicMock, patch
|
||||
|
||||
from .models import Country, CountryAccessRule, RestrictedCourse
|
||||
|
||||
|
||||
@@ -8,14 +8,14 @@ from ..models import Country, CountryAccessRule, RestrictedCourse
|
||||
|
||||
|
||||
class CountryFactory(DjangoModelFactory):
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
model = Country
|
||||
|
||||
country = 'US'
|
||||
|
||||
|
||||
class RestrictedCourseFactory(DjangoModelFactory): # lint-amnesty, pylint: disable=missing-class-docstring
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
model = RestrictedCourse
|
||||
|
||||
@factory.lazy_attribute
|
||||
@@ -24,7 +24,7 @@ class RestrictedCourseFactory(DjangoModelFactory): # lint-amnesty, pylint: disa
|
||||
|
||||
|
||||
class CountryAccessRuleFactory(DjangoModelFactory): # lint-amnesty, pylint: disable=missing-class-docstring
|
||||
class Meta(object):
|
||||
class Meta:
|
||||
model = CountryAccessRule
|
||||
|
||||
country = factory.SubFactory(CountryFactory)
|
||||
|
||||
@@ -3,13 +3,13 @@ Tests for EmbargoMiddleware
|
||||
"""
|
||||
|
||||
from contextlib import contextmanager
|
||||
from unittest import mock
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import geoip2.database
|
||||
import maxminddb
|
||||
import ddt
|
||||
import pytest
|
||||
import mock
|
||||
from mock import patch, MagicMock
|
||||
|
||||
from django.conf import settings
|
||||
from django.test.utils import override_settings
|
||||
@@ -49,7 +49,7 @@ class EmbargoCheckAccessApiTests(ModuleStoreTestCase):
|
||||
ENABLED_CACHES = ['default', 'mongo_metadata_inheritance', 'loc_cache']
|
||||
|
||||
def setUp(self):
|
||||
super(EmbargoCheckAccessApiTests, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().setUp()
|
||||
self.course = CourseFactory.create()
|
||||
self.user = UserFactory.create()
|
||||
self.restricted_course = RestrictedCourse.objects.create(course_key=self.course.id)
|
||||
@@ -162,7 +162,7 @@ class EmbargoCheckAccessApiTests(ModuleStoreTestCase):
|
||||
# exception when the embargo middleware treated the value as a string.
|
||||
# In order to simulate this behavior, we can't simply set `profile.country = None`.
|
||||
# (because when we save it, it will set the database field to an empty string instead of NULL)
|
||||
query = u"UPDATE auth_userprofile SET country = NULL WHERE id = %s"
|
||||
query = "UPDATE auth_userprofile SET country = NULL WHERE id = %s"
|
||||
connection.cursor().execute(query, [str(self.user.profile.id)])
|
||||
|
||||
# Verify that we can check the user's access without error
|
||||
@@ -270,7 +270,7 @@ class EmbargoMessageUrlApiTests(UrlResetMixin, ModuleStoreTestCase):
|
||||
|
||||
@patch.dict(settings.FEATURES, {'EMBARGO': True})
|
||||
def setUp(self):
|
||||
super(EmbargoMessageUrlApiTests, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().setUp()
|
||||
self.course = CourseFactory.create()
|
||||
|
||||
@ddt.data(
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Unit tests for embargo app admin forms.
|
||||
"""
|
||||
|
||||
|
||||
import six
|
||||
# Explicitly import the cache from ConfigurationModel so we can reset it after each test
|
||||
from config_models.models import cache
|
||||
from django.test import TestCase
|
||||
@@ -23,7 +21,7 @@ class RestrictedCourseFormTest(ModuleStoreTestCase):
|
||||
def test_save_valid_data(self):
|
||||
course = CourseFactory.create()
|
||||
data = {
|
||||
'course_key': six.text_type(course.id),
|
||||
'course_key': str(course.id),
|
||||
'enroll_msg_key': 'default',
|
||||
'access_msg_key': 'default'
|
||||
}
|
||||
@@ -59,7 +57,7 @@ class IPFilterFormTest(TestCase):
|
||||
"""Test form for adding [black|white]list IP addresses"""
|
||||
|
||||
def tearDown(self):
|
||||
super(IPFilterFormTest, self).tearDown() # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().tearDown()
|
||||
# Explicitly clear ConfigurationModel's cache so tests have a clear cache
|
||||
# and don't interfere with each other
|
||||
cache.clear()
|
||||
@@ -69,31 +67,31 @@ class IPFilterFormTest(TestCase):
|
||||
# should be able to do both ipv4 and ipv6
|
||||
# spacing should not matter
|
||||
form_data = {
|
||||
'whitelist': u'127.0.0.1, 2003:dead:beef:4dad:23:46:bb:101, 1.1.0.1/32, 1.0.0.0/24',
|
||||
'blacklist': u' 18.244.1.5 , 2002:c0a8:101::42, 18.36.22.1, 1.0.0.0/16'
|
||||
'whitelist': '127.0.0.1, 2003:dead:beef:4dad:23:46:bb:101, 1.1.0.1/32, 1.0.0.0/24',
|
||||
'blacklist': ' 18.244.1.5 , 2002:c0a8:101::42, 18.36.22.1, 1.0.0.0/16'
|
||||
}
|
||||
form = IPFilterForm(data=form_data)
|
||||
assert form.is_valid()
|
||||
form.save()
|
||||
whitelist = IPFilter.current().whitelist_ips
|
||||
blacklist = IPFilter.current().blacklist_ips
|
||||
for addr in u'127.0.0.1, 2003:dead:beef:4dad:23:46:bb:101'.split(','):
|
||||
for addr in '127.0.0.1, 2003:dead:beef:4dad:23:46:bb:101'.split(','):
|
||||
assert addr.strip() in whitelist
|
||||
for addr in u'18.244.1.5, 2002:c0a8:101::42, 18.36.22.1'.split(','):
|
||||
for addr in '18.244.1.5, 2002:c0a8:101::42, 18.36.22.1'.split(','):
|
||||
assert addr.strip() in blacklist
|
||||
|
||||
# Network tests
|
||||
# ips not in whitelist network
|
||||
for addr in [u'1.1.0.2', u'1.0.1.0']:
|
||||
for addr in ['1.1.0.2', '1.0.1.0']:
|
||||
assert addr.strip() not in whitelist
|
||||
# ips in whitelist network
|
||||
for addr in [u'1.1.0.1', u'1.0.0.100']:
|
||||
for addr in ['1.1.0.1', '1.0.0.100']:
|
||||
assert addr.strip() in whitelist
|
||||
# ips not in blacklist network
|
||||
for addr in [u'2.0.0.0', u'1.1.0.0']:
|
||||
for addr in ['2.0.0.0', '1.1.0.0']:
|
||||
assert addr.strip() not in blacklist
|
||||
# ips in blacklist network
|
||||
for addr in [u'1.0.100.0', u'1.0.0.10']:
|
||||
for addr in ['1.0.100.0', '1.0.0.10']:
|
||||
assert addr.strip() in blacklist
|
||||
|
||||
# Test clearing by adding an empty list is OK too
|
||||
@@ -110,26 +108,18 @@ class IPFilterFormTest(TestCase):
|
||||
def test_add_invalid_ips(self):
|
||||
# test adding invalid ip addresses
|
||||
form_data = {
|
||||
'whitelist': u'.0.0.1, :dead:beef:::, 1.0.0.0/55',
|
||||
'blacklist': u' 18.244.* , 999999:c0a8:101::42, 1.0.0.0/'
|
||||
'whitelist': '.0.0.1, :dead:beef:::, 1.0.0.0/55',
|
||||
'blacklist': ' 18.244.* , 999999:c0a8:101::42, 1.0.0.0/'
|
||||
}
|
||||
form = IPFilterForm(data=form_data)
|
||||
assert not form.is_valid()
|
||||
|
||||
if six.PY2:
|
||||
wmsg = "Invalid IP Address(es): [u'.0.0.1', u':dead:beef:::', u'1.0.0.0/55']" \
|
||||
" Please fix the error(s) and try again."
|
||||
else:
|
||||
wmsg = "Invalid IP Address(es): ['.0.0.1', ':dead:beef:::', '1.0.0.0/55']" \
|
||||
" Please fix the error(s) and try again."
|
||||
wmsg = "Invalid IP Address(es): ['.0.0.1', ':dead:beef:::', '1.0.0.0/55']"\
|
||||
" Please fix the error(s) and try again."
|
||||
assert wmsg == form._errors['whitelist'][0] # pylint: disable=protected-access
|
||||
|
||||
if six.PY2:
|
||||
bmsg = "Invalid IP Address(es): [u'18.244.*', u'999999:c0a8:101::42', u'1.0.0.0/']" \
|
||||
" Please fix the error(s) and try again."
|
||||
else:
|
||||
bmsg = "Invalid IP Address(es): ['18.244.*', '999999:c0a8:101::42', '1.0.0.0/']" \
|
||||
" Please fix the error(s) and try again."
|
||||
bmsg = "Invalid IP Address(es): ['18.244.*', '999999:c0a8:101::42', '1.0.0.0/']"\
|
||||
" Please fix the error(s) and try again."
|
||||
assert bmsg == form._errors['blacklist'][0] # pylint: disable=protected-access
|
||||
|
||||
with self.assertRaisesRegex(ValueError, "The IPFilter could not be created because the data didn't validate."):
|
||||
|
||||
@@ -3,13 +3,12 @@ Tests for EmbargoMiddleware with CountryAccessRules
|
||||
"""
|
||||
|
||||
|
||||
from unittest.mock import patch
|
||||
import ddt
|
||||
import six
|
||||
from config_models.models import cache as config_cache
|
||||
from django.conf import settings
|
||||
from django.core.cache import cache as django_cache
|
||||
from django.urls import reverse
|
||||
from mock import patch
|
||||
|
||||
from openedx.core.djangolib.testing.utils import skip_unless_lms
|
||||
from common.djangoapps.student.tests.factories import UserFactory
|
||||
@@ -38,14 +37,14 @@ class EmbargoMiddlewareAccessTests(UrlResetMixin, ModuleStoreTestCase):
|
||||
|
||||
@patch.dict(settings.FEATURES, {'EMBARGO': True})
|
||||
def setUp(self):
|
||||
super(EmbargoMiddlewareAccessTests, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().setUp()
|
||||
self.user = UserFactory(username=self.USERNAME, password=self.PASSWORD)
|
||||
self.course = CourseFactory.create()
|
||||
self.client.login(username=self.USERNAME, password=self.PASSWORD)
|
||||
|
||||
self.courseware_url = reverse(
|
||||
'openedx.course_experience.course_home',
|
||||
kwargs={'course_id': six.text_type(self.course.id)}
|
||||
kwargs={'course_id': str(self.course.id)}
|
||||
)
|
||||
self.non_courseware_url = reverse('dashboard')
|
||||
|
||||
@@ -82,14 +81,14 @@ class EmbargoMiddlewareAccessTests(UrlResetMixin, ModuleStoreTestCase):
|
||||
@patch.dict(settings.FEATURES, {'EMBARGO': True})
|
||||
@ddt.data(
|
||||
# request_ip, blacklist, whitelist, is_enabled, allow_access
|
||||
(u'173.194.123.35', ['173.194.123.35'], [], True, False),
|
||||
(u'173.194.123.35', ['173.194.0.0/16'], [], True, False),
|
||||
(u'173.194.123.35', ['127.0.0.0/32', '173.194.0.0/16'], [], True, False),
|
||||
(u'173.195.10.20', ['173.194.0.0/16'], [], True, True),
|
||||
(u'173.194.123.35', ['173.194.0.0/16'], ['173.194.0.0/16'], True, False),
|
||||
(u'173.194.123.35', [], ['173.194.0.0/16'], True, True),
|
||||
(u'192.178.2.3', [], ['173.194.0.0/16'], True, True),
|
||||
(u'173.194.123.35', ['173.194.123.35'], [], False, True),
|
||||
('173.194.123.35', ['173.194.123.35'], [], True, False),
|
||||
('173.194.123.35', ['173.194.0.0/16'], [], True, False),
|
||||
('173.194.123.35', ['127.0.0.0/32', '173.194.0.0/16'], [], True, False),
|
||||
('173.195.10.20', ['173.194.0.0/16'], [], True, True),
|
||||
('173.194.123.35', ['173.194.0.0/16'], ['173.194.0.0/16'], True, False),
|
||||
('173.194.123.35', [], ['173.194.0.0/16'], True, True),
|
||||
('192.178.2.3', [], ['173.194.0.0/16'], True, True),
|
||||
('173.194.123.35', ['173.194.123.35'], [], False, True),
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_ip_access_rules(self, request_ip, blacklist, whitelist, is_enabled, allow_access):
|
||||
@@ -155,7 +154,7 @@ class EmbargoMiddlewareAccessTests(UrlResetMixin, ModuleStoreTestCase):
|
||||
def test_whitelist_ip_skips_country_access_checks(self):
|
||||
# Whitelist an IP address
|
||||
IPFilter.objects.create(
|
||||
whitelist=u"192.168.10.20",
|
||||
whitelist="192.168.10.20",
|
||||
enabled=True
|
||||
)
|
||||
|
||||
@@ -165,8 +164,8 @@ class EmbargoMiddlewareAccessTests(UrlResetMixin, ModuleStoreTestCase):
|
||||
# Make a request from the whitelisted IP address
|
||||
response = self.client.get(
|
||||
self.courseware_url,
|
||||
HTTP_X_FORWARDED_FOR=u"192.168.10.20",
|
||||
REMOTE_ADDR=u"192.168.10.20"
|
||||
HTTP_X_FORWARDED_FOR="192.168.10.20",
|
||||
REMOTE_ADDR="192.168.10.20"
|
||||
)
|
||||
|
||||
# Expect that we were still able to access the page,
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
|
||||
import json
|
||||
import pytest
|
||||
import six
|
||||
from django.db.utils import IntegrityError
|
||||
from django.test import TestCase
|
||||
from opaque_keys.edx.locator import CourseLocator
|
||||
@@ -37,14 +36,14 @@ class EmbargoModelsTest(CacheIsolationTestCase):
|
||||
|
||||
# Now, course should be embargoed
|
||||
assert EmbargoedCourse.is_embargoed(course_id)
|
||||
assert six.text_type(cauth) == u"Course '{course_id}' is Embargoed".format(course_id=course_id)
|
||||
assert str(cauth) == f"Course '{course_id}' is Embargoed"
|
||||
|
||||
# Unauthorize by explicitly setting email_enabled to False
|
||||
cauth.embargoed = False
|
||||
cauth.save()
|
||||
# Test that course is now unauthorized
|
||||
assert not EmbargoedCourse.is_embargoed(course_id)
|
||||
assert six.text_type(cauth) == u"Course '{course_id}' is Not Embargoed".format(course_id=course_id)
|
||||
assert str(cauth) == f"Course '{course_id}' is Not Embargoed"
|
||||
|
||||
def test_state_embargo(self):
|
||||
# Azerbaijan and France should not be blocked
|
||||
@@ -78,8 +77,8 @@ class EmbargoModelsTest(CacheIsolationTestCase):
|
||||
assert state in currently_blocked
|
||||
|
||||
def test_ip_blocking(self):
|
||||
whitelist = u'127.0.0.1'
|
||||
blacklist = u'18.244.51.3'
|
||||
whitelist = '127.0.0.1'
|
||||
blacklist = '18.244.51.3'
|
||||
|
||||
cwhitelist = IPFilter.current().whitelist_ips
|
||||
assert whitelist not in cwhitelist
|
||||
@@ -94,20 +93,20 @@ class EmbargoModelsTest(CacheIsolationTestCase):
|
||||
assert blacklist in cblacklist
|
||||
|
||||
def test_ip_network_blocking(self):
|
||||
whitelist = u'1.0.0.0/24'
|
||||
blacklist = u'1.1.0.0/16'
|
||||
whitelist = '1.0.0.0/24'
|
||||
blacklist = '1.1.0.0/16'
|
||||
|
||||
IPFilter(whitelist=whitelist, blacklist=blacklist).save()
|
||||
|
||||
cwhitelist = IPFilter.current().whitelist_ips
|
||||
assert u'1.0.0.100' in cwhitelist
|
||||
assert u'1.0.0.10' in cwhitelist
|
||||
assert u'1.0.1.0' not in cwhitelist
|
||||
assert '1.0.0.100' in cwhitelist
|
||||
assert '1.0.0.10' in cwhitelist
|
||||
assert '1.0.1.0' not in cwhitelist
|
||||
cblacklist = IPFilter.current().blacklist_ips
|
||||
assert u'1.1.0.0' in cblacklist
|
||||
assert u'1.1.0.1' in cblacklist
|
||||
assert u'1.1.1.0' in cblacklist
|
||||
assert u'1.2.0.0' not in cblacklist
|
||||
assert '1.1.0.0' in cblacklist
|
||||
assert '1.1.0.1' in cblacklist
|
||||
assert '1.1.1.0' in cblacklist
|
||||
assert '1.2.0.0' not in cblacklist
|
||||
|
||||
|
||||
class RestrictedCourseTest(CacheIsolationTestCase):
|
||||
@@ -118,7 +117,7 @@ class RestrictedCourseTest(CacheIsolationTestCase):
|
||||
def test_unicode_values(self):
|
||||
course_id = CourseLocator('abc', '123', 'doremi')
|
||||
restricted_course = RestrictedCourse.objects.create(course_key=course_id)
|
||||
assert six.text_type(restricted_course) == six.text_type(course_id)
|
||||
assert str(restricted_course) == str(course_id)
|
||||
|
||||
def test_restricted_course_cache_with_save_delete(self):
|
||||
course_id = CourseLocator('abc', '123', 'doremi')
|
||||
@@ -167,7 +166,7 @@ class CountryTest(TestCase):
|
||||
|
||||
def test_unicode_values(self):
|
||||
country = Country.objects.create(country='NZ')
|
||||
assert six.text_type(country) == 'New Zealand (NZ)'
|
||||
assert str(country) == 'New Zealand (NZ)'
|
||||
|
||||
|
||||
class CountryAccessRuleTest(CacheIsolationTestCase):
|
||||
@@ -184,7 +183,7 @@ class CountryAccessRuleTest(CacheIsolationTestCase):
|
||||
country=country
|
||||
)
|
||||
|
||||
assert six.text_type(access_rule) == u'Whitelist New Zealand (NZ) for {course_key}'.format(course_key=course_id)
|
||||
assert str(access_rule) == f'Whitelist New Zealand (NZ) for {course_id}'
|
||||
|
||||
course_id = CourseLocator('def', '123', 'doremi')
|
||||
restricted_course1 = RestrictedCourse.objects.create(course_key=course_id)
|
||||
@@ -194,7 +193,7 @@ class CountryAccessRuleTest(CacheIsolationTestCase):
|
||||
country=country
|
||||
)
|
||||
|
||||
assert six.text_type(access_rule) == u'Blacklist New Zealand (NZ) for {course_key}'.format(course_key=course_id)
|
||||
assert str(access_rule) == f'Blacklist New Zealand (NZ) for {course_id}'
|
||||
|
||||
def test_unique_together_constraint(self):
|
||||
"""
|
||||
@@ -246,7 +245,7 @@ class CourseAccessRuleHistoryTest(TestCase):
|
||||
"""Test course access rule history. """
|
||||
|
||||
def setUp(self):
|
||||
super(CourseAccessRuleHistoryTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().setUp()
|
||||
self.course_key = CourseLocator('edx', 'DemoX', 'Demo_Course')
|
||||
self.restricted_course = RestrictedCourse.objects.create(course_key=self.course_key)
|
||||
self.countries = {
|
||||
|
||||
@@ -1,13 +1,14 @@
|
||||
"""Tests for embargo app views. """
|
||||
|
||||
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import ddt
|
||||
import maxminddb
|
||||
import geoip2.database
|
||||
|
||||
from django.urls import reverse
|
||||
from django.conf import settings
|
||||
from mock import patch, MagicMock
|
||||
|
||||
from .factories import CountryAccessRuleFactory, RestrictedCourseFactory
|
||||
from .. import messages
|
||||
@@ -46,7 +47,7 @@ class CourseAccessMessageViewTest(CacheIsolationTestCase, UrlResetMixin):
|
||||
|
||||
@patch.dict(settings.FEATURES, {'EMBARGO': True})
|
||||
def setUp(self):
|
||||
super(CourseAccessMessageViewTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().setUp()
|
||||
|
||||
@ddt.data(*list(messages.ENROLL_MESSAGES.keys()))
|
||||
def test_enrollment_messages(self, msg_key):
|
||||
@@ -94,7 +95,7 @@ class CheckCourseAccessViewTest(CourseApiFactoryMixin, ModuleStoreTestCase):
|
||||
|
||||
@patch.dict(settings.FEATURES, {'EMBARGO': True})
|
||||
def setUp(self):
|
||||
super(CheckCourseAccessViewTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
|
||||
super().setUp()
|
||||
self.url = reverse('api_embargo:v1_course_access')
|
||||
user = UserFactory(is_staff=True)
|
||||
self.client.login(username=user.username, password=UserFactory._DEFAULT_PASSWORD) # lint-amnesty, pylint: disable=protected-access
|
||||
|
||||
Reference in New Issue
Block a user