Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from __future__ import unicode_literals
import pytest
from django.conf import settings
try:
from django.core.urlresolvers import reverse, resolve
except ImportError: # >= Django 2.0
from django.urls import reverse, resolve
from .admin import Search404
skip_old_haystack = pytest.mark.skipif(settings.OLD_HAYSTACK is True,
reason="Doesn't apply to Haystack 1.2.x")
skip_new_haystack = pytest.mark.skipif(settings.OLD_HAYSTACK is False,
reason="Doesn't apply to Haystack 2.x")
@skip_new_haystack
def test_env_setting_old_haystack():
assert settings.OLD_HAYSTACK is True
@skip_old_haystack
def test_env_setting_new_haystack():
assert settings.OLD_HAYSTACK is False
def test_app_is_mounted_accessing_changelist_but_no_models_loaded(admin_user, rf):
user = models.ForeignKey(
settings.AUTH_USER_MODEL,
related_name='notices',
db_index=True)
has_read = models.BooleanField(db_index=True, default=False)
created = models.DateTimeField(auto_now_add=True, db_index=True)
message = models.TextField()
category = models.CharField(max_length=200, db_index=True)
target_type = models.ForeignKey('contenttypes.ContentType', null=True, blank=True)
target_id = models.PositiveIntegerField(default=0, null=True)
target_slug = models.CharField(max_length=20, default='')
target = GenericForeignKey('target_type', 'target_id')
actor = models.ForeignKey(settings.AUTH_USER_MODEL, null=True)
objects = NoticeQuerySet.as_manager()
class Meta:
ordering = ('-has_read', '-created')
def mark_read(self):
self.has_read = True
self.save()
def oauth_redirect_to_root(request: HttpRequest, url: str,
sso_type: str, is_signup: bool=False,
extra_url_params: Dict[str, str]={}) -> HttpResponse:
main_site_uri = settings.ROOT_DOMAIN_URI + url
if settings.SOCIAL_AUTH_SUBDOMAIN is not None and sso_type == 'social':
main_site_uri = (settings.EXTERNAL_URI_SCHEME +
settings.SOCIAL_AUTH_SUBDOMAIN +
"." +
settings.EXTERNAL_HOST) + url
params = {
'subdomain': get_subdomain(request),
'is_signup': '1' if is_signup else '0',
}
params['multiuse_object_key'] = request.GET.get('multiuse_object_key', '')
# mobile_flow_otp is a one-time pad provided by the app that we
# can use to encrypt the API key when passing back to the app.
mobile_flow_otp = request.GET.get('mobile_flow_otp')
if mobile_flow_otp is not None:
if not is_valid_otp(mobile_flow_otp):
def get_username():
return settings.GITHUB_USERNAME
def middleware(request):
response = get_response(request)
if settings.CSP_POLICIES:
policies = []
for directive, values in settings.CSP_POLICIES.items():
policies.append(directive + ' ' + ' '.join(values))
response['Content-Security-Policy'] = '; '.join(policies)
return response
class BaseAdmin(admin.ModelAdmin):
readonly_fields = ('created_datetime','updated_datetime','created_by','last_updated_by')
def save_model(self, request, obj, form, change):
if hasattr(obj, 'created_by') and not obj.id:
obj.created_by = request.user
if hasattr(obj, 'last_updated_by'):
obj.last_updated_by = request.user
obj.save()
class DataConnectionAdmin(BaseAdmin):
formfield_overrides = {
EncryptedCharField: {'widget': PasswordInput(render_value=True)},
}
class ParameterAdmin(BaseAdmin):
if not getattr(settings,'MR_REPORTS_ALLOW_NATIVE_PYTHON_CODE_EXEC_ON_SERVER',False):
readonly_fields = ('python_create_default',)
class DataSetParameterInline(admin.TabularInline):
model = DataSetParameter
#fields = (edit_parameter_link,)
extra = 0
#def edit_parameter_link(self, instance):
# return 'test' #instance.parameter.edit_link()
#edit_parameter_link.short_description = "Edit Parameter"
class DataSetAdmin(BaseAdmin):
class Media:
js = ('js/create_parameter_edit_links.js',)
if not getattr(settings,'MR_REPORTS_ALLOW_NATIVE_PYTHON_CODE_EXEC_ON_SERVER',False):
readonly_fields = ('python_post_processing',)
def process_statement(stmt, auth, payload_sha2s):
# Add id to statement if not present
if 'id' not in stmt:
stmt['id'] = str(uuid.uuid4())
# Add version to statement if not present
if 'version' not in stmt:
stmt['version'] = settings.XAPI_VERSIONS[0]
# Convert context activities to list if dict
if 'context' in stmt and 'contextActivities' in stmt['context']:
for k, v in stmt['context']['contextActivities'].items():
if isinstance(v, dict):
stmt['context']['contextActivities'][k] = [v]
# Convert context activities to list if dict (for substatements)
if 'objectType' in stmt['object'] and stmt['object']['objectType'] == 'SubStatement':
if 'context' in stmt['object'] and 'contextActivities' in stmt['object']['context']:
for k, v in stmt['object']['context']['contextActivities'].items():
if isinstance(v, dict):
stmt['object']['context']['contextActivities'][k] = [v]
# Add stored time
stmt['stored'] = datetime.utcnow().replace(tzinfo=utc).isoformat()
from datetime import datetime, timedelta
from django.conf import settings
from odk_viewer.models.parsed_instance import dict_for_mongo, _encode_for_mongo,\
DATETIME_FORMAT
audit = settings.MONGO_DB.auditlog
DEFAULT_LIMIT = 1000
class AuditLog(object):
ACCOUNT = u"account"
DEFAULT_BATCHSIZE = 1000
CREATED_ON = u"created_on"
def __init__(self, data):
self.data = data
def save(self):
return audit.save(self.data)
@classmethod
def query_mongo(cls, username, query=None, fields=None, sort=None, start=0,
limit=DEFAULT_LIMIT, count=False):
query = dict_for_mongo(query) if query else {}
('name', models.CharField(max_length=255)),
('created_at', models.DateTimeField(auto_now_add=True)),
('last_updated_on', models.DateTimeField(auto_now=True)),
('is_active', models.BooleanField(default=True)),
('api_key', models.CharField(max_length=50, null=True)),
],
),
migrations.CreateModel(
name='BusinessTeamMember',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('activation_key', models.CharField(max_length=36)),
('created_at', models.DateTimeField(auto_now_add=True)),
('last_updated_on', models.DateTimeField(auto_now=True)),
('business', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='saas.Business')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
],
),
migrations.CreateModel(
name='Plan',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=255)),
('description', models.CharField(blank=True, max_length=255, null=True)),
('duration_days', models.IntegerField(default=15)),
('usd_price', models.DecimalField(decimal_places=2, max_digits=6)),
('created_at', models.DateTimeField(auto_now_add=True)),
('last_updated_on', models.DateTimeField(auto_now=True)),
],
),
migrations.CreateModel(
name='Subscription',
def _setup(self):
self._wrapped = get_storage_class(settings.STATICFILES_STORAGE)()