bruks/dynamicweb
1
0
Fork 0
forked from ungleich-public/dynamicweb
Code Pull requests Activity

vendor backport_collections, sekizai, stored_messages, treebeard

Browse source
This commit is contained in:
bruks 2026-10-02 15:44:50 +01:00
commit 6defe00adb
133 changed files with 13549 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

2
alplora/urls.py
View file

@ -1,4 +1,4 @@
from django.conf.urls import url
from django.urls import re_path as url
from .views import IndexView, LoginView, ContactView

2
datacenterlight/cms_models.py
View file

@ -44,7 +44,7 @@ class CMSIntegration(models.Model):
class CMSFaviconExtension(PageExtension):
favicon = FilerFileField(related_name="cms_favicon_image")
favicon = FilerFileField(related_name="cms_favicon_image", on_delete=models.CASCADE)
extension_pool.register(CMSFaviconExtension)

2
datacenterlight/urls.py
View file

@ -1,4 +1,4 @@
from django.conf.urls import url
from django.urls import re_path as url
from django.views.generic import TemplateView, RedirectView
from utils.views import AskSSHKeyView

2
digitalglarus/models.py
View file

@ -338,7 +338,7 @@ class DGGallery(models.Model):
class DGPicture(models.Model):
gallery = models.ForeignKey(DGGallery, on_delete=models.SET_NULL, null=True,)
image = FilerImageField(related_name='dg_gallery')
image = FilerImageField(related_name='dg_gallery', on_delete=models.CASCADE)
description = models.CharField(max_length=60)
def __str__(self):

2
digitalglarus/urls.py
View file

@ -1,4 +1,4 @@
from django.conf.urls import url
from django.urls import re_path as url
from django.utils.translation import gettext_lazy as _
from django.contrib.auth import views as auth_views

4
dynamicweb/settings/base.py
View file

@ -112,12 +112,12 @@ INSTALLED_APPS = (
'cms', # django CMS itself
'aldryn_apphooks_config',
'aldryn_boilerplates',
'aldryn_categories',
#TODO:'aldryn_categories',
'aldryn_common',
#TODO:'aldryn_newsblog',
#TODO:'aldryn_people',
#TODO: 'aldryn_reversion',
'aldryn_translation_tools',
#TODO: 'aldryn_translation_tools',
'treebeard', # utilities for implementing a tree
'sekizai', # for javascript and css management
'menus', # helper for model independent hierarchical website navigation

17
dynamicweb/wsgi.py
View file

@ -2,18 +2,23 @@
WSGI config for dynamicweb project.
It exposes the WSGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/1.8/howto/deployment/wsgi/
"""
import os
import sys
from django.core.wsgi import get_wsgi_application
# sys.path.append(os.path.dirname(__file__))
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
VENDOR_DIR = os.path.join(BASE_DIR, "vendor")
# Load vendored packages before site-packages
if VENDOR_DIR not in sys.path:
sys.path.insert(0, VENDOR_DIR)
# Legacy Python 3.4 packages, if still required
sys.path.append('/home/app/pyvenv/lib/python3.4/site-packages/')
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "dynamicweb.settings.prod")
application = get_wsgi_application()
from django.core.wsgi import get_wsgi_application
application = get_wsgi_application()

6
hosting/models.py
View file

@ -104,8 +104,8 @@ class HostingOrder(AssignPermissionsMixin, models.Model):
ORDER_DECLINED_STATUS = 'Declined'
vm_id = models.IntegerField(default=0)
customer = models.ForeignKey(StripeCustomer)
billing_address = models.ForeignKey(BillingAddress)
customer = models.ForeignKey(StripeCustomer, on_delete=models.CASCADE)
billing_address = models.ForeignKey(BillingAddress, on_delete=models.CASCADE)
created_at = models.DateTimeField(auto_now_add=True)
approved = models.BooleanField(default=False)
last4 = models.CharField(max_length=4)
@ -574,7 +574,7 @@ class VMDetail(models.Model):
class UserCardDetail(AssignPermissionsMixin, models.Model):
permissions = ('view_usercarddetail',)
stripe_customer = models.ForeignKey(StripeCustomer)
stripe_customer = models.ForeignKey(StripeCustomer, on_delete=models.CASCADE)
last4 = models.CharField(max_length=4)
brand = models.CharField(max_length=128)
card_id = models.CharField(max_length=100, blank=True, default='')

2
hosting/urls.py
View file

@ -1,4 +1,4 @@
from django.conf.urls import url
from django.urls import re_path as url
from django.contrib.auth import views as auth_views
from utils.views import SSHKeyCreateView, AskSSHKeyView

8
legacy_compat.py
View file

@ -545,4 +545,10 @@ try:
filer_compat.python_2_unicode_compatible = python_2_unicode_compatible
except ImportError:
pass
pass
# Legacy packages expect django.utils.six
if not hasattr(django.utils, "six"):
django.utils.six = six
sys.modules["django.utils.six"] = six

2
membership/urls.py
View file

@ -1,5 +1,5 @@
__author__ = 'tomislav'
from django.conf.urls import url
from django.urls import re_path as url
from django.contrib.auth.decorators import login_required
from . import views

3
ungleich/models.py
View file

@ -1,6 +1,7 @@
from cms.extensions import PageExtension
from cms.extensions.extension_pool import extension_pool
from filer.fields.image import FilerImageField
from django.db import models
# Create your models here.
@ -9,7 +10,7 @@ from filer.fields.image import FilerImageField
class UngleichPage(PageExtension):
# image_header = models.ImageField(upload_to='image_header')
image = FilerImageField(null=True, blank=True,
related_name="ungleinch_page_image")
related_name="ungleinch_page_image", on_delete=models.CASCADE)
class Meta:
app_label = 'ungleich'

2
ungleich/urls.py
View file

@ -1,4 +1,4 @@
from django.conf.urls import url
from django.urls import re_path as url
from . import views
from .views import PostDetailViewUngleich

2
ungleich_page/urls.py
View file

@ -1,4 +1,4 @@
from django.conf.urls import url
from django.urls import re_path as url
from .views import ContactView, LandingView
from django.utils.translation import gettext_lazy as _

732
vendor/backport_collections.py vendored Normal file
View file

@ -0,0 +1,732 @@
__all__ = ['Counter', 'deque', 'defaultdict', 'namedtuple', 'OrderedDict']
# For bootstrapping reasons, the collection ABCs are defined in _abcoll.py.
# They should however be considered an integral part of collections.py.
from backport_abcoll import *
import backport_abcoll
__all__ += backport_abcoll.__all__
from _collections import defaultdict
from collections import deque as _deque
from operator import itemgetter as _itemgetter, eq as _eq
from keyword import iskeyword as _iskeyword
import sys as _sys
import heapq as _heapq
from itertools import repeat as _repeat, chain as _chain, starmap as _starmap
from itertools import imap as _imap
try:
from thread import get_ident as _get_ident
except ImportError:
from dummy_thread import get_ident as _get_ident
if _sys.version_info >= (2, 7):
import warnings as _warnings
_warnings.warn('Use the stock collections modules instead.', DeprecationWarning)
################################################################################
### OrderedDict
################################################################################
class OrderedDict(dict):
'Dictionary that remembers insertion order'
# An inherited dict maps keys to values.
# The inherited dict provides __getitem__, __len__, __contains__, and get.
# The remaining methods are order-aware.
# Big-O running times for all methods are the same as regular dictionaries.
# The internal self.__map dict maps keys to links in a doubly linked list.
# The circular doubly linked list starts and ends with a sentinel element.
# The sentinel element never gets deleted (this simplifies the algorithm).
# Each link is stored as a list of length three: [PREV, NEXT, KEY].
def __init__(self, *args, **kwds):
'''Initialize an ordered dictionary. The signature is the same as
regular dictionaries, but keyword arguments are not recommended because
their insertion order is arbitrary.
'''
if len(args) > 1:
raise TypeError('expected at most 1 arguments, got %d' % len(args))
try:
self.__root
except AttributeError:
self.__root = root = [] # sentinel node
root[:] = [root, root, None]
self.__map = {}
self.__update(*args, **kwds)
def __setitem__(self, key, value, dict_setitem=dict.__setitem__):
'od.__setitem__(i, y) <==> od[i]=y'
# Setting a new item creates a new link at the end of the linked list,
# and the inherited dictionary is updated with the new key/value pair.
if key not in self:
root = self.__root
last = root[0]
last[1] = root[0] = self.__map[key] = [last, root, key]
return dict_setitem(self, key, value)
def __delitem__(self, key, dict_delitem=dict.__delitem__):
'od.__delitem__(y) <==> del od[y]'
# Deleting an existing item uses self.__map to find the link which gets
# removed by updating the links in the predecessor and successor nodes.
dict_delitem(self, key)
link_prev, link_next, _ = self.__map.pop(key)
link_prev[1] = link_next # update link_prev[NEXT]
link_next[0] = link_prev # update link_next[PREV]
def __iter__(self):
'od.__iter__() <==> iter(od)'
# Traverse the linked list in order.
root = self.__root
curr = root[1] # start at the first node
while curr is not root:
yield curr[2] # yield the curr[KEY]
curr = curr[1] # move to next node
def __reversed__(self):
'od.__reversed__() <==> reversed(od)'
# Traverse the linked list in reverse order.
root = self.__root
curr = root[0] # start at the last node
while curr is not root:
yield curr[2] # yield the curr[KEY]
curr = curr[0] # move to previous node
def clear(self):
'od.clear() -> None. Remove all items from od.'
root = self.__root
root[:] = [root, root, None]
self.__map.clear()
dict.clear(self)
# -- the following methods do not depend on the internal structure --
def keys(self):
'od.keys() -> list of keys in od'
return list(self)
def values(self):
'od.values() -> list of values in od'
return [self[key] for key in self]
def items(self):
'od.items() -> list of (key, value) pairs in od'
return [(key, self[key]) for key in self]
def iterkeys(self):
'od.iterkeys() -> an iterator over the keys in od'
return iter(self)
def itervalues(self):
'od.itervalues -> an iterator over the values in od'
for k in self:
yield self[k]
def iteritems(self):
'od.iteritems -> an iterator over the (key, value) pairs in od'
for k in self:
yield (k, self[k])
update = MutableMapping.update
__update = update # let subclasses override update without breaking __init__
__marker = object()
def pop(self, key, default=__marker):
'''od.pop(k[,d]) -> v, remove specified key and return the corresponding
value. If key is not found, d is returned if given, otherwise KeyError
is raised.
'''
if key in self:
result = self[key]
del self[key]
return result
if default is self.__marker:
raise KeyError(key)
return default
def setdefault(self, key, default=None):
'od.setdefault(k[,d]) -> od.get(k,d), also set od[k]=d if k not in od'
if key in self:
return self[key]
self[key] = default
return default
def popitem(self, last=True):
'''od.popitem() -> (k, v), return and remove a (key, value) pair.
Pairs are returned in LIFO order if last is true or FIFO order if false.
'''
if not self:
raise KeyError('dictionary is empty')
key = next(reversed(self) if last else iter(self))
value = self.pop(key)
return key, value
def __repr__(self, _repr_running={}):
'od.__repr__() <==> repr(od)'
call_key = id(self), _get_ident()
if call_key in _repr_running:
return '...'
_repr_running[call_key] = 1
try:
if not self:
return '%s()' % (self.__class__.__name__,)
return '%s(%r)' % (self.__class__.__name__, self.items())
finally:
del _repr_running[call_key]
def __reduce__(self):
'Return state information for pickling'
items = [[k, self[k]] for k in self]
inst_dict = vars(self).copy()
for k in vars(OrderedDict()):
inst_dict.pop(k, None)
if inst_dict:
return (self.__class__, (items,), inst_dict)
return self.__class__, (items,)
def copy(self):
'od.copy() -> a shallow copy of od'
return self.__class__(self)
@classmethod
def fromkeys(cls, iterable, value=None):
'''OD.fromkeys(S[, v]) -> New ordered dictionary with keys from S.
If not specified, the value defaults to None.
'''
self = cls()
for key in iterable:
self[key] = value
return self
def __eq__(self, other):
'''od.__eq__(y) <==> od==y. Comparison to another OD is order-sensitive
while comparison to a regular mapping is order-insensitive.
'''
if isinstance(other, OrderedDict):
return dict.__eq__(self, other) and all(_imap(_eq, self, other))
return dict.__eq__(self, other)
def __ne__(self, other):
'od.__ne__(y) <==> od!=y'
return not self == other
# -- the following methods support python 3.x style dictionary views --
def viewkeys(self):
"od.viewkeys() -> a set-like object providing a view on od's keys"
return KeysView(self)
def viewvalues(self):
"od.viewvalues() -> an object providing a view on od's values"
return ValuesView(self)
def viewitems(self):
"od.viewitems() -> a set-like object providing a view on od's items"
return ItemsView(self)
################################################################################
### namedtuple
################################################################################
_class_template = '''\
class {typename}(tuple):
'{typename}({arg_list})'
__slots__ = ()
_fields = {field_names!r}
def __new__(_cls, {arg_list}):
'Create new instance of {typename}({arg_list})'
return _tuple.__new__(_cls, ({arg_list}))
@classmethod
def _make(cls, iterable, new=tuple.__new__, len=len):
'Make a new {typename} object from a sequence or iterable'
result = new(cls, iterable)
if len(result) != {num_fields:d}:
raise TypeError('Expected {num_fields:d} arguments, got %d' % len(result))
return result
def __repr__(self):
'Return a nicely formatted representation string'
return '{typename}({repr_fmt})' % self
def _asdict(self):
'Return a new OrderedDict which maps field names to their values'
return OrderedDict(zip(self._fields, self))
def _replace(_self, **kwds):
'Return a new {typename} object replacing specified fields with new values'
result = _self._make(map(kwds.pop, {field_names!r}, _self))
if kwds:
raise ValueError('Got unexpected field names: %r' % kwds.keys())
return result
def __getnewargs__(self):
'Return self as a plain tuple. Used by copy and pickle.'
return tuple(self)
__dict__ = _property(_asdict)
def __getstate__(self):
'Exclude the OrderedDict from pickling'
pass
{field_defs}
'''
_repr_template = '{name}=%r'
_field_template = '''\
{name} = _property(_itemgetter({index:d}), doc='Alias for field number {index:d}')
'''
def namedtuple(typename, field_names, verbose=False, rename=False):
"""Returns a new subclass of tuple with named fields.
>>> Point = namedtuple('Point', ['x', 'y'])
>>> Point.__doc__ # docstring for the new class
'Point(x, y)'
>>> p = Point(11, y=22) # instantiate with positional args or keywords
>>> p[0] + p[1] # indexable like a plain tuple
33
>>> x, y = p # unpack like a regular tuple
>>> x, y
(11, 22)
>>> p.x + p.y # fields also accessable by name
33
>>> d = p._asdict() # convert to a dictionary
>>> d['x']
11
>>> Point(**d) # convert from a dictionary
Point(x=11, y=22)
>>> p._replace(x=100) # _replace() is like str.replace() but targets named fields
Point(x=100, y=22)
"""
# Validate the field names. At the user's option, either generate an error
# message or automatically replace the field name with a valid name.
if isinstance(field_names, basestring):
field_names = field_names.replace(',', ' ').split()
field_names = map(str, field_names)
typename = str(typename)
if rename:
seen = set()
for index, name in enumerate(field_names):
if (not all(c.isalnum() or c=='_' for c in name)
or _iskeyword(name)
or not name
or name[0].isdigit()
or name.startswith('_')
or name in seen):
field_names[index] = '_%d' % index
seen.add(name)
for name in [typename] + field_names:
if type(name) != str:
raise TypeError('Type names and field names must be strings')
if not all(c.isalnum() or c=='_' for c in name):
raise ValueError('Type names and field names can only contain '
'alphanumeric characters and underscores: %r' % name)
if _iskeyword(name):
raise ValueError('Type names and field names cannot be a '
'keyword: %r' % name)
if name[0].isdigit():
raise ValueError('Type names and field names cannot start with '
'a number: %r' % name)
seen = set()
for name in field_names:
if name.startswith('_') and not rename:
raise ValueError('Field names cannot start with an underscore: '
'%r' % name)
if name in seen:
raise ValueError('Encountered duplicate field name: %r' % name)
seen.add(name)
# Fill-in the class template
class_definition = _class_template.format(
typename = typename,
field_names = tuple(field_names),
num_fields = len(field_names),
arg_list = repr(tuple(field_names)).replace("'", "")[1:-1],
repr_fmt = ', '.join(_repr_template.format(name=name)
for name in field_names),
field_defs = '\n'.join(_field_template.format(index=index, name=name)
for index, name in enumerate(field_names))
)
if verbose:
print class_definition
# Execute the template string in a temporary namespace and support
# tracing utilities by setting a value for frame.f_globals['__name__']
namespace = dict(_itemgetter=_itemgetter, __name__='namedtuple_%s' % typename,
OrderedDict=OrderedDict, _property=property, _tuple=tuple)
try:
exec class_definition in namespace
except SyntaxError as e:
raise SyntaxError(e.message + ':\n' + class_definition)
result = namespace[typename]
# For pickling to work, the __module__ variable needs to be set to the frame
# where the named tuple is created. Bypass this step in environments where
# sys._getframe is not defined (Jython for example) or sys._getframe is not
# defined for arguments greater than 0 (IronPython).
try:
result.__module__ = _sys._getframe(1).f_globals.get('__name__', '__main__')
except (AttributeError, ValueError):
pass
return result
########################################################################
### Counter
########################################################################
class Counter(dict):
'''Dict subclass for counting hashable items. Sometimes called a bag
or multiset. Elements are stored as dictionary keys and their counts
are stored as dictionary values.
>>> c = Counter('abcdeabcdabcaba') # count elements from a string
>>> c.most_common(3) # three most common elements
[('a', 5), ('b', 4), ('c', 3)]
>>> sorted(c) # list all unique elements
['a', 'b', 'c', 'd', 'e']
>>> ''.join(sorted(c.elements())) # list elements with repetitions
'aaaaabbbbcccdde'
>>> sum(c.values()) # total of all counts
15
>>> c['a'] # count of letter 'a'
5
>>> for elem in 'shazam': # update counts from an iterable
... c[elem] += 1 # by adding 1 to each element's count
>>> c['a'] # now there are seven 'a'
7
>>> del c['b'] # remove all 'b'
>>> c['b'] # now there are zero 'b'
0
>>> d = Counter('simsalabim') # make another counter
>>> c.update(d) # add in the second counter
>>> c['a'] # now there are nine 'a'
9
>>> c.clear() # empty the counter
>>> c
Counter()
Note: If a count is set to zero or reduced to zero, it will remain
in the counter until the entry is deleted or the counter is cleared:
>>> c = Counter('aaabbc')
>>> c['b'] -= 2 # reduce the count of 'b' by two
>>> c.most_common() # 'b' is still in, but its count is zero
[('a', 3), ('c', 1), ('b', 0)]
'''
# References:
# http://en.wikipedia.org/wiki/Multiset
# http://www.gnu.org/software/smalltalk/manual-base/html_node/Bag.html
# http://www.demo2s.com/Tutorial/Cpp/0380__set-multiset/Catalog0380__set-multiset.htm
# http://code.activestate.com/recipes/259174/
# Knuth, TAOCP Vol. II section 4.6.3
def __init__(self, iterable=None, **kwds):
'''Create a new, empty Counter object. And if given, count elements
from an input iterable. Or, initialize the count from another mapping
of elements to their counts.
>>> c = Counter() # a new, empty counter
>>> c = Counter('gallahad') # a new counter from an iterable
>>> c = Counter({'a': 4, 'b': 2}) # a new counter from a mapping
>>> c = Counter(a=4, b=2) # a new counter from keyword args
'''
super(Counter, self).__init__()
self.update(iterable, **kwds)
def __missing__(self, key):
'The count of elements not in the Counter is zero.'
# Needed so that self[missing_item] does not raise KeyError
return 0
def most_common(self, n=None):
'''List the n most common elements and their counts from the most
common to the least. If n is None, then list all element counts.
>>> Counter('abcdeabcdabcaba').most_common(3)
[('a', 5), ('b', 4), ('c', 3)]
'''
# Emulate Bag.sortedByCount from Smalltalk
if n is None:
return sorted(self.iteritems(), key=_itemgetter(1), reverse=True)
return _heapq.nlargest(n, self.iteritems(), key=_itemgetter(1))
def elements(self):
'''Iterator over elements repeating each as many times as its count.
>>> c = Counter('ABCABC')
>>> sorted(c.elements())
['A', 'A', 'B', 'B', 'C', 'C']
# Knuth's example for prime factors of 1836: 2**2 * 3**3 * 17**1
>>> prime_factors = Counter({2: 2, 3: 3, 17: 1})
>>> product = 1
>>> for factor in prime_factors.elements(): # loop over factors
... product *= factor # and multiply them
>>> product
1836
Note, if an element's count has been set to zero or is a negative
number, elements() will ignore it.
'''
# Emulate Bag.do from Smalltalk and Multiset.begin from C++.
return _chain.from_iterable(_starmap(_repeat, self.iteritems()))
# Override dict methods where necessary
@classmethod
def fromkeys(cls, iterable, v=None):
# There is no equivalent method for counters because setting v=1
# means that no element can have a count greater than one.
raise NotImplementedError(
'Counter.fromkeys() is undefined. Use Counter(iterable) instead.')
def update(self, iterable=None, **kwds):
'''Like dict.update() but add counts instead of replacing them.
Source can be an iterable, a dictionary, or another Counter instance.
>>> c = Counter('which')
>>> c.update('witch') # add elements from another iterable
>>> d = Counter('watch')
>>> c.update(d) # add elements from another counter
>>> c['h'] # four 'h' in which, witch, and watch
4
'''
# The regular dict.update() operation makes no sense here because the
# replace behavior results in the some of original untouched counts
# being mixed-in with all of the other counts for a mismash that
# doesn't have a straight-forward interpretation in most counting
# contexts. Instead, we implement straight-addition. Both the inputs
# and outputs are allowed to contain zero and negative counts.
if iterable is not None:
if isinstance(iterable, Mapping):
if self:
self_get = self.get
for elem, count in iterable.iteritems():
self[elem] = self_get(elem, 0) + count
else:
super(Counter, self).update(iterable) # fast path when counter is empty
else:
self_get = self.get
for elem in iterable:
self[elem] = self_get(elem, 0) + 1
if kwds:
self.update(kwds)
def subtract(self, iterable=None, **kwds):
'''Like dict.update() but subtracts counts instead of replacing them.
Counts can be reduced below zero. Both the inputs and outputs are
allowed to contain zero and negative counts.
Source can be an iterable, a dictionary, or another Counter instance.
>>> c = Counter('which')
>>> c.subtract('witch') # subtract elements from another iterable
>>> c.subtract(Counter('watch')) # subtract elements from another counter
>>> c['h'] # 2 in which, minus 1 in witch, minus 1 in watch
0
>>> c['w'] # 1 in which, minus 1 in witch, minus 1 in watch
-1
'''
if iterable is not None:
self_get = self.get
if isinstance(iterable, Mapping):
for elem, count in iterable.items():
self[elem] = self_get(elem, 0) - count
else:
for elem in iterable:
self[elem] = self_get(elem, 0) - 1
if kwds:
self.subtract(kwds)
def copy(self):
'Return a shallow copy.'
return self.__class__(self)
def __reduce__(self):
return self.__class__, (dict(self),)
def __delitem__(self, elem):
'Like dict.__delitem__() but does not raise KeyError for missing values.'
if elem in self:
super(Counter, self).__delitem__(elem)
def __repr__(self):
if not self:
return '%s()' % self.__class__.__name__
items = ', '.join(map('%r: %r'.__mod__, self.most_common()))
return '%s({%s})' % (self.__class__.__name__, items)
# Multiset-style mathematical operations discussed in:
# Knuth TAOCP Volume II section 4.6.3 exercise 19
# and at http://en.wikipedia.org/wiki/Multiset
#
# Outputs guaranteed to only include positive counts.
#
# To strip negative and zero counts, add-in an empty counter:
# c += Counter()
def __add__(self, other):
'''Add counts from two counters.
>>> Counter('abbb') + Counter('bcc')
Counter({'b': 4, 'c': 2, 'a': 1})
'''
if not isinstance(other, Counter):
return NotImplemented
result = Counter()
for elem, count in self.items():
newcount = count + other[elem]
if newcount > 0:
result[elem] = newcount
for elem, count in other.items():
if elem not in self and count > 0:
result[elem] = count
return result
def __sub__(self, other):
''' Subtract count, but keep only results with positive counts.
>>> Counter('abbbc') - Counter('bccd')
Counter({'b': 2, 'a': 1})
'''
if not isinstance(other, Counter):
return NotImplemented
result = Counter()
for elem, count in self.items():
newcount = count - other[elem]
if newcount > 0:
result[elem] = newcount
for elem, count in other.items():
if elem not in self and count < 0:
result[elem] = 0 - count
return result
def __or__(self, other):
'''Union is the maximum of value in either of the input counters.
>>> Counter('abbb') | Counter('bcc')
Counter({'b': 3, 'c': 2, 'a': 1})
'''
if not isinstance(other, Counter):
return NotImplemented
result = Counter()
for elem, count in self.items():
other_count = other[elem]
newcount = other_count if count < other_count else count
if newcount > 0:
result[elem] = newcount
for elem, count in other.items():
if elem not in self and count > 0:
result[elem] = count
return result
def __and__(self, other):
''' Intersection is the minimum of corresponding counts.
>>> Counter('abbb') & Counter('bcc')
Counter({'b': 1})
'''
if not isinstance(other, Counter):
return NotImplemented
result = Counter()
for elem, count in self.items():
other_count = other[elem]
newcount = count if count < other_count else other_count
if newcount > 0:
result[elem] = newcount
return result
if __name__ == '__main__':
# verify that instances can be pickled
from cPickle import loads, dumps
Point = namedtuple('Point', 'x, y', True)
p = Point(x=10, y=20)
assert p == loads(dumps(p))
# test and demonstrate ability to override methods
class Point(namedtuple('Point', 'x y')):
__slots__ = ()
@property
def hypot(self):
return (self.x ** 2 + self.y ** 2) ** 0.5
def __str__(self):
return 'Point: x=%6.3f y=%6.3f hypot=%6.3f' % (self.x, self.y, self.hypot)
for p in Point(3, 4), Point(14, 5/7.):
print p
class Point(namedtuple('Point', 'x y')):
'Point class with optimized _make() and _replace() without error-checking'
__slots__ = ()
_make = classmethod(tuple.__new__)
def _replace(self, _map=map, **kwds):
return self._make(_map(kwds.get, ('x', 'y'), self))
print Point(11, 22)._replace(x=100)
Point3D = namedtuple('Point3D', Point._fields + ('z',))
print Point3D.__doc__
import doctest
TestResults = namedtuple('TestResults', 'failed attempted')
print TestResults(*doctest.testmod())
########################################################################
### deque
########################################################################
class deque(_deque):
"""Extension of deque to support Python 2.7's operations."""
def __init__(self, iterable=[], maxlen=None):
_deque.__init__(self, iterable, maxlen)
self._maxlen = maxlen
@property
def maxlen(self):
return self._maxlen
def reverse(self):
data = []
while self:
data.append(self.pop())
self.extend(data)
def count(self, value):
return sum(1 for element in self if element == value)

1
vendor/sekizai/__init__.py vendored Normal file
View file

@ -0,0 +1 @@
__version__ = '0.10.0'

12
vendor/sekizai/context.py vendored Normal file
View file

@ -0,0 +1,12 @@
from django.template import Context
from sekizai.context_processors import sekizai
class SekizaiContext(Context):
"""
An alternative context to be used instead of RequestContext in places where
no request is available.
"""
def __init__(self, *args, **kwargs):
super(SekizaiContext, self).__init__(*args, **kwargs)
self.update(sekizai())

12
vendor/sekizai/context_processors.py vendored Normal file
View file

@ -0,0 +1,12 @@
from collections import defaultdict
from sekizai.data import UniqueSequence
from sekizai.helpers import get_varname
def sekizai(request=None):
"""
Simple context processor which makes sure that the SekizaiDictionary is
available in all templates.
"""
return {get_varname(): defaultdict(UniqueSequence)}

28
vendor/sekizai/data.py vendored Normal file
View file

@ -0,0 +1,28 @@
from collections.abc import MutableSequence
class UniqueSequence(MutableSequence):
def __init__(self):
self.data = []
def __contains__(self, item):
return item in self.data
def __iter__(self):
return iter(self.data)
def __getitem__(self, item):
return self.data[item]
def __setitem__(self, key, value):
self.data[key] = value
def __delitem__(self, key):
del self.data[key]
def __len__(self):
return len(self.data)
def insert(self, index, value):
if value not in self:
self.data.insert(index, value)

177
vendor/sekizai/helpers.py vendored Normal file
View file

@ -0,0 +1,177 @@
# -*- coding: utf-8 -*-
from django.conf import settings
from django.template.base import VariableNode, Variable, Template
from django.template import Context
from django.template.loader import get_template
from django.template.loader_tags import BlockNode, ExtendsNode
try:
from django.template import engines
except ImportError:
engines = None
def _get_nodelist(tpl):
if isinstance(tpl, Template):
return tpl.nodelist
else:
return tpl.template.nodelist
def is_variable_extend_node(node):
if hasattr(node, 'parent_name_expr') and node.parent_name_expr:
return True
if hasattr(node, 'parent_name') and hasattr(node.parent_name, 'filters'):
if (node.parent_name.filters or
isinstance(node.parent_name.var, Variable)):
return True
return False
def get_context():
if engines is not None:
context = Context()
context.template = Template('')
return context
else:
return Context()
def _extend_blocks(extend_node, blocks):
"""
Extends the dictionary `blocks` with *new* blocks in the parent node
(recursive)
"""
# we don't support variable extensions
if is_variable_extend_node(extend_node):
return
parent = extend_node.get_parent(get_context())
# Search for new blocks
for node in _get_nodelist(parent).get_nodes_by_type(BlockNode):
if node.name not in blocks:
blocks[node.name] = node
else:
# set this node as the super node (for {{ block.super }})
block = blocks[node.name]
seen_supers = []
while (hasattr(block.super, 'nodelist') and
block.super not in seen_supers):
seen_supers.append(block.super)
block = block.super
block.super = node
# search for further ExtendsNodes
for node in _get_nodelist(parent).get_nodes_by_type(ExtendsNode):
_extend_blocks(node, blocks)
break
def _extend_nodelist(extend_node):
"""
Returns a list of namespaces found in the parent template(s) of this
ExtendsNode
"""
# we don't support variable extensions (1.3 way)
if is_variable_extend_node(extend_node):
return []
blocks = extend_node.blocks
_extend_blocks(extend_node, blocks)
found = []
for block in blocks.values():
found += _scan_namespaces(block.nodelist, block)
parent_template = extend_node.get_parent(get_context())
# if this is the topmost template, check for namespaces outside of blocks
if not _get_nodelist(parent_template).get_nodes_by_type(ExtendsNode):
found += _scan_namespaces(
_get_nodelist(parent_template),
None
)
else:
found += _scan_namespaces(
_get_nodelist(parent_template),
extend_node
)
return found
def _scan_namespaces(nodelist, current_block=None):
from sekizai.templatetags.sekizai_tags import RenderBlock
found = []
for node in nodelist:
# check if this is RenderBlock node
if isinstance(node, RenderBlock):
# resolve it's name against a dummy context
found.append(node.kwargs['name'].resolve({}))
found += _scan_namespaces(node.blocks['nodelist'], node)
# handle {% extends ... %} tags if check_inheritance is True
elif isinstance(node, ExtendsNode):
found += _extend_nodelist(node)
# in block nodes we have to scan for super blocks
elif isinstance(node, VariableNode) and current_block:
if node.filter_expression.token == 'block.super':
if hasattr(current_block.super, 'nodelist'):
found += _scan_namespaces(
current_block.super.nodelist,
current_block.super
)
return found
def get_namespaces(template):
compiled_template = get_template(template)
return _scan_namespaces(_get_nodelist(compiled_template))
def validate_template(template, namespaces):
"""
Validates that a template (or it's parents if check_inheritance is True)
contain all given namespaces
"""
if getattr(settings, 'SEKIZAI_IGNORE_VALIDATION', False):
return True
found = get_namespaces(template)
for namespace in namespaces:
if namespace not in found:
return False
return True
def get_varname():
return getattr(settings, 'SEKIZAI_VARNAME', 'SEKIZAI_CONTENT_HOLDER')
class Watcher(object):
"""
Watches a context for changes to the sekizai data, so it can be replayed
later. This is useful for caching.
NOTE: This class assumes you ONLY ADD, NEVER REMOVE data from the context!
"""
def __init__(self, context):
self.context = context
self.frozen = dict(
(key, list(value)) for key, value in self.data.items()
)
@property
def data(self):
return self.context.get(get_varname(), {})
def get_changes(self):
sfrozen = set(self.frozen)
sdata = set(self.data)
new_keys = sfrozen ^ sdata
changes = {}
for key in new_keys:
changes[key] = list(self.data[key])
shared_keys = sfrozen & sdata
for key in shared_keys:
old_set = set(self.frozen[key])
new_values = [
item for item in self.data[key] if item not in old_set
]
changes[key] = new_values
return changes

0
vendor/sekizai/models.py vendored Normal file
View file

0
vendor/sekizai/templatetags/__init__.py vendored Normal file
View file

164
vendor/sekizai/templatetags/sekizai_tags.py vendored Normal file
View file

@ -0,0 +1,164 @@
try:
from importlib import import_module
except ImportError:
from django.utils.importlib import import_module
from classytags.arguments import Argument, Flag
from classytags.core import Tag, Options
from classytags.parser import Parser
from django import template
from django.conf import settings
from sekizai.helpers import get_varname
register = template.Library()
def validate_context(context):
"""
Validates a given context.
Returns True if the context is valid.
Returns False if the context is invalid but the error should be silently
ignored.
Raises a TemplateSyntaxError if the context is invalid and we're in debug
mode.
"""
try:
template_debug = context.template.engine.debug
except AttributeError:
try:
# Get the default engine debug value
template_debug = template.Engine.get_default().debug
except AttributeError:
# Django 1.9 and below fallback
template_debug = settings.TEMPLATE_DEBUG
if get_varname() in context:
return True
if not template_debug:
return False
raise template.TemplateSyntaxError(
"You must enable the 'sekizai.context_processors.sekizai' template "
"context processor or use 'sekizai.context.SekizaiContext' to "
"render your templates."
)
def import_processor(import_path):
if '.' not in import_path:
raise TypeError("Import paths must contain at least one '.'")
module_name, object_name = import_path.rsplit('.', 1)
module = import_module(module_name)
return getattr(module, object_name)
class SekizaiParser(Parser):
def parse_blocks(self):
super(SekizaiParser, self).parse_blocks()
self.blocks['nodelist'] = self.parser.parse()
class AddtoblockParser(Parser):
def parse_blocks(self):
name = self.kwargs['name'].var.token
self.blocks['nodelist'] = self.parser.parse(
('endaddtoblock', 'endaddtoblock %s' % name)
)
self.parser.delete_first_token()
class SekizaiTag(Tag):
def render(self, context):
if validate_context(context):
return super(SekizaiTag, self).render(context)
return ''
class RenderBlock(Tag):
name = 'render_block'
options = Options(
Argument('name'),
'postprocessor',
Argument('postprocessor', required=False, default=None, resolve=False),
parser_class=SekizaiParser,
)
def render_tag(self, context, name, postprocessor, nodelist):
if not validate_context(context):
return nodelist.render(context)
rendered_contents = nodelist.render(context)
varname = get_varname()
data = '\n'.join(context[varname][name])
if postprocessor:
func = import_processor(postprocessor)
data = func(context, data, name)
return '%s\n%s' % (data, rendered_contents)
register.tag('render_block', RenderBlock)
class AddData(SekizaiTag):
name = 'add_data'
options = Options(
Argument('key'),
Argument('value'),
)
def render_tag(self, context, key, value):
varname = get_varname()
context[varname][key].append(value)
return ''
register.tag('add_data', AddData)
class WithData(SekizaiTag):
name = 'with_data'
options = Options(
Argument('name'),
'as',
Argument('variable', resolve=False),
blocks=[
('end_with_data', 'inner_nodelist'),
],
parser_class=SekizaiParser,
)
def render_tag(self, context, name, variable, inner_nodelist, nodelist):
rendered_contents = nodelist.render(context)
varname = get_varname()
data = context[varname][name]
context.push()
context[variable] = data
inner_contents = inner_nodelist.render(context)
context.pop()
return '%s\n%s' % (inner_contents, rendered_contents)
register.tag('with_data', WithData)
class Addtoblock(SekizaiTag):
name = 'addtoblock'
options = Options(
Argument('name'),
Flag('strip', default=False, true_values=['strip']),
'preprocessor',
Argument('preprocessor', required=False, default=None, resolve=False),
parser_class=AddtoblockParser,
)
def render_tag(self, context, name, strip, preprocessor, nodelist):
rendered_contents = nodelist.render(context)
if strip:
rendered_contents = rendered_contents.strip()
if preprocessor:
func = import_processor(preprocessor)
rendered_contents = func(context, rendered_contents, name)
varname = get_varname()
context[varname][name].append(rendered_contents)
return ""
register.tag('addtoblock', Addtoblock)

19
vendor/sekizai/test_templates/basic.html vendored Normal file
View file

@ -0,0 +1,19 @@
{% load sekizai_tags %}
{% render_block "css" %}
some content
{% addtoblock "css" %}
my css file
{% endaddtoblock %}
more content
{% addtoblock "js" %}
my js file
{% endaddtoblock %}
final content
{% render_block "js" %}

7
vendor/sekizai/test_templates/css.html vendored Normal file
View file

@ -0,0 +1,7 @@
{% load sekizai_tags %}
{% render_block "css-to-file" %}
{% addtoblock "css-to-file" %}
<style type="text/css">body { color: red; }</style>
{% endaddtoblock %}

6
vendor/sekizai/test_templates/css2.html vendored Normal file
View file

@ -0,0 +1,6 @@
{% load sekizai_tags %}
{% render_block "css-onefile" %}
{% addtoblock "css-onefile" %}<link rel="stylesheet" href="/media/css/one.css" />{% endaddtoblock %}
{% addtoblock "css-onefile" %}<link rel="stylesheet" href="/media/css/two.css" />{% endaddtoblock %}

5
vendor/sekizai/test_templates/easy_base.html vendored Normal file
View file

@ -0,0 +1,5 @@
{% load sekizai_tags %}
{% render_block "css" %}
{% block main %}content{% endblock %}

3
vendor/sekizai/test_templates/easy_inherit.html vendored Normal file
View file

@ -0,0 +1,3 @@
{% extends "easy_base.html" %}
{% block main %}{{ block.super }}{% endblock %}

7
vendor/sekizai/test_templates/eat.html vendored Normal file
View file

@ -0,0 +1,7 @@
{% load sekizai_tags %}
{% addtoblock "css" %}
my css file
{% endaddtoblock %}
mycontent

5
vendor/sekizai/test_templates/errors/failadd.html vendored Normal file
View file

@ -0,0 +1,5 @@
{% load sekizai_tags %}
{% addtoblock %}
file one
{% endaddtoblock %}

9
vendor/sekizai/test_templates/errors/failbase.html vendored Normal file
View file

@ -0,0 +1,9 @@
{% load sekizai_tags %}
{% render_block "js" %}
{% include "errors/failinc.html" %}
{% addtoblock "js" %}
file one
{% endaddtoblock %}

9
vendor/sekizai/test_templates/errors/failbase2.html vendored Normal file
View file

@ -0,0 +1,9 @@
{% load sekizai_tags %}
{% include "errors/failinc.html" %}
{% addtoblock "js" %}
file one
{% endaddtoblock %}
{% render_block "js" %}

5
vendor/sekizai/test_templates/errors/failinc.html vendored Normal file
View file

@ -0,0 +1,5 @@
{% load sekizai_tags %}
{% addtoblock "js %}
file two
{% endaddtoblock %}

3
vendor/sekizai/test_templates/errors/failrender.html vendored Normal file
View file

@ -0,0 +1,3 @@
{% load sekizai_tags %}
{% render_block %}

29
vendor/sekizai/test_templates/inherit/base.html vendored Normal file
View file

@ -0,0 +1,29 @@
{% load sekizai_tags %}
head start
{% render_block "css" %}
head end
include start
{% include "inherit/baseinc.html" %}
include end
block main start
{% block "main" %}
block main base contents
{% endblock %}
block main end
body pre-end
{% render_block "js" %}
body end
{% addtoblock "css" %}
some css file
{% endaddtoblock %}

7
vendor/sekizai/test_templates/inherit/baseinc.html vendored Normal file
View file

@ -0,0 +1,7 @@
{% load sekizai_tags %}
inc add js
{% addtoblock "js" %}
inc js file
{% endaddtoblock %}

7
vendor/sekizai/test_templates/inherit/chain.html vendored Normal file
View file

@ -0,0 +1,7 @@
{% extends "inherit/extend.html" %}
{% block "main" %}
{{ block.super }}
{{ other_var }}
{% endblock %}

11
vendor/sekizai/test_templates/inherit/extend.html vendored Normal file
View file

@ -0,0 +1,11 @@
{% extends "inherit/base.html" %}
{% load sekizai_tags %}
{% addtoblock "css" %}
invisible css file
{% endaddtoblock %}
{% block "main" %}
{% include "inherit/extinc.html" %}
{% endblock %}

3
vendor/sekizai/test_templates/inherit/extinc.html vendored Normal file
View file

@ -0,0 +1,3 @@
{% load sekizai_tags %}
extinc

0
vendor/sekizai/test_templates/inherit/nullbase.html vendored Normal file
View file

5
vendor/sekizai/test_templates/inherit/nullext.html vendored Normal file
View file

@ -0,0 +1,5 @@
{% extends "inherit/nullbase.html" %}
{% load sekizai_tags %}
{% render_block "js" %}

1
vendor/sekizai/test_templates/inherit/spacechain.html vendored Normal file
View file

@ -0,0 +1 @@
{% extends "inherit/extend.html" %}

1
vendor/sekizai/test_templates/inherit/subvarchain.html vendored Normal file
View file

@ -0,0 +1 @@
{% extends "inherit/varchain.html" %}

15
vendor/sekizai/test_templates/inherit/super_blocks.html vendored Normal file
View file

@ -0,0 +1,15 @@
{% extends "inherit/base.html" %}
{% load sekizai_tags %}
{% addtoblock "css" %}
invisible css file
{% endaddtoblock %}
{% block "main" %}
{{ block.super }}
{% addtoblock "css" %}
visible css file
{% endaddtoblock %}
more contents
{% endblock %}

2
vendor/sekizai/test_templates/inherit/varchain.html vendored Normal file
View file

@ -0,0 +1,2 @@
{% extends var %}

7
vendor/sekizai/test_templates/named_end.html vendored Normal file
View file

@ -0,0 +1,7 @@
{% load sekizai_tags %}
{% addtoblock "myblock" %}
mycontent
{% endaddtoblock "myblock" %}
{% render_block "myblock" %}

16
vendor/sekizai/test_templates/namespaces.html vendored Normal file
View file

@ -0,0 +1,16 @@
{% load sekizai_tags %}
{% render_block "one" %}
{% render_block "two" %}
{% addtoblock "one" %}
the same file
{% endaddtoblock %}
{% addtoblock "two" %}
the same file
{% endaddtoblock %}
{% addtoblock "one" %}
the same file
{% endaddtoblock %}

11
vendor/sekizai/test_templates/processors/addtoblock_namespace.html vendored Normal file
View file

@ -0,0 +1,11 @@
{% load sekizai_tags %}
header
{% addtoblock "js" preprocessor "sekizai.tests.namespace_processor" %}
javascript
{% endaddtoblock %}
footer
{% render_block "js" %}

11
vendor/sekizai/test_templates/processors/addtoblock_null.html vendored Normal file
View file

@ -0,0 +1,11 @@
{% load sekizai_tags %}
header
{% addtoblock "js" preprocessor "sekizai.tests.null_processor" %}
javascript
{% endaddtoblock %}
footer
{% render_block "js" %}

11
vendor/sekizai/test_templates/processors/namespace.html vendored Normal file
View file

@ -0,0 +1,11 @@
{% load sekizai_tags %}
header
{% addtoblock "js" %}
javascript
{% endaddtoblock %}
footer
{% render_block "js" postprocessor "sekizai.tests.namespace_processor" %}

11
vendor/sekizai/test_templates/processors/null.html vendored Normal file
View file

@ -0,0 +1,11 @@
{% load sekizai_tags %}
header
{% addtoblock "js" %}
javascript
{% endaddtoblock %}
footer
{% render_block "js" postprocessor "sekizai.tests.null_processor" %}

6
vendor/sekizai/test_templates/unique.html vendored Normal file
View file

@ -0,0 +1,6 @@
{% load sekizai_tags %}
{% render_block "unique" %}
{% addtoblock "unique" %}unique data{% endaddtoblock %}
{% addtoblock "unique" %}unique data{% endaddtoblock %}

16
vendor/sekizai/test_templates/variables.html vendored Normal file
View file

@ -0,0 +1,16 @@
{% load sekizai_tags %}
{% addtoblock blockname %}
file two
{% endaddtoblock %}
{% addtoblock "one" %}
file two
{% endaddtoblock %}
{% addtoblock blockname|upper %}
file one
{% endaddtoblock %}
{% render_block "ONE" %}
{% render_block blockname %}

9
vendor/sekizai/test_templates/with_data.html vendored Normal file
View file

@ -0,0 +1,9 @@
{% load sekizai_tags %}
{% with_data "key" as mylist %}
{% for obj in mylist %}
{{ obj }}
{% endfor %}
{% end_with_data %}
{% add_data "key" 1 %}
{% add_data "key" 2 %}

483
vendor/sekizai/tests.py vendored Normal file
View file

@ -0,0 +1,483 @@
from __future__ import with_statement
from difflib import SequenceMatcher
import os
from unittest import TestCase
import sys
import django
from django import template
from django.conf import settings
from django.template.loader import render_to_string
import pep8
from sekizai.context import SekizaiContext
from sekizai.helpers import get_namespaces
from sekizai.helpers import get_varname
from sekizai.helpers import validate_template
from sekizai.helpers import Watcher
from sekizai.templatetags.sekizai_tags import import_processor
from sekizai.templatetags.sekizai_tags import validate_context
try:
unicode_compat = unicode
except NameError:
unicode_compat = str
try:
from io import StringIO
except ImportError:
from StringIO import StringIO
def null_processor(context, data, namespace):
return ''
def namespace_processor(context, data, namespace):
return namespace
class SettingsOverride(object):
"""
Overrides Django settings within a context and resets them to their inital
values on exit.
Example:
with SettingsOverride(DEBUG=True):
# do something
"""
class NULL:
pass
def __init__(self, **overrides):
self.overrides = overrides
def __enter__(self):
self.old = {}
for key, value in self.overrides.items():
self.old[key] = getattr(settings, key, self.NULL)
setattr(settings, key, value)
def __exit__(self, type, value, traceback):
for key, value in self.old.items():
if value is self.NULL:
delattr(settings, key)
else:
setattr(settings, key, value)
class CaptureStdout(object):
"""
Overrides sys.stdout with a StringIO stream.
"""
def __init__(self):
self.old = None
def __enter__(self):
self.old = sys.stdout
new = sys.stdout = StringIO()
return new
def __exit__(self, exc_type, exc_val, exc_tb):
sys.stdout = self.old
class Match(tuple): # pragma: no cover
@property
def a(self):
return self[0]
@property
def b(self):
return self[1]
@property
def size(self):
return self[2]
def _backwards_compat_match(thing): # pragma: no cover
if isinstance(thing, tuple):
return Match(thing)
return thing
class BitDiffResult(object):
def __init__(self, status, message):
self.status = status
self.message = message
class BitDiff(object):
"""
Visual aid for failing tests
"""
def __init__(self, expected):
self.expected = [repr(unicode_compat(bit)) for bit in expected]
def test(self, result):
result = [repr(unicode_compat(bit)) for bit in result]
if self.expected == result:
return BitDiffResult(True, "success")
else: # pragma: no cover
longest = max(
[len(x) for x in self.expected] +
[len(x) for x in result] +
[len('Expected')]
)
sm = SequenceMatcher()
sm.set_seqs(self.expected, result)
matches = sm.get_matching_blocks()
lasta = 0
lastb = 0
data = []
for match in [_backwards_compat_match(match) for match in matches]:
unmatcheda = self.expected[lasta:match.a]
unmatchedb = result[lastb:match.b]
unmatchedlen = max([len(unmatcheda), len(unmatchedb)])
unmatcheda += ['' for x in range(unmatchedlen)]
unmatchedb += ['' for x in range(unmatchedlen)]
for i in range(unmatchedlen):
data.append((False, unmatcheda[i], unmatchedb[i]))
for i in range(match.size):
data.append((
True, self.expected[match.a + i], result[match.b + i]
))
lasta = match.a + match.size
lastb = match.b + match.size
padlen = (longest - len('Expected'))
padding = ' ' * padlen
line1 = '-' * padlen
line2 = '-' * (longest - len('Result'))
msg = '\nExpected%s | | Result' % padding
msg += '\n--------%s-|---|-------%s' % (line1, line2)
for success, a, b in data:
pad = ' ' * (longest - len(a))
if success:
msg += '\n%s%s | | %s' % (a, pad, b)
else:
msg += '\n%s%s | ! | %s' % (a, pad, b)
return BitDiffResult(False, msg)
def update_template_debug(debug=True):
"""
Helper method for updating the template debug option based on
the django version. Use the results of this function as the context.
:return: SettingsOverride object
"""
if django.VERSION[0] == 1 and django.VERSION[1] < 8:
return SettingsOverride(TEMPLATE_DEBUG=debug)
else:
# Create our overridden template settings with debug turned off.
templates_override = settings.TEMPLATES
templates_override[0]['OPTIONS'].update({
'debug': debug
})
from django.template.engine import Engine
# Engine gets created based on template settings initial value so
# changing the settings after the fact won't update, so do it
# manually. Necessary when testing validate_context
# with render method and want debug off.
Engine.get_default().debug = debug
return SettingsOverride(TEMPLATES=templates_override)
class SekizaiTestCase(TestCase):
@classmethod
def setUpClass(cls):
cls._template_dirs = settings.TEMPLATE_DIRS
template_dir = os.path.join(
os.path.dirname(__file__),
'test_templates'
)
settings.TEMPLATE_DIRS = list(cls._template_dirs) + [template_dir]
@classmethod
def tearDownClass(cls):
settings.TEMPLATE_DIRS = cls._template_dirs
def _render(self, tpl, ctx=None, ctxclass=SekizaiContext):
ctx = ctx or {}
return render_to_string(tpl, ctxclass(ctx))
def _get_bits(self, tpl, ctx=None, ctxclass=SekizaiContext):
ctx = ctx or {}
rendered = self._render(tpl, ctx, ctxclass)
bits = [
bit for bit in [bit.strip('\n')
for bit in rendered.split('\n')] if bit
]
return bits, rendered
def _test(self, tpl, res, ctx=None, ctxclass=SekizaiContext):
"""
Helper method to render template and compare it's bits
"""
ctx = ctx or {}
bits, rendered = self._get_bits(tpl, ctx, ctxclass)
differ = BitDiff(res)
result = differ.test(bits)
self.assertTrue(result.status, result.message)
return rendered
def test_pep8(self):
sekizai_dir = os.path.dirname(os.path.abspath(__file__))
pep8style = pep8.StyleGuide()
with CaptureStdout() as stdout:
result = pep8style.check_files([sekizai_dir])
errors = stdout.getvalue()
self.assertEqual(
result.total_errors, 0,
"Code not PEP8 compliant:\n{0}".format(errors)
)
def test_basic_dual_block(self):
"""
Basic dual block testing
"""
bits = [
'my css file', 'some content', 'more content', 'final content',
'my js file'
]
self._test('basic.html', bits)
def test_named_endaddtoblock(self):
"""
Testing with named endaddblock
"""
bits = ["mycontent"]
self._test('named_end.html', bits)
def test_eat_content_before_render_block(self):
"""
Testing that content get's eaten if no render_blocks is available
"""
bits = ["mycontent"]
self._test("eat.html", bits)
def test_sekizai_context_required(self):
"""
Test that the template tags properly fail if not used with either
SekizaiContext or the context processor.
"""
self.assertRaises(
template.TemplateSyntaxError,
self._render, 'basic.html', {}, template.Context
)
def test_complex_template_inheritance(self):
"""
Test that (complex) template inheritances work properly
"""
bits = [
"head start",
"some css file",
"head end",
"include start",
"inc add js",
"include end",
"block main start",
"extinc",
"block main end",
"body pre-end",
"inc js file",
"body end"
]
self._test("inherit/extend.html", bits)
"""
Test that blocks (and block.super) work properly with sekizai
"""
bits = [
"head start",
"visible css file",
"some css file",
"head end",
"include start",
"inc add js",
"include end",
"block main start",
"block main base contents",
"more contents",
"block main end",
"body pre-end",
"inc js file",
"body end"
]
self._test("inherit/super_blocks.html", bits)
def test_namespace_isolation(self):
"""
Tests that namespace isolation works
"""
bits = ["the same file", "the same file"]
self._test('namespaces.html', bits)
def test_variable_namespaces(self):
"""
Tests variables and filtered variables as block names.
"""
bits = ["file one", "file two"]
self._test('variables.html', bits, {'blockname': 'one'})
def test_invalid_addtoblock(self):
"""
Tests that template syntax errors are raised properly in templates
rendered by sekizai tags
"""
self.assertRaises(
template.TemplateSyntaxError,
self._render, 'errors/failadd.html'
)
def test_invalid_renderblock(self):
self.assertRaises(
template.TemplateSyntaxError,
self._render, 'errors/failrender.html'
)
def test_invalid_include(self):
self.assertRaises(
template.TemplateSyntaxError,
self._render, 'errors/failinc.html'
)
def test_invalid_basetemplate(self):
self.assertRaises(
template.TemplateSyntaxError,
self._render, 'errors/failbase.html'
)
def test_invalid_basetemplate_two(self):
self.assertRaises(
template.TemplateSyntaxError,
self._render, 'errors/failbase2.html'
)
def test_with_data(self):
"""
Tests the with_data/add_data tags.
"""
bits = ["1", "2"]
self._test('with_data.html', bits)
def test_easy_inheritance(self):
self.assertEqual('content', self._render("easy_inherit.html").strip())
def test_validate_context(self):
sekizai_ctx = SekizaiContext()
django_ctx = template.Context()
self.assertRaises(
template.TemplateSyntaxError,
validate_context, django_ctx
)
self.assertEqual(validate_context(sekizai_ctx), True)
with update_template_debug(debug=False):
self.assertEqual(validate_context(django_ctx), False)
self.assertEqual(validate_context(sekizai_ctx), True)
bits = ['some content', 'more content', 'final content']
self._test('basic.html', bits, ctxclass=template.Context)
def test_post_processor_null(self):
bits = ['header', 'footer']
self._test('processors/null.html', bits)
def test_post_processor_namespace(self):
bits = ['header', 'footer', 'js']
self._test('processors/namespace.html', bits)
def test_import_processor_failfast(self):
self.assertRaises(TypeError, import_processor, 'invalidpath')
def test_unique(self):
bits = ['unique data']
self._test('unique.html', bits)
def test_strip(self):
tpl = template.Template("""
{% load sekizai_tags %}
{% addtoblock 'a' strip %} test{% endaddtoblock %}
{% addtoblock 'a' strip %}test {% endaddtoblock %}
{% render_block 'a' %}""")
context = SekizaiContext()
output = tpl.render(context)
self.assertEqual(output.count('test'), 1, output)
def test_addtoblock_processor_null(self):
bits = ['header', 'footer']
self._test('processors/addtoblock_null.html', bits)
def test_addtoblock_processor_namespace(self):
bits = ['header', 'footer', 'js']
self._test('processors/addtoblock_namespace.html', bits)
class HelperTests(TestCase):
def test_validate_template_js_css(self):
self.assertTrue(validate_template('basic.html', ['js', 'css']))
def test_validate_template_js(self):
self.assertTrue(validate_template('basic.html', ['js']))
def test_validate_template_css(self):
self.assertTrue(validate_template('basic.html', ['css']))
def test_validate_template_empty(self):
self.assertTrue(validate_template('basic.html', []))
def test_validate_template_notfound(self):
self.assertFalse(validate_template('basic.html', ['notfound']))
def test_get_namespaces_easy_inherit(self):
self.assertEqual(get_namespaces('easy_inherit.html'), ['css'])
def test_get_namespaces_chain_inherit(self):
self.assertEqual(get_namespaces('inherit/chain.html'), ['css', 'js'])
def test_get_namespaces_space_chain_inherit(self):
self.assertEqual(
get_namespaces('inherit/spacechain.html'),
['css', 'js']
)
def test_get_namespaces_var_inherit(self):
self.assertEqual(get_namespaces('inherit/varchain.html'), [])
def test_get_namespaces_sub_var_inherit(self):
self.assertEqual(get_namespaces('inherit/subvarchain.html'), [])
def test_get_namespaces_null_ext(self):
self.assertEqual(get_namespaces('inherit/nullext.html'), [])
def test_deactivate_validate_template(self):
with SettingsOverride(SEKIZAI_IGNORE_VALIDATION=True):
self.assertTrue(validate_template('basic.html', ['js', 'css']))
self.assertTrue(validate_template('basic.html', ['js']))
self.assertTrue(validate_template('basic.html', ['css']))
self.assertTrue(validate_template('basic.html', []))
self.assertTrue(validate_template('basic.html', ['notfound']))
def test_watcher_add_namespace(self):
context = SekizaiContext()
watcher = Watcher(context)
varname = get_varname()
context[varname]['key'].append('value')
changes = watcher.get_changes()
self.assertEqual(changes, {'key': ['value']})
def test_watcher_add_data(self):
context = SekizaiContext()
varname = get_varname()
context[varname]['key'].append('value')
watcher = Watcher(context)
context[varname]['key'].append('value2')
changes = watcher.get_changes()
self.assertEqual(changes, {'key': ['value2']})

11
vendor/stored_messages/__init__.py vendored Normal file
View file

@ -0,0 +1,11 @@
# flake8: noqa
from __future__ import absolute_import
from django.contrib.messages.api import *
from stored_messages.api import *
from stored_messages.constants import *
__version__ = '1.4.0'
VERSION = __version__ # synonym

7
vendor/stored_messages/admin.py vendored Normal file
View file

@ -0,0 +1,7 @@
from django.contrib import admin
from .models import Inbox, Message, MessageArchive
admin.site.register(Inbox)
admin.site.register(Message)
admin.site.register(MessageArchive)

69
vendor/stored_messages/api.py vendored Normal file
View file

@ -0,0 +1,69 @@
from .settings import stored_messages_settings
__all__ = (
'add_message_for',
'broadcast_message',
'mark_read',
'mark_all_read',
)
def add_message_for(users, level, message_text, extra_tags='', date=None, url=None, fail_silently=False):
"""
Send a message to a list of users without passing through `django.contrib.messages`
:param users: an iterable containing the recipients of the messages
:param level: message level
:param message_text: the string containing the message
:param extra_tags: like the Django api, a string containing extra tags for the message
:param date: a date, different than the default timezone.now
:param url: an optional url
:param fail_silently: not used at the moment
"""
BackendClass = stored_messages_settings.STORAGE_BACKEND
backend = BackendClass()
m = backend.create_message(level, message_text, extra_tags, date, url)
backend.archive_store(users, m)
backend.inbox_store(users, m)
def broadcast_message(level, message_text, extra_tags='', date=None, url=None, fail_silently=False):
"""
Send a message to all users aka broadcast.
:param level: message level
:param message_text: the string containing the message
:param extra_tags: like the Django api, a string containing extra tags for the message
:param date: a date, different than the default timezone.now
:param url: an optional url
:param fail_silently: not used at the moment
"""
from django.contrib.auth import get_user_model
users = get_user_model().objects.all()
add_message_for(users, level, message_text, extra_tags=extra_tags, date=date, url=url, fail_silently=fail_silently)
def mark_read(user, message):
"""
Mark message instance as read for user.
Returns True if the message was `unread` and thus actually marked as `read` or False in case
it is already `read` or it does not exist at all.
:param user: user instance for the recipient
:param message: a Message instance to mark as read
"""
BackendClass = stored_messages_settings.STORAGE_BACKEND
backend = BackendClass()
backend.inbox_delete(user, message)
def mark_all_read(user):
"""
Mark all message instances for a user as read.
:param user: user instance for the recipient
"""
BackendClass = stored_messages_settings.STORAGE_BACKEND
backend = BackendClass()
backend.inbox_purge(user)

3
vendor/stored_messages/backends/__init__.py vendored Normal file
View file

@ -0,0 +1,3 @@
# flake8: noqa
from .default.backend import DefaultBackend
from .redis.backend import RedisBackend

159
vendor/stored_messages/backends/base.py vendored Normal file
View file

@ -0,0 +1,159 @@
class StoredMessagesBackend(object):
"""
"""
def create_message(self, level, msg_text, extra_tags, date=None):
"""
Create and return a `Message` instance.
Instance types depend on backends implementation.
Params:
`level`: message level (see django.contrib.messages)
`msg_text`: what you think it is
`extra_tags`: see django.contrib.messages
`date`: a DateTime (optional)
Return:
`Message` instance
"""
raise NotImplementedError()
def inbox_list(self, user):
"""
Retrieve all the messages in `user`'s Inbox.
Params:
`user`: Django User instance
Return:
An iterable containing `Message` instances
"""
raise NotImplementedError()
def inbox_purge(self, user):
"""
Delete all the messages in `user`'s Inbox.
Params:
user: Django User instance
Return:
None
"""
raise NotImplementedError()
def inbox_store(self, users, msg_instance):
"""
Store a `Message` instance in the inbox for a list
of users.
Params:
users: a list or iterable containing Django User instances
msg_instance: Message instance to persist in inbox
Return:
None
Raise:
MessageTypeNotSupported if `msg_instance` cannot be managed by current backend
"""
raise NotImplementedError()
def inbox_delete(self, user, msg_id):
"""
Remove a `Message` instance from `user`'s inbox.
Params:
user: Django User instance
msg_id: Message identifier
Return:
None
Raise:
MessageDoesNotExist if msg_id was not found
"""
raise NotImplementedError()
def inbox_get(self, user, msg_id):
"""
Retrieve a `Message` instance from `user`'s inbox.
Params:
user: Django User instance
msg_id: Message identifier
Return:
A `Message` instance
Raise:
MessageDoesNotExist if msg_id was not found
"""
def archive_store(self, users, msg_instance):
"""
Store a `Message` instance in the archive for a list
of users.
Params:
users: a list or iterable containing Django User instances
msg_instance: Message instance to persist in archive
Return:
None
Raise:
MessageTypeNotSupported if `msg_instance` cannot be managed by current backend
"""
raise NotImplementedError()
def archive_list(self, user):
"""
Retrieve all the messages in `user`'s archive.
Params:
user: Django User instance
Return:
An iterable containing `Message` instances
"""
raise NotImplementedError()
def can_handle(self, msg_instance):
"""
Determine if this backend can handle messages
of the same type of `msg_instance`.
Params:
`msg_instance`: `Message` instance
Return:
True if type is correct, False otherwise
"""
raise NotImplementedError()
def expired_messages_cleanup(self):
"""
Remove messages that have been expired.
Params:
None
Return:
None
"""
raise NotImplementedError()
def _flush(self):
"""
Clear all backend data.
Warning: heavily destructive! Here for convenience, not used by the API anyway.
Params:
None
Return:
None
"""
raise NotImplementedError()

2
vendor/stored_messages/backends/default/__init__.py vendored Normal file
View file

@ -0,0 +1,2 @@
# flake8: noqa
from .backend import DefaultBackend

81
vendor/stored_messages/backends/default/backend.py vendored Normal file
View file

@ -0,0 +1,81 @@
from django.utils import timezone
from ..base import StoredMessagesBackend
from ..exceptions import MessageTypeNotSupported, MessageDoesNotExist
from .. import signals
from ...models import Inbox, Message, MessageArchive
from ...settings import stored_messages_settings
class DefaultBackend(StoredMessagesBackend):
"""
"""
def inbox_list(self, user):
if user.is_anonymous():
return []
inbox = Inbox.objects.filter(user=user).select_related("message")
return [m.message for m in inbox]
def inbox_purge(self, user):
if user.is_authenticated():
Inbox.objects.filter(user=user).delete()
signals.inbox_purged.send(sender=self.__class__, user=user)
def inbox_store(self, users, msg_instance):
if not self.can_handle(msg_instance):
raise MessageTypeNotSupported()
for user in users:
Inbox.objects.get_or_create(user=user, message=msg_instance)
signals.inbox_stored.send(sender=self.__class__, user=user, message=msg_instance)
def inbox_delete(self, user, msg_id):
try:
inbox_m = Inbox.objects.filter(user=user, message=msg_id).get()
inbox_m.delete()
signals.inbox_deleted.send(sender=self.__class__, user=user, message_id=msg_id)
except Inbox.DoesNotExist:
raise MessageDoesNotExist("Message with id %s does not exist" % msg_id)
def inbox_get(self, user, msg_id):
try:
return Inbox.objects.get(pk=msg_id).message
except Inbox.DoesNotExist:
raise MessageDoesNotExist("Message with id %s does not exist" % msg_id)
def create_message(self, level, msg_text, extra_tags='', date=None, url=None):
kwargs = {
'message': msg_text,
'level': level,
'tags': extra_tags,
'url': url,
}
if date:
kwargs['date'] = date
m_instance = Message.objects.create(**kwargs)
return m_instance
def archive_store(self, users, msg_instance):
if not self.can_handle(msg_instance):
raise MessageTypeNotSupported()
for user in users:
MessageArchive.objects.create(user=user, message=msg_instance)
signals.archive_stored.send(sender=self.__class__, user=user, message=msg_instance)
def archive_list(self, user):
return list(MessageArchive.objects.filter(user=user))
def can_handle(self, message):
return isinstance(message, Message)
def expired_messages_cleanup(self):
expiration_date = timezone.now() + timezone.timedelta(
days=-stored_messages_settings.MESSAGE_EXPIRE_DAYS)
Message.objects.filter(date__lte=expiration_date).delete()
def _flush(self):
Inbox.objects.all().delete()
MessageArchive.objects.all().delete()
Message.objects.all().delete()

8
vendor/stored_messages/backends/exceptions.py vendored Normal file
View file

@ -0,0 +1,8 @@
class MessageTypeNotSupported(Exception):
pass
class MessageDoesNotExist(Exception):
pass

2
vendor/stored_messages/backends/redis/__init__.py vendored Normal file
View file

@ -0,0 +1,2 @@
# flake8: noqa
from .backend import RedisBackend, Message

143
vendor/stored_messages/backends/redis/backend.py vendored Normal file
View file

@ -0,0 +1,143 @@
from __future__ import unicode_literals
from django.utils import timezone
from django.utils.encoding import force_text
from django.core.serializers.json import DjangoJSONEncoder
import json
from collections import namedtuple
import hashlib
from ..exceptions import MessageTypeNotSupported, MessageDoesNotExist
from ..base import StoredMessagesBackend
from .. import signals
from ...settings import stored_messages_settings
try:
# Let django project bootstrap anyway when not using this backend
import redis
except ImportError:
pass
Message = namedtuple('Message', ['id', 'message', 'level', 'tags', 'date', 'url'])
class RedisBackend(StoredMessagesBackend):
"""
"""
def __init__(self):
self.client = redis.StrictRedis.from_url(stored_messages_settings.REDIS_URL)
def _flush(self):
self.client.flushdb()
def _toJSON(self, msg_instance):
"""
Dump a Message instance into a JSON string
"""
return json.dumps(msg_instance._asdict(), cls=DjangoJSONEncoder)
def _fromJSON(self, json_msg):
"""
Return a Message instance built from data contained in a JSON string
"""
return Message(**json.loads(force_text(json_msg)))
def _list_key(self, key):
"""
boilerplate
"""
ret = []
for msg_json in self.client.lrange(key, 0, -1):
ret.append(self._fromJSON(msg_json))
return ret
def _list(self, key_tpl, user):
return self._list_key(key_tpl % user.pk)
def create_message(self, level, msg_text, extra_tags='', date=None, url=None):
"""
Message instances are namedtuples of type `Message`.
The date field is already serialized in datetime.isoformat ECMA-262 format
"""
if not date:
now = timezone.now()
else:
now = date
r = now.isoformat()
if now.microsecond:
r = r[:23] + r[26:]
if r.endswith('+00:00'):
r = r[:-6] + 'Z'
fingerprint = r + msg_text
msg_id = hashlib.sha256(fingerprint.encode('ascii', 'ignore')).hexdigest()
return Message(id=msg_id, message=msg_text, level=level, tags=extra_tags, date=r, url=url)
def inbox_list(self, user):
if user.is_anonymous():
return []
return self._list('user:%d:notifications', user)
def inbox_purge(self, user):
if user.is_authenticated():
self.client.delete('user:%d:notifications' % user.pk)
self.client.delete('user:%d:notificationsidx' % user.pk)
signals.inbox_purged.send(sender=self.__class__, user=user)
def inbox_store(self, users, msg_instance):
if not self.can_handle(msg_instance):
raise MessageTypeNotSupported()
for user in users:
if self.client.sismember('user:%d:notificationsidx' % user.pk, msg_instance.id):
return # a duplicate, NOOP
else:
self.client.sadd('user:%d:notificationsidx' % user.pk, msg_instance.id)
self.client.rpush('user:%d:notifications' % user.pk, self._toJSON(msg_instance))
signals.inbox_stored.send(sender=self.__class__, user=user, message=msg_instance)
def inbox_delete(self, user, msg_id):
for m in self._list('user:%d:notifications', user):
if m.id == msg_id:
msg = self.client.lrem('user:%d:notifications' % user.pk, 0, json.dumps(m._asdict()))
signals.inbox_deleted.send(sender=self.__class__, user=user, message_id=msg_id)
return msg
raise MessageDoesNotExist("Message with id %s does not exist" % msg_id)
def inbox_get(self, user, msg_id):
for m in self._list('user:%d:notifications', user):
if m.id == msg_id:
return m
raise MessageDoesNotExist("Message with id %s does not exist" % msg_id)
def archive_store(self, users, msg_instance):
if not self.can_handle(msg_instance):
raise MessageTypeNotSupported()
for user in users:
self.client.rpush('user:%d:archive' % user.pk, self._toJSON(msg_instance))
signals.archive_stored.send(sender=self.__class__, user=user, message=msg_instance)
def archive_list(self, user):
return self._list('user:%d:archive', user)
def can_handle(self, msg_instance):
return isinstance(msg_instance, Message)
def expired_messages_cleanup(self):
expiration_date = timezone.now() + timezone.timedelta(
days=-stored_messages_settings.MESSAGE_EXPIRE_DAYS)
keys = self.client.keys('user:*:notifications')
for k in keys:
_, user, _ = k.split(':')
for m in self._list_key(k):
if m.date <= expiration_date:
msg = json.dumps(m._asdict())
self.client.lrem(k, 0, msg)
self.client.srem('user:%s:notificationsidx' % user, m.id)
self.client.rpop('user:%d:archive' % user, msg)

7
vendor/stored_messages/backends/signals.py vendored Normal file
View file

@ -0,0 +1,7 @@
from django.dispatch import Signal
inbox_stored = Signal(providing_args=["user", "message"])
inbox_deleted = Signal(providing_args=["user", "message_id"])
inbox_purged = Signal(providing_args=["user"])
archive_stored = Signal(providing_args=["user", "message"])

6
vendor/stored_messages/compat.py vendored Normal file
View file

@ -0,0 +1,6 @@
"""
The `compat` module provides support for backwards compatibility with older
versions of Django and Python. Please, add in this module the compatibility code.
"""
from __future__ import unicode_literals

16
vendor/stored_messages/constants.py vendored Normal file
View file

@ -0,0 +1,16 @@
from django.contrib.messages.constants import *
STORED_DEBUG = DEBUG + 1
STORED_INFO = INFO + 1
STORED_SUCCESS = SUCCESS + 1
STORED_WARNING = WARNING + 1
STORED_ERROR = ERROR + 1
DEFAULT_TAGS.update({
STORED_DEBUG: 'persisted debug',
STORED_INFO: 'persisted info',
STORED_SUCCESS: 'persisted success',
STORED_WARNING: 'persisted warning',
STORED_ERROR: 'persisted error',
})

54
vendor/stored_messages/migrations/0001_initial.py vendored Normal file
View file

@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import django.utils.timezone
from django.db import models, migrations
from django.conf import settings
class Migration(migrations.Migration):
dependencies = [
migrations.swappable_dependency(settings.AUTH_USER_MODEL),
]
operations = [
migrations.CreateModel(
name='Inbox',
fields=[
('id', models.AutoField(auto_created=True, serialize=False, primary_key=True, verbose_name='ID')),
],
options={
'verbose_name_plural': 'inboxes',
},
),
migrations.CreateModel(
name='Message',
fields=[
('id', models.AutoField(auto_created=True, serialize=False, primary_key=True, verbose_name='ID')),
('message', models.TextField()),
('level', models.IntegerField()),
('tags', models.TextField()),
('date', models.DateTimeField(default=django.utils.timezone.now)),
],
),
migrations.CreateModel(
name='MessageArchive',
fields=[
('id', models.AutoField(auto_created=True, serialize=False, primary_key=True, verbose_name='ID')),
('message', models.ForeignKey(to='stored_messages.Message')),
('user', models.ForeignKey(to=settings.AUTH_USER_MODEL)),
],
),
migrations.AddField(
model_name='inbox',
name='message',
field=models.ForeignKey(to='stored_messages.Message'),
),
migrations.AddField(
model_name='inbox',
name='user',
field=models.ForeignKey(to=settings.AUTH_USER_MODEL),
),
]

19
vendor/stored_messages/migrations/0002_message_url.py vendored Normal file
View file

@ -0,0 +1,19 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import models, migrations
class Migration(migrations.Migration):
dependencies = [
('stored_messages', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='message',
name='url',
field=models.URLField(blank=True, null=True),
),
]

28
vendor/stored_messages/migrations/0003_alter_inbox_id_alter_message_id_and_more.py vendored Normal file
View file

@ -0,0 +1,28 @@
# Generated by Django 6.1.1 on 2026-10-01 14:16
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('stored_messages', '0002_message_url'),
]
operations = [
migrations.AlterField(
model_name='inbox',
name='id',
field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
),
migrations.AlterField(
model_name='message',
name='id',
field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
),
migrations.AlterField(
model_name='messagearchive',
name='id',
field=models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
),
]

0
vendor/stored_messages/migrations/__init__.py vendored Normal file
View file

66
vendor/stored_messages/models.py vendored Normal file
View file

@ -0,0 +1,66 @@
from __future__ import unicode_literals
from django.db import models
from django.conf import settings
from django.utils import timezone
try:
from django.utils.encoding import python_2_unicode_compatible
except ImportError:
def python_2_unicode_compatible(cls):
return cls
from django.utils.translation import pgettext_lazy as _
from .settings import stored_messages_settings
@python_2_unicode_compatible
class Message(models.Model):
"""
This model represents a message on the database. Fields are the same as in
`contrib.messages`
"""
message = models.TextField()
level = models.IntegerField()
tags = models.TextField()
url = models.URLField(null=True, blank=True)
date = models.DateTimeField(default=timezone.now)
def __str__(self):
return self.message
@python_2_unicode_compatible
class MessageArchive(models.Model):
"""
This model holds all the messages users received. Corresponding
database table will grow indefinitely depending on messages traffic.
"""
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
message = models.ForeignKey(Message, on_delete=models.CASCADE)
def __str__(self):
return "[%s] %s" % (self.user, self.message)
@python_2_unicode_compatible
class Inbox(models.Model):
"""
Inbox messages are stored in this model until users read them. Once read,
inbox messages are deleted. Inbox messages have an expire time, after
that they could be removed by a proper django command. We do not expect
database table corresponding to this model to grow much.
"""
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
message = models.ForeignKey(Message, models.CASCADE)
class Meta:
verbose_name_plural = _('inboxes')
def expired(self):
expiration_date = self.message.date + timezone.timedelta(
days=stored_messages_settings.INBOX_EXPIRE_DAYS)
return expiration_date <= timezone.now()
expired.boolean = True # show a nifty icon in the admin
def __str__(self):
return "[%s] %s" % (self.user, self.message)

10
vendor/stored_messages/serializers.py vendored Normal file
View file

@ -0,0 +1,10 @@
from rest_framework import serializers
class InboxSerializer(serializers.Serializer):
id = serializers.CharField()
message = serializers.CharField()
level = serializers.IntegerField()
tags = serializers.CharField()
date = serializers.DateTimeField(format=None)
url = serializers.URLField(required=False)

130
vendor/stored_messages/settings.py vendored Normal file
View file

@ -0,0 +1,130 @@
"""
This module is largely inspired by django-rest-framework settings.
Settings for the Stored Messages framework are all namespaced in the
STORED_MESSAGES setting.
For example your project's `settings.py` file might look like this::
STORED_MESSAGES = {
'INBOX_EXPIRE_DAYS': 30,
}
This module provides the `stored_messages_settings` object, that is used to
access Stored Messages framework settings, checking for user settings first,
then falling back to the defaults.
"""
from __future__ import unicode_literals
import importlib
from django.conf import settings
import six
from .constants import *
USER_SETTINGS = getattr(settings, 'STORED_MESSAGES', None)
MESSAGE_TAGS = getattr(settings, 'MESSAGE_TAGS', {})
DEFAULTS = {
'STORE_LEVELS': (
STORED_DEBUG,
STORED_INFO,
STORED_SUCCESS,
STORED_WARNING,
STORED_ERROR,
),
'MESSAGE_TAGS': MESSAGE_TAGS.update({
STORED_DEBUG: 'stored debug',
STORED_INFO: 'stored info',
STORED_SUCCESS: 'stored success',
STORED_WARNING: 'stored warning',
STORED_ERROR: 'stored error',
}),
'INBOX_EXPIRE_DAYS': 30,
'MESSAGE_EXPIRE_DAYS': 120,
'STORAGE_BACKEND': 'stored_messages.backends.DefaultBackend',
# Only for Redis backend
'REDIS_URL': 'redis://username:password@localhost:6379/0',
}
# List of settings that cannot be empty
MANDATORY = (
)
# List of settings that may be in string import notation.
IMPORT_STRINGS = (
'STORAGE_BACKEND',
)
def perform_import(val, setting_name):
"""
If the given setting is a string import notation,
then perform the necessary import or imports.
"""
if isinstance(val, six.string_types):
return import_from_string(val, setting_name)
elif isinstance(val, (list, tuple)):
return [import_from_string(item, setting_name) for item in val]
return val
def import_from_string(val, setting_name):
"""
Attempt to import a class from a string representation.
"""
try:
parts = val.split('.')
module_path, class_name = '.'.join(parts[:-1]), parts[-1]
module = importlib.import_module(module_path)
return getattr(module, class_name)
except ImportError as e:
msg = "Could not import '%s' for setting '%s'. %s: %s." % (val, setting_name,
e.__class__.__name__, e)
raise ImportError(msg)
class StoredMessagesSettings(object):
"""
A settings object, that allows settings to be accessed as properties.
Any setting with string import paths will be automatically resolved
and return the class, rather than the string literal.
"""
def __init__(self, user_settings=None, defaults=None, import_strings=None, mandatory=None):
self.user_settings = user_settings or {}
self.defaults = defaults or {}
self.import_strings = import_strings or ()
self.mandatory = mandatory or ()
def __getattr__(self, attr):
if attr not in self.defaults.keys():
raise AttributeError("Invalid Stored Messages setting: '%s'" % attr)
try:
# Check if present in user settings
val = self.user_settings[attr]
except KeyError:
# Fall back to defaults
val = self.defaults[attr]
# Coerce import strings into classes
if val and attr in self.import_strings:
val = perform_import(val, attr)
self.validate_setting(attr, val)
# Cache the result
setattr(self, attr, val)
return val
def validate_setting(self, attr, val):
if not val and attr in self.mandatory:
raise AttributeError("Stored Messages setting: '%s' is mandatory" % attr)
stored_messages_settings = StoredMessagesSettings(USER_SETTINGS, DEFAULTS, IMPORT_STRINGS,
MANDATORY)

0
vendor/stored_messages/static/js/stored_messages.js vendored Normal file
View file

103
vendor/stored_messages/storage.py vendored Normal file
View file

@ -0,0 +1,103 @@
from django.contrib.messages.storage.fallback import FallbackStorage
from .settings import stored_messages_settings
from .backends.exceptions import MessageTypeNotSupported
class StorageMixin(object):
"""
This mixin implements a message storage compliant with
`django.contrib.messages` which stores messages on the database when
needed, delegating to the sibling class otherwise. Messages are stored
only when the user is authenticated and the message level is configured
for being persisted. The mixin can be derived together with another class
implementing a storage, tipically one of the three provided by Django out
of the box.
"""
def __init__(self, request, *args, **kwargs):
self.user = request.user
BackendClass = stored_messages_settings.STORAGE_BACKEND
self.backend = BackendClass()
super(StorageMixin, self).__init__(request, *args, **kwargs)
def _get(self, *args, **kwargs):
"""
Retrieve unread messages for current user, both from the inbox and
from other storages
"""
messages, all_retrieved = super(StorageMixin, self)._get(*args, **kwargs)
if self.user.is_authenticated():
inbox_messages = self.backend.inbox_list(self.user)
else:
inbox_messages = []
return messages + inbox_messages, all_retrieved
def add(self, level, message, extra_tags=''):
"""
If the message level was configured for being stored and request.user
is not anonymous, save it to the database. Otherwise, let some other
class handle the message.
Notice: controls like checking the message is not empty and the level
is above the filter need to be performed here, but it could happen
they'll be performed again later if the message does not need to be
stored.
"""
if not message:
return
# Check that the message level is not less than the recording level.
level = int(level)
if level < self.level:
return
# Check if the message doesn't have a level that needs to be persisted
if level not in stored_messages_settings.STORE_LEVELS or self.user.is_anonymous():
return super(StorageMixin, self).add(level, message, extra_tags)
self.added_new = True
m = self.backend.create_message(level, message, extra_tags)
self.backend.archive_store([self.user], m)
self._queued_messages.append(m)
def _store(self, messages, response, *args, **kwargs):
"""
persistent messages are already in the database inside the 'archive',
so we can say they're already "stored".
Here we put them in the inbox, or remove from the inbox in case the
messages were iterated.
messages contains only new msgs if self.used==True
else contains both new and unread messages
"""
contrib_messages = []
if self.user.is_authenticated():
if not messages:
# erase inbox
self.backend.inbox_purge(self.user)
else:
for m in messages:
try:
self.backend.inbox_store([self.user], m)
except MessageTypeNotSupported:
contrib_messages.append(m)
super(StorageMixin, self)._store(contrib_messages, response, *args, **kwargs)
def _prepare_messages(self, messages):
"""
Like the base class method, prepares a list of messages for storage
but avoid to do this for `models.Message` instances.
"""
for message in messages:
if not self.backend.can_handle(message):
message._prepare()
class PersistentStorage(StorageMixin, FallbackStorage):
"""
This class is provided for convenience: it implements `StorageMixin` which
persists messages having a type configured to be persisted and passes
other messages to `FallbackStorage`, one of the defaults provided by
Django which stores messages inside cookies or sessions.
"""
pass

21
vendor/stored_messages/templates/stored_messages/base.html vendored Normal file
View file

@ -0,0 +1,21 @@
{% comment %}
As the developer of this package, don't place anything here if you can help it
since this allows developers to have interoperability between your template
structure and their own.
Example: Developer melding the 2SoD pattern to fit inside with another pattern::
{% extends "base.html" %}
{% load static %}
<!-- Their site uses old school block layout -->
{% block extra_js %}
<!-- Your package using 2SoD block layout -->
{% block javascript %}
<script src="{% static 'js/ninja.js' %}" type="text/javascript"></script>
{% endblock javascript %}
{% endblock extra_js %}
{% endcomment %}

7
vendor/stored_messages/templates/stored_messages/stored_messages_list.html vendored Normal file
View file

@ -0,0 +1,7 @@
{% if messages %}
<ul>
{% for message in messages %}
<li>{{ message }}</li>
{% endfor %}
</ul>
{% endif %}

0
vendor/stored_messages/templatetags/__init__.py vendored Normal file
View file

49
vendor/stored_messages/templatetags/stored_messages_tags.py vendored Normal file
View file

@ -0,0 +1,49 @@
from __future__ import unicode_literals
from django import template
from ..models import Inbox, MessageArchive
register = template.Library()
@register.inclusion_tag("stored_messages/stored_messages_list.html", takes_context=True)
def stored_messages_list(context, num_elements=10):
"""
Renders a list of unread stored messages for the current user
"""
if "user" in context:
user = context["user"]
if user.is_authenticated():
qs = Inbox.objects.select_related("message").filter(user=user)
return {
"messages": qs[:num_elements],
"count": qs.count(),
}
@register.simple_tag(takes_context=True)
def stored_messages_count(context):
"""
Renders a list of unread stored messages for the current user
"""
if "user" in context:
user = context["user"]
if user.is_authenticated():
return Inbox.objects.select_related("message").filter(user=user).count()
@register.inclusion_tag("stored_messages/stored_messages_list.html", takes_context=True)
def stored_messages_archive(context, num_elements=10):
"""
Renders a list of archived messages for the current user
"""
if "user" in context:
user = context["user"]
if user.is_authenticated():
qs = MessageArchive.objects.select_related("message").filter(user=user)
return {
"messages": qs[:num_elements],
"count": qs.count(),
}

0
vendor/stored_messages/tests/__init__.py vendored Normal file
View file

40
vendor/stored_messages/tests/base.py vendored Normal file
View file

@ -0,0 +1,40 @@
from __future__ import unicode_literals
from django.contrib.auth import get_user_model
from django.test import TestCase, RequestFactory
from django.utils.six.moves import reload_module
import mock
from stored_messages import storage
from stored_messages import settings
class BaseTest(TestCase):
def setUp(self):
# settings and storage modules should be reloaded
reload_module(settings)
reload_module(storage)
self.factory = RequestFactory()
self.user = get_user_model().objects.create_user("test_user", "t@user.com", "123456")
self.request = RequestFactory().get('/')
self.request.session = mock.MagicMock()
self.request.user = self.user
def tearDown(self):
self.user.delete()
class BackendBaseTest(BaseTest):
"""
Tests that need to access a Backend.
Given the dynamic nature of Stored Messages settings, retrieving the backend class when we
need to ovveride settings is a little bit tricky
"""
def setUp(self):
super(BackendBaseTest, self).setUp()
self.backend = settings.stored_messages_settings.STORAGE_BACKEND()
def tearDown(self):
self.backend._flush()

69
vendor/stored_messages/tests/settings.py vendored Normal file
View file

@ -0,0 +1,69 @@
# flake8: noqa
import django
INSTALLED_APPS = [
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sites",
"django.contrib.messages",
"django.contrib.sessions",
"stored_messages",
]
try:
import rest_framework
INSTALLED_APPS.append("rest_framework")
except ImportError:
pass
DEBUG = True
USE_TZ = True
DATABASES = {
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": "stored_messages.sqlite",
}
}
ROOT_URLCONF = "stored_messages.tests.urls"
# Django 1.8 has a new, minimal default set for MIDDLEWARE_CLASSES so be explicit
MIDDLEWARE_CLASSES = (
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.auth.middleware.SessionAuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
'django.middleware.security.SecurityMiddleware',
)
# Django 1.10 requires the TEMPLATES settings. Deprecated since Django 1.8
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
SITE_ID = 1
MESSAGE_STORAGE = 'stored_messages.storage.PersistentStorage'
STORED_MESSAGES = {
'REDIS_URL': 'redis://localhost:6379/0',
}
MOCK_REDIS_SERVER = False
SECRET_KEY = 'FUUUFU'

93
vendor/stored_messages/tests/test_api.py vendored Normal file
View file

@ -0,0 +1,93 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from .base import BaseTest
from django.contrib.auth import get_user_model
from django.utils import timezone
from stored_messages import mark_read, add_message_for, broadcast_message, mark_all_read
from stored_messages.models import Inbox, MessageArchive
from stored_messages.backends.exceptions import MessageDoesNotExist
import stored_messages
class TestApi(BaseTest):
def test_mark_as_read(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
inbox = Inbox.objects.filter(user=self.user)
self.assertEqual(len(inbox), 1)
mark_read(self.user, inbox.get().message)
self.assertEqual(Inbox.objects.filter(user=self.user).count(), 0)
def test_mark_as_read_idempotent(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
msg_archive = MessageArchive.objects.filter(user=self.user).get()
mark_read(self.user, msg_archive.message)
self.assertRaises(MessageDoesNotExist, mark_read, self.user, msg_archive.message)
def test_add_message_for(self):
now = timezone.now() + timezone.timedelta(days=-1)
url = 'http://example.com/error'
user2 = get_user_model().objects.create_user("another_user", "u@user.com", "123456")
add_message_for([user2, self.user], stored_messages.STORED_ERROR, 'Multiple errors', 'extra', now, url)
self.assertEqual(Inbox.objects.count(), 2)
self.assertEqual(MessageArchive.objects.count(), 2)
self.assertEqual(Inbox.objects.get(user=user2.id).message.tags, 'extra')
self.assertEqual(Inbox.objects.get(user=self.user).message.tags, 'extra')
self.assertEqual(Inbox.objects.get(user=user2.id).message.date, now)
self.assertEqual(Inbox.objects.get(user=self.user).message.date, now)
self.assertEqual(Inbox.objects.get(user=user2.id).message.url, url)
self.assertEqual(Inbox.objects.get(user=self.user).message.url, url)
self.assertEqual(Inbox.objects.get(user=user2.id).message.message, "Multiple errors")
self.assertEqual(Inbox.objects.get(user=self.user).message.message, "Multiple errors")
self.assertEqual(MessageArchive.objects.get(user=user2.id).message.message,
"Multiple errors")
self.assertEqual(MessageArchive.objects.get(user=self.user).message.message,
"Multiple errors")
def test_broadcast_message(self):
user1 = get_user_model().objects.create_user("user1", "u1@user.com", "123456")
user2 = get_user_model().objects.create_user("user2", "u2@user.com", "123456")
user3 = get_user_model().objects.create_user("user3", "u3@user.com", "123456")
now = timezone.now() + timezone.timedelta(days=-1)
url = 'http://example.com/error'
broadcast_message(stored_messages.STORED_INFO, 'broadcast test message', 'extra', now, url)
self.assertEqual(Inbox.objects.get(user=user1.id).message.message, "broadcast test message")
self.assertEqual(Inbox.objects.get(user=user2.id).message.message, "broadcast test message")
self.assertEqual(Inbox.objects.get(user=user3.id).message.message, "broadcast test message")
self.assertEqual(Inbox.objects.get(user=user1.id).message.tags, 'extra')
self.assertEqual(Inbox.objects.get(user=user2.id).message.tags, 'extra')
self.assertEqual(Inbox.objects.get(user=user3.id).message.tags, 'extra')
self.assertEqual(Inbox.objects.get(user=user1.id).message.date, now)
self.assertEqual(Inbox.objects.get(user=user2.id).message.date, now)
self.assertEqual(Inbox.objects.get(user=user3.id).message.date, now)
self.assertEqual(Inbox.objects.get(user=user1.id).message.url, url)
self.assertEqual(Inbox.objects.get(user=user2.id).message.url, url)
self.assertEqual(Inbox.objects.get(user=user3.id).message.url, url)
self.assertEqual(MessageArchive.objects.get(user=user1.id).message.message,
"broadcast test message")
self.assertEqual(MessageArchive.objects.get(user=user2.id).message.message,
"broadcast test message")
self.assertEqual(MessageArchive.objects.get(user=user3.id).message.message,
"broadcast test message")
def test_mark_all_read(self):
for i in range(20):
stored_messages.add_message_for([self.user], stored_messages.INFO, "unicode message ❤")
inbox = Inbox.objects.filter(user=self.user)
self.assertEqual(len(inbox), 20)
mark_all_read(self.user)
self.assertEqual(Inbox.objects.filter(user=self.user).count(), 0)

134
vendor/stored_messages/tests/test_default_backend.py vendored Normal file
View file

@ -0,0 +1,134 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from .base import BaseTest
from django.contrib.auth.models import AnonymousUser
from django.utils import timezone
from stored_messages.backends.exceptions import MessageTypeNotSupported, MessageDoesNotExist
from stored_messages.backends.default import DefaultBackend
from stored_messages.models import Message, Inbox, MessageArchive
from stored_messages import STORED_ERROR
from stored_messages.backends import signals
class TestDefaultBackend(BaseTest):
def setUp(self):
super(TestDefaultBackend, self).setUp()
self.backend = DefaultBackend()
self.message = self.backend.create_message(STORED_ERROR, 'A message for you ♡')
self.anon = AnonymousUser()
self.signals = {}
def test_inbox_store(self):
self.backend.inbox_store([self.user], self.message)
m = Inbox.objects.filter(user=self.user).get()
self.assertEqual(m.message, self.message)
self.assertRaises(MessageTypeNotSupported, self.backend.inbox_store, [], {})
def test_inbox_list(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you ♡♡♡')
self.backend.inbox_store([self.user], message)
self.backend.inbox_store([self.user], self.message)
messages = self.backend.inbox_list(self.user)
self.assertEqual(messages[0], message)
self.assertEqual(messages[1], self.message)
self.assertEqual(len(self.backend.inbox_list(self.anon)), 0)
def test_inbox_purge(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you')
self.backend.inbox_store([self.user], self.message)
self.backend.inbox_store([self.user], message)
self.backend.inbox_purge(self.user)
self.assertEqual(len(self.backend.inbox_list(self.user)), 0)
self.backend.inbox_purge(self.anon)
def test_inbox_delete(self):
self.backend.inbox_store([self.user], self.message)
self.backend.inbox_delete(self.user, self.message.id)
self.assertEqual(len(self.backend.inbox_list(self.user)), 0)
self.assertRaises(MessageDoesNotExist, self.backend.inbox_delete, self.user, -1)
def test_archive_store(self):
self.backend.archive_store([self.user], self.message)
m = MessageArchive.objects.filter(user=self.user).all()[0]
self.assertEqual(m.message, self.message)
self.assertRaises(MessageTypeNotSupported, self.backend.archive_store, [], {})
def test_archive_list(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you')
self.backend.archive_store([self.user], message)
self.backend.archive_store([self.user], self.message)
messages = self.backend.archive_list(self.user)
self.assertEqual(messages[0].message, message)
self.assertEqual(messages[1].message, self.message)
def test_create_message(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you')
self.assertIsInstance(message, Message)
def test_inbox_get(self):
self.backend.inbox_store([self.user], self.message)
m = self.backend.inbox_get(self.user, self.message.id)
self.assertEqual(m, self.message)
self.assertRaises(MessageDoesNotExist, self.backend.inbox_get, self.user, -1)
def test_can_handle(self):
self.assertFalse(self.backend.can_handle({}))
self.assertTrue(self.backend.can_handle(self.message))
def test_message_expiration(self):
six_months_ago = timezone.now() + timezone.timedelta(days=-180)
self.message.date = six_months_ago
self.message.save()
self.backend.expired_messages_cleanup()
n_archives = MessageArchive.objects.count()
n_inbox = Inbox.objects.count()
n_messages = Message.objects.count()
self.assertEqual(n_archives, 0)
self.assertEqual(n_inbox, 0)
self.assertEqual(n_messages, 0)
def test_inbox_signals(self):
# connect
signals.inbox_stored.connect(self.inbox_stored)
signals.inbox_deleted.connect(self.inbox_deleted)
signals.inbox_purged.connect(self.inbox_purged)
self.backend.inbox_store([self.user], self.message)
self.backend.inbox_delete(self.user, self.message.id)
self.backend.inbox_purge(self.user)
# disconnect
signals.inbox_stored.disconnect(self.inbox_stored)
signals.inbox_deleted.disconnect(self.inbox_deleted)
signals.inbox_purged.disconnect(self.inbox_purged)
self.assertIn('inbox_stored', self.signals)
self.assertIn('inbox_deleted', self.signals)
self.assertIn('inbox_purged', self.signals)
def inbox_stored(self, **kwargs):
self.signals['inbox_stored'] = (kwargs['user'], kwargs['message'])
def inbox_deleted(self, **kwargs):
self.signals['inbox_deleted'] = (kwargs['user'], kwargs['message_id'])
def inbox_purged(self, **kwargs):
self.signals['inbox_purged'] = (kwargs['user'])
def test_archive_signals(self):
# connect
signals.archive_stored.connect(self.archive_stored)
self.backend.archive_store([self.user], self.message)
# disconnect
signals.archive_stored.disconnect(self.archive_stored)
self.assertIn('archive_stored', self.signals)
def archive_stored(self, **kwargs):
self.signals['archive_stored'] = (kwargs['user'], kwargs['message'])

22
vendor/stored_messages/tests/test_models.py vendored Normal file
View file

@ -0,0 +1,22 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from .base import BaseTest
from django.utils.timezone import timedelta, now
from django.contrib.messages.storage import default_storage
from stored_messages import add_message, STORED_ERROR
from stored_messages.models import MessageArchive, Inbox
class TestInbox(BaseTest):
def test_expire(self):
self.request._messages = default_storage(self.request)
add_message(self.request, STORED_ERROR, "an SOS to the world ☢")
mas = MessageArchive.objects.all()
self.assertEqual(len(mas), 1)
ma = mas[0]
ma.message.date = now() - timedelta(days=365)
ma.message.save()
inbox = Inbox.objects.create(user=ma.user, message=ma.message)
self.assertTrue(inbox.expired())

201
vendor/stored_messages/tests/test_redis_backend.py vendored Normal file
View file

@ -0,0 +1,201 @@
from .base import BaseTest
import unittest
import mock
from django.conf import settings
from django.core.cache import cache
from django.contrib.auth.models import AnonymousUser
from django.utils import timezone
from stored_messages.backends.exceptions import MessageTypeNotSupported, MessageDoesNotExist
from stored_messages.backends.redis import RedisBackend
from stored_messages.backends.redis.backend import Message
from stored_messages.backends import signals
from stored_messages import STORED_ERROR
class RedisMock(object):
"""
Mock the Redis server instance with Django in-memory cache
"""
def rpush(self, key, data):
l = cache.get(key)
if l is None:
l = []
l.append(data)
cache.set(key, l)
def lrange(self, key, *args, **kwargs):
return cache.get(key) or []
def delete(self, key):
cache.delete(key)
def lrem(self, key, count, data):
l = cache.get(key)
l.remove(data)
cache.set(key, l)
def flushdb(self):
cache.clear()
def sismember(self, key, val):
r = cache.get(key)
return r is not None and val in r
def sadd(self, key, val):
l = cache.get(key)
if l is None:
l = []
if val not in l:
l.append(val)
cache.set(key, l)
@staticmethod
def from_url(*args, **kwargs):
return RedisMock()
def StrictRedis(self, *args, **kwargs):
return self
try:
from stored_messages.backends.redis.backend import redis
REDISPY_MISSING = False
if getattr(settings, 'MOCK_REDIS_SERVER', True):
patcher = mock.patch('stored_messages.backends.redis.backend.redis')
redis = patcher.start() # noqa
redis.StrictRedis = RedisMock
except ImportError:
REDISPY_MISSING = True
@unittest.skipIf(REDISPY_MISSING, "redis-py not installed")
class TestRedisBackend(BaseTest):
def setUp(self):
super(TestRedisBackend, self).setUp()
self.client = redis.StrictRedis.from_url(settings.STORED_MESSAGES['REDIS_URL'])
self.backend = RedisBackend()
self.backend._flush()
self.message = self.backend.create_message(STORED_ERROR, 'A message for you')
self.anon = AnonymousUser()
self.signals = {}
def test_inbox_store(self):
self.backend.inbox_store([self.user], self.message)
data = self.client.lrange('user:%d:notifications' % self.user.pk, 0, -1).pop()
self.assertEqual(self.backend._fromJSON(data), self.message)
self.assertRaises(MessageTypeNotSupported, self.backend.inbox_store, [], {})
def test_inbox_list(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you')
self.backend.inbox_store([self.user], message)
self.backend.inbox_store([self.user], self.message)
messages = self.backend.inbox_list(self.user)
self.assertEqual(messages[0], message)
self.assertEqual(messages[1], self.message)
self.assertEqual(self.backend.inbox_list(self.anon), [])
def test_inbox_purge(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you')
self.backend.inbox_store([self.user], self.message)
self.backend.inbox_store([self.user], message)
self.backend.inbox_purge(self.user)
self.assertEqual(len(self.backend.inbox_list(self.user)), 0)
self.backend.inbox_purge(self.anon)
def test_inbox_delete(self):
self.backend.inbox_store([self.user], self.message)
self.backend.inbox_delete(self.user, self.message.id)
self.assertEqual(len(self.backend.inbox_list(self.user)), 0)
self.assertRaises(MessageDoesNotExist, self.backend.inbox_delete, self.user, -1)
def test_archive_store(self):
self.backend.archive_store([self.user], self.message)
data = self.client.lrange('user:%d:archive' % self.user.pk, 0, -1).pop()
self.assertEqual(self.backend._fromJSON(data), self.message)
self.assertRaises(MessageTypeNotSupported, self.backend.archive_store, [], {})
def test_archive_list(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you')
self.backend.archive_store([self.user], message)
self.backend.archive_store([self.user], self.message)
messages = self.backend.archive_list(self.user)
self.assertEqual(messages[0], message)
self.assertEqual(messages[1], self.message)
def test_create_message(self):
message = self.backend.create_message(STORED_ERROR, 'Another message for you')
self.assertIsInstance(message, Message)
def test_inbox_get(self):
self.backend.inbox_store([self.user], self.message)
m = self.backend.inbox_get(self.user, self.message.id)
self.assertEqual(m, self.message)
self.assertRaises(MessageDoesNotExist, self.backend.inbox_get, self.user, -1)
def test_can_handle(self):
self.assertFalse(self.backend.can_handle({}))
self.assertTrue(self.backend.can_handle(self.message))
def test_message_expiration(self):
# start clean
self.backend._flush()
six_months_ago = timezone.now() + timezone.timedelta(days=-180)
self.backend.create_message(STORED_ERROR, 'A message for you', date=six_months_ago)
self.backend.expired_messages_cleanup()
keys = self.client.keys('user:*:notifications')
for k in keys:
_, user, _ = k.split(':')
self.assertEqual(self.backend._list_key(k), [])
self.assertEqual(self.backend._list('user:%s:notifications', user), [])
self.assertEqual(self.backend._list('user:%s:archive', user), [])
def test_inbox_signals(self):
# connect
signals.inbox_stored.connect(self.inbox_stored)
signals.inbox_deleted.connect(self.inbox_deleted)
signals.inbox_purged.connect(self.inbox_purged)
self.backend.inbox_store([self.user], self.message)
self.backend.inbox_delete(self.user, self.message.id)
self.backend.inbox_purge(self.user)
# disconnect
signals.inbox_stored.disconnect(self.inbox_stored)
signals.inbox_deleted.disconnect(self.inbox_deleted)
signals.inbox_purged.disconnect(self.inbox_purged)
self.assertIn('inbox_stored', self.signals)
self.assertIn('inbox_deleted', self.signals)
self.assertIn('inbox_purged', self.signals)
def inbox_stored(self, **kwargs):
self.signals['inbox_stored'] = (kwargs['user'], kwargs['message'])
def inbox_deleted(self, **kwargs):
self.signals['inbox_deleted'] = (kwargs['user'], kwargs['message_id'])
def inbox_purged(self, **kwargs):
self.signals['inbox_purged'] = (kwargs['user'])
def test_archive_signals(self):
# connect
signals.archive_stored.connect(self.archive_stored)
self.backend.archive_store([self.user], self.message)
# disconnect
signals.archive_stored.disconnect(self.archive_stored)
self.assertIn('archive_stored', self.signals)
def archive_stored(self, **kwargs):
self.signals['archive_stored'] = (kwargs['user'], kwargs['message'])

80
vendor/stored_messages/tests/test_restapi.py vendored Normal file
View file

@ -0,0 +1,80 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import json
from unittest import skipUnless
from django.core.urlresolvers import reverse
from django.test.utils import override_settings
from stored_messages.constants import STORED_ERROR
from .base import BackendBaseTest
try:
import rest_framework # noqa
rest_framework_installed = True
except ImportError:
rest_framework_installed = False
@skipUnless(rest_framework_installed, "Django restframework is not installed")
class TestRESTApi(BackendBaseTest):
def test_list(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
self.client.get('/create')
r = self.client.get(reverse('stored_messages:inbox-list'))
messages = json.loads(r.content.decode('utf-8'))
self.assertEqual(len(messages), 2)
self.assertEqual(messages[0]['message'], 'an error ☢')
self.assertEqual(messages[1]['message'], 'an error ☢')
self.assertEqual(messages[0]['level'], STORED_ERROR)
self.assertEqual(messages[1]['level'], STORED_ERROR)
def test_retrieve(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
r = self.client.get(reverse('stored_messages:inbox-list'))
messages = json.loads(r.content.decode('utf-8'))
id = messages[0]['id']
r = self.client.get(reverse('stored_messages:inbox-detail', kwargs={'pk': id}))
message = json.loads(r.content.decode('utf-8'))
self.assertEqual(message['message'], 'an error ☢')
self.assertEqual(message['id'], id)
def test_make_read(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
r = self.client.get(reverse('stored_messages:inbox-list'))
messages = json.loads(r.content.decode('utf-8'))
self.assertEqual(len(messages), 1)
msg_id = messages[0]['id']
r = self.client.post(reverse('stored_messages:inbox-read', args=(msg_id,)))
self.assertEqual(r.status_code, 200)
r = self.client.get(reverse('stored_messages:inbox-list'))
messages = json.loads(r.content.decode('utf-8'))
self.assertEqual(len(messages), 0)
def test_anon(self):
self.client.get('/create')
r = self.client.get(reverse('stored_messages:inbox-list'))
messages = json.loads(r.content.decode('utf-8'))
self.assertEqual(len(messages), 0)
def test_mark_all_read(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
r = self.client.post(reverse('stored_messages:mark_all_read'))
self.assertEqual(r.status_code, 200)
r = self.client.get(reverse('stored_messages:inbox-list'))
messages = json.loads(r.content.decode('utf-8'))
self.assertEqual(len(messages), 0)
@skipUnless(rest_framework_installed, "Django restframework is not installed")
@override_settings(STORED_MESSAGES={'STORAGE_BACKEND': 'stored_messages.backends.RedisBackend'})
class TestRESTApiWithRedis(TestRESTApi):
pass

59
vendor/stored_messages/tests/test_storage.py vendored Normal file
View file

@ -0,0 +1,59 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from .base import BackendBaseTest
from django.contrib.messages.storage import default_storage
from stored_messages import add_message, get_messages, STORED_ERROR, DEBUG, ERROR
import mock
class TestStorage(BackendBaseTest):
"""
Test Storage class using default backend
"""
def test_store(self):
self.request._messages = default_storage(self.request)
self.request._messages.level = DEBUG
add_message(self.request, STORED_ERROR, "an SOS to the world ☢")
add_message(self.request, DEBUG, "this won't be persisted ☢")
self.assertEqual(len(get_messages(self.request)), 2)
self.assertEqual(len(self.backend.archive_list(self.user)), 1)
def test_store_with_middleware(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
inbox_msg = len(self.backend.inbox_list(self.user))
self.assertEqual(inbox_msg, 1)
self.client.get('/consume')
self.assertEqual(len(self.backend.inbox_list(self.user)), 0)
self.assertEqual(len(self.backend.archive_list(self.user)), 1)
def test_store_keep_unread(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create')
self.client.get('/consume', data={'keep_storage': True})
self.assertEqual(len(self.backend.inbox_list(self.user)), 1)
self.assertEqual(len(self.backend.archive_list(self.user)), 1)
def test_store_anonymous(self):
self.request.user = mock.MagicMock(wraps=self.user)
self.request.user.is_anonymous.return_value = True
self.request.user.is_authenticated.return_value = False
self.request._messages = default_storage(self.request)
add_message(self.request, STORED_ERROR, "an SOS to the world ☢")
add_message(self.request, ERROR, "this error won't be persisted ☢")
storage = get_messages(self.request)
self.assertEqual(len(storage), 2)
def test_add_empty(self):
self.request._messages = default_storage(self.request)
add_message(self.request, DEBUG, '')
self.assertEqual(len(get_messages(self.request)), 0)
def test_add_mixed(self):
self.client.login(username='test_user', password='123456')
self.client.get('/create_mixed')
self.client.get('/consume')
self.assertEqual(len(self.backend.inbox_list(self.user)), 0)

68
vendor/stored_messages/tests/test_templatetags.py vendored Normal file
View file

@ -0,0 +1,68 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.template import RequestContext, Template
from .base import BaseTest
import stored_messages
class TestStoredMessagesTags(BaseTest):
def _create_messages(self):
# create a message
for i in range(20):
stored_messages.add_message_for([self.user], stored_messages.INFO, "unicode message ❤")
def test_stored_messages_list(self):
self._create_messages()
t = Template("{% load stored_messages_tags %}"
"{% stored_messages_list 15 %}")
render = t.render(RequestContext(self.request))
self.assertInHTML("<li>[test_user] unicode message ❤</li>", render, 15)
def test_stored_messages_archive(self):
self._create_messages()
t = Template("{% load stored_messages_tags %}"
"{% stored_messages_archive 15 %}")
render = t.render(RequestContext(self.request))
self.assertInHTML("<li>[test_user] unicode message ❤</li>", render, 15)
def test_stored_messages_count(self):
self._create_messages()
t = Template("{% load stored_messages_tags %}"
"{% stored_messages_count as count %}"
"<p>There are {{ count }} messages</p>")
render = t.render(RequestContext(self.request))
self.assertInHTML("<p>There are 20 messages</p>", render, 1)
def test_stored_messages_list_empty_for_unauthenticated_user(self):
stored_messages.add_message_for([self.user], stored_messages.INFO, "unicode message ❤")
t = Template("{% load stored_messages_tags %}"
"{% stored_messages_list 15 %}")
render = t.render(RequestContext(self.factory.get("/")))
self.assertInHTML("<li>[test_user] unicode message ❤</li>", render, 0)
def test_stored_messages_archive_empty_for_unauthenticated_user(self):
stored_messages.add_message_for([self.user], stored_messages.INFO, "unicode message ❤")
t = Template("{% load stored_messages_tags %}"
"{% stored_messages_archive 15 %}")
render = t.render(RequestContext(self.factory.get("/")))
self.assertInHTML("<li>[test_user] unicode message ❤</li>", render, 0)
def assertInHTML(self, needle, haystack, count=None, msg_prefix=''):
import django
if django.VERSION < (1, 5):
real_count = haystack.count(needle)
return self.assertEqual(real_count, count)
else:
return super(TestStoredMessagesTags, self).assertInHTML(
needle, haystack, count, msg_prefix)

14
vendor/stored_messages/tests/urls.py vendored Normal file
View file

@ -0,0 +1,14 @@
from django.contrib import admin
from django.urls import re_path as url, include
from stored_messages.tests.views import message_view, message_create, message_create_mixed
admin.autodiscover()
urlpatterns = [
url(r'^consume$', message_view),
url(r'^create$', message_create),
url(r'^create_mixed$', message_create_mixed),
url(r'^messages', include('stored_messages.urls', namespace='stored_messages'))
]

27
vendor/stored_messages/tests/views.py vendored Normal file
View file

@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.http import HttpResponse
import stored_messages
def message_view(request):
keep_storage = request.GET.get('keep_storage', False)
storage = stored_messages.get_messages(request)
for _ in storage:
pass
if keep_storage:
storage.used = False
return HttpResponse()
def message_create(request):
stored_messages.add_message(request, stored_messages.STORED_ERROR, "an error ☢")
stored_messages.add_message(request, stored_messages.STORED_DEBUG, "a debug message ☢")
return HttpResponse()
def message_create_mixed(request):
stored_messages.add_message(request, stored_messages.STORED_ERROR, "an error ☢")
stored_messages.add_message(request, stored_messages.ERROR, "an error not persisted ☢")
return HttpResponse()

17
vendor/stored_messages/urls.py vendored Normal file
View file

@ -0,0 +1,17 @@
"""
At the moment this module does something only when restframework is available
"""
from django.conf import settings
if 'rest_framework' in settings.INSTALLED_APPS:
from rest_framework.routers import DefaultRouter
from django.urls import re_path as url, include
from . import views
router = DefaultRouter()
router.register(r'inbox', views.InboxViewSet, base_name='inbox')
urlpatterns = [
url(r'^', include(router.urls)),
url(r'^mark_all_read/$', views.mark_all_read, name='mark_all_read'),
]

60
vendor/stored_messages/views.py vendored Normal file
View file

@ -0,0 +1,60 @@
from rest_framework import viewsets
from rest_framework.decorators import api_view, detail_route
from rest_framework.response import Response
from django.contrib.auth.decorators import login_required
from .serializers import InboxSerializer
from .backends.exceptions import MessageDoesNotExist
class InboxViewSet(viewsets.ViewSet):
"""
Provides `list` and `detail` actions, plus a `read` POST endpoint for
marking inbox messages as read.
"""
def list(self, request):
from .settings import stored_messages_settings
backend = stored_messages_settings.STORAGE_BACKEND()
messages = backend.inbox_list(request.user)
serializer = InboxSerializer(messages, many=True)
return Response(serializer.data)
def retrieve(self, request, pk=None):
from .settings import stored_messages_settings
backend = stored_messages_settings.STORAGE_BACKEND()
try:
msg = backend.inbox_get(request.user, pk)
except MessageDoesNotExist as e:
return Response(e.message, status='404')
serializer = InboxSerializer(msg, many=False)
return Response(serializer.data)
@detail_route(methods=['POST'])
def read(self, request, pk=None):
"""
Mark the message as read (i.e. delete from inbox)
"""
from .settings import stored_messages_settings
backend = stored_messages_settings.STORAGE_BACKEND()
try:
backend.inbox_delete(request.user, pk)
except MessageDoesNotExist as e:
return Response(e.message, status='404')
return Response({'status': 'message marked as read'})
@login_required
@api_view(['POST'])
def mark_all_read(request):
"""
Mark all messages as read (i.e. delete from inbox) for current logged in user
"""
from .settings import stored_messages_settings
backend = stored_messages_settings.STORAGE_BACKEND()
backend.inbox_purge(request.user)
return Response({"message": "All messages read"})

20
vendor/treebeard/__init__.py vendored Normal file
View file

@ -0,0 +1,20 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
"""
See PEP 386 (https://www.python.org/dev/peps/pep-0386/)
Release logic:
1. Remove ".devX" from __version__ (below)
2. git add treebeard/__init__.py
3. git commit -m 'Bump to <version>'
4. git tag <version>
5. git push
6. assure that all tests pass on https://travis-ci.org/django-treebeard/django-treebeard/builds/
7. git push --tags
8. python setup.py sdist upload
9. bump the version, append ".dev0" to __version__
10. git add treebeard/__init__.py
11. git commit -m 'Start with <version>'
12. git push
"""
__version__ = '4.3.1'

147
vendor/treebeard/admin.py vendored Normal file
View file

@ -0,0 +1,147 @@
"""Django admin support for treebeard"""
import sys
import django
from django.conf import settings
from django.urls import re_path as url
from django.contrib import admin, messages
from django.http import HttpResponse, HttpResponseBadRequest
from django.utils.translation import gettext_lazy as _
if sys.version_info >= (3, 0):
from django.utils.encoding import force_str
else:
from django.utils.encoding import force_unicode as force_str
from treebeard.exceptions import (InvalidPosition, MissingNodeOrderBy,
InvalidMoveToDescendant, PathOverflow)
from treebeard.al_tree import AL_Node
try:
from django.contrib.admin.options import TO_FIELD_VAR
except ImportError:
from django.contrib.admin.views.main import TO_FIELD_VAR
class TreeAdmin(admin.ModelAdmin):
"""Django Admin class for treebeard."""
change_list_template = 'admin/tree_change_list.html'
def get_queryset(self, request):
if issubclass(self.model, AL_Node):
# AL Trees return a list instead of a QuerySet for .get_tree()
# So we're returning the regular .get_queryset cause we will use
# the old admin
return super(TreeAdmin, self).get_queryset(request)
else:
return self.model.get_tree()
def changelist_view(self, request, extra_context=None):
if issubclass(self.model, AL_Node):
# For AL trees, use the old admin display
self.change_list_template = 'admin/tree_list.html'
if extra_context is None:
extra_context = {}
if django.VERSION < (1, 10):
request_context = 'django.core.context_processors.request' in settings.TEMPLATE_CONTEXT_PROCESSORS
else:
request_context = any(
map(
lambda tmpl:
tmpl.get('BACKEND', None) == 'django.template.backends.django.DjangoTemplates' and
tmpl.get('APP_DIRS', False) and
'django.template.context_processors.request' in tmpl.get('OPTIONS', {}).get('context_processors', []),
settings.TEMPLATES
)
)
lacks_request = ('request' not in extra_context and not request_context)
if lacks_request:
extra_context['request'] = request
return super(TreeAdmin, self).changelist_view(request, extra_context)
def get_urls(self):
"""
Adds a url to move nodes to this admin
"""
urls = super(TreeAdmin, self).get_urls()
if django.VERSION < (1, 10):
from django.views.i18n import javascript_catalog
jsi18n_url = url(r'^jsi18n/$', javascript_catalog, {'packages': ('treebeard',)})
else:
from django.views.i18n import JavaScriptCatalog
jsi18n_url = url(r'^jsi18n/$',
JavaScriptCatalog.as_view(packages=['treebeard']),
name='javascript-catalog'
)
new_urls = [
url('^move/$', self.admin_site.admin_view(self.move_node), ),
jsi18n_url,
]
return new_urls + urls
def get_node(self, node_id):
return self.model.objects.get(pk=node_id)
def try_to_move_node(self, as_child, node, pos, request, target):
try:
node.move(target, pos=pos)
# Call the save method on the (reloaded) node in order to trigger
# possible signal handlers etc.
node = self.get_node(node.pk)
node.save()
except (MissingNodeOrderBy, PathOverflow, InvalidMoveToDescendant,
InvalidPosition):
e = sys.exc_info()[1]
# An error was raised while trying to move the node, then set an
# error message and return 400, this will cause a reload on the
# client to show the message
messages.error(request,
_('Exception raised while moving node: %s') % _(
force_str(e)))
return HttpResponseBadRequest('Exception raised during move')
if as_child:
msg = _('Moved node "%(node)s" as child of "%(other)s"')
else:
msg = _('Moved node "%(node)s" as sibling of "%(other)s"')
messages.info(request, msg % {'node': node, 'other': target})
return HttpResponse('OK')
def move_node(self, request):
try:
node_id = request.POST['node_id']
target_id = request.POST['sibling_id']
as_child = bool(int(request.POST.get('as_child', 0)))
except (KeyError, ValueError):
# Some parameters were missing return a BadRequest
return HttpResponseBadRequest('Malformed POST params')
node = self.get_node(node_id)
target = self.get_node(target_id)
is_sorted = True if node.node_order_by else False
pos = {
(True, True): 'sorted-child',
(True, False): 'last-child',
(False, True): 'sorted-sibling',
(False, False): 'left',
}[as_child, is_sorted]
return self.try_to_move_node(as_child, node, pos, request, target)
def admin_factory(form_class):
"""Dynamically build a TreeAdmin subclass for the given form class.
:param form_class:
:return: A TreeAdmin subclass.
"""
return type(
form_class.__name__ + 'Admin',
(TreeAdmin,),
dict(form=form_class))

404
vendor/treebeard/al_tree.py vendored Normal file
View file

@ -0,0 +1,404 @@
"""Adjacency List"""
from django.core import serializers
from django.db import models, transaction
from django.utils.translation import gettext_noop as _
from treebeard.exceptions import InvalidMoveToDescendant, NodeAlreadySaved
from treebeard.models import Node
def get_result_class(cls):
"""
For the given model class, determine what class we should use for the
nodes returned by its tree methods (such as get_children).
Usually this will be trivially the same as the initial model class,
but there are special cases when model inheritance is in use:
* If the model extends another via multi-table inheritance, we need to
use whichever ancestor originally implemented the tree behaviour (i.e.
the one which defines the 'parent' field). We can't use the
subclass, because it's not guaranteed that the other nodes reachable
from the current one will be instances of the same subclass.
* If the model is a proxy model, the returned nodes should also use
the proxy class.
"""
base_class = cls._meta.get_field('parent').model
if cls._meta.proxy_for_model == base_class:
return cls
else:
return base_class
class AL_NodeManager(models.Manager):
"""Custom manager for nodes in an Adjacency List tree."""
def get_queryset(self):
"""Sets the custom queryset as the default."""
if self.model.node_order_by:
order_by = ['parent'] + list(self.model.node_order_by)
else:
order_by = ['parent', 'sib_order']
return super(AL_NodeManager, self).get_queryset().order_by(*order_by)
class AL_Node(Node):
"""Abstract model to create your own Adjacency List Trees."""
objects = AL_NodeManager()
node_order_by = None
@classmethod
def add_root(cls, **kwargs):
"""Adds a root node to the tree."""
if len(kwargs) == 1 and 'instance' in kwargs:
# adding the passed (unsaved) instance to the tree
newobj = kwargs['instance']
if newobj.pk:
raise NodeAlreadySaved("Attempted to add a tree node that is "\
"already in the database")
else:
newobj = cls(**kwargs)
newobj._cached_depth = 1
if not cls.node_order_by:
try:
max = get_result_class(cls).objects.filter(
parent__isnull=True).order_by(
'sib_order').reverse()[0].sib_order
except IndexError:
max = 0
newobj.sib_order = max + 1
newobj.save()
return newobj
@classmethod
def get_root_nodes(cls):
""":returns: A queryset containing the root nodes in the tree."""
return get_result_class(cls).objects.filter(parent__isnull=True)
def get_depth(self, update=False):
"""
:returns: the depth (level) of the node
Caches the result in the object itself to help in loops.
:param update: Updates the cached value.
"""
if self.parent_id is None:
return 1
try:
if update:
del self._cached_depth
else:
return self._cached_depth
except AttributeError:
pass
depth = 0
node = self
while node:
node = node.parent
depth += 1
self._cached_depth = depth
return depth
def get_children(self):
""":returns: A queryset of all the node's children"""
return get_result_class(self.__class__).objects.filter(parent=self)
def get_parent(self, update=False):
""":returns: the parent node of the current node object."""
if self._meta.proxy_for_model:
# the current node is a proxy model; the returned parent
# should be the same proxy model, so we need to explicitly
# fetch it as an instance of that model rather than simply
# following the 'parent' relation
if self.parent_id is None:
return None
else:
return self.__class__.objects.get(pk=self.parent_id)
else:
return self.parent
def get_ancestors(self):
"""
:returns: A *list* containing the current node object's ancestors,
starting by the root node and descending to the parent.
"""
ancestors = []
if self._meta.proxy_for_model:
# the current node is a proxy model; our result set
# should use the same proxy model, so we need to
# explicitly fetch instances of that model
# when following the 'parent' relation
cls = self.__class__
node = self
while node.parent_id:
node = cls.objects.get(pk=node.parent_id)
ancestors.insert(0, node)
else:
node = self.parent
while node:
ancestors.insert(0, node)
node = node.parent
return ancestors
def get_root(self):
""":returns: the root node for the current node object."""
ancestors = self.get_ancestors()
if ancestors:
return ancestors[0]
return self
def is_descendant_of(self, node):
"""
:returns: ``True`` if the node if a descendant of another node given
as an argument, else, returns ``False``
"""
return self.pk in [obj.pk for obj in node.get_descendants()]
@classmethod
def dump_bulk(cls, parent=None, keep_ids=True):
"""Dumps a tree branch to a python data structure."""
serializable_cls = cls._get_serializable_model()
if (
parent and serializable_cls != cls and
parent.__class__ != serializable_cls
):
parent = serializable_cls.objects.get(pk=parent.pk)
# a list of nodes: not really a queryset, but it works
objs = serializable_cls.get_tree(parent)
ret, lnk = [], {}
for node, pyobj in zip(objs, serializers.serialize('python', objs)):
depth = node.get_depth()
# django's serializer stores the attributes in 'fields'
fields = pyobj['fields']
del fields['parent']
# non-sorted trees have this
if 'sib_order' in fields:
del fields['sib_order']
if 'id' in fields:
del fields['id']
newobj = {'data': fields}
if keep_ids:
newobj['id'] = pyobj['pk']
if (not parent and depth == 1) or\
(parent and depth == parent.get_depth()):
ret.append(newobj)
else:
parentobj = lnk[node.parent_id]
if 'children' not in parentobj:
parentobj['children'] = []
parentobj['children'].append(newobj)
lnk[node.pk] = newobj
return ret
def add_child(self, **kwargs):
"""Adds a child to the node."""
cls = get_result_class(self.__class__)
if len(kwargs) == 1 and 'instance' in kwargs:
# adding the passed (unsaved) instance to the tree
newobj = kwargs['instance']
if newobj.pk:
raise NodeAlreadySaved("Attempted to add a tree node that is "\
"already in the database")
else:
newobj = cls(**kwargs)
try:
newobj._cached_depth = self._cached_depth + 1
except AttributeError:
pass
if not cls.node_order_by:
try:
max = cls.objects.filter(parent=self).reverse(
)[0].sib_order
except IndexError:
max = 0
newobj.sib_order = max + 1
newobj.parent = self
newobj.save()
return newobj
@classmethod
def _get_tree_recursively(cls, results, parent, depth):
if parent:
nodes = parent.get_children()
else:
nodes = cls.get_root_nodes()
for node in nodes:
node._cached_depth = depth
results.append(node)
cls._get_tree_recursively(results, node, depth + 1)
@classmethod
def get_tree(cls, parent=None):
"""
:returns: A list of nodes ordered as DFS, including the parent. If
no parent is given, the entire tree is returned.
"""
if parent:
depth = parent.get_depth() + 1
results = [parent]
else:
depth = 1
results = []
cls._get_tree_recursively(results, parent, depth)
return results
def get_descendants(self):
"""
:returns: A *list* of all the node's descendants, doesn't
include the node itself
"""
return self.__class__.get_tree(parent=self)[1:]
def get_descendant_count(self):
""":returns: the number of descendants of a nodee"""
return len(self.get_descendants())
def get_siblings(self):
"""
:returns: A queryset of all the node's siblings, including the node
itself.
"""
if self.parent:
return get_result_class(self.__class__).objects.filter(
parent=self.parent)
return self.__class__.get_root_nodes()
def add_sibling(self, pos=None, **kwargs):
"""Adds a new node as a sibling to the current node object."""
pos = self._prepare_pos_var_for_add_sibling(pos)
if len(kwargs) == 1 and 'instance' in kwargs:
# adding the passed (unsaved) instance to the tree
newobj = kwargs['instance']
if newobj.pk:
raise NodeAlreadySaved("Attempted to add a tree node that is "\
"already in the database")
else:
# creating a new object
newobj = get_result_class(self.__class__)(**kwargs)
if not self.node_order_by:
newobj.sib_order = self.__class__._get_new_sibling_order(pos,
self)
newobj.parent_id = self.parent_id
newobj.save()
return newobj
@classmethod
def _is_target_pos_the_last_sibling(cls, pos, target):
return pos == 'last-sibling' or (
pos == 'right' and target == target.get_last_sibling())
@classmethod
def _make_hole_in_db(cls, min, target_node):
qset = get_result_class(cls).objects.filter(sib_order__gte=min)
if target_node.is_root():
qset = qset.filter(parent__isnull=True)
else:
qset = qset.filter(parent=target_node.parent)
qset.update(sib_order=models.F('sib_order') + 1)
@classmethod
def _make_hole_and_get_sibling_order(cls, pos, target_node):
siblings = target_node.get_siblings()
siblings = {
'left': siblings.filter(sib_order__gte=target_node.sib_order),
'right': siblings.filter(sib_order__gt=target_node.sib_order),
'first-sibling': siblings
}[pos]
sib_order = {
'left': target_node.sib_order,
'right': target_node.sib_order + 1,
'first-sibling': 1
}[pos]
try:
min = siblings.order_by('sib_order')[0].sib_order
except IndexError:
min = 0
if min:
cls._make_hole_in_db(min, target_node)
return sib_order
@classmethod
def _get_new_sibling_order(cls, pos, target_node):
if cls._is_target_pos_the_last_sibling(pos, target_node):
sib_order = target_node.get_last_sibling().sib_order + 1
else:
sib_order = cls._make_hole_and_get_sibling_order(pos, target_node)
return sib_order
def move(self, target, pos=None):
"""
Moves the current node and all it's descendants to a new position
relative to another node.
"""
pos = self._prepare_pos_var_for_move(pos)
sib_order = None
parent = None
if pos in ('first-child', 'last-child', 'sorted-child'):
# moving to a child
if not target.is_leaf():
target = target.get_last_child()
pos = {'first-child': 'first-sibling',
'last-child': 'last-sibling',
'sorted-child': 'sorted-sibling'}[pos]
else:
parent = target
if pos == 'sorted-child':
pos = 'sorted-sibling'
else:
pos = 'first-sibling'
sib_order = 1
if target.is_descendant_of(self):
raise InvalidMoveToDescendant(
_("Can't move node to a descendant."))
if self == target and (
(pos == 'left') or
(pos in ('right', 'last-sibling') and
target == target.get_last_sibling()) or
(pos == 'first-sibling' and
target == target.get_first_sibling())):
# special cases, not actually moving the node so no need to UPDATE
return
if pos == 'sorted-sibling':
if parent:
self.parent = parent
else:
self.parent = target.parent
else:
if sib_order:
self.sib_order = sib_order
else:
self.sib_order = self.__class__._get_new_sibling_order(pos,
target)
if parent:
self.parent = parent
else:
self.parent = target.parent
self.save()
class Meta:
"""Abstract model."""
abstract = True

31
vendor/treebeard/exceptions.py vendored Normal file
View file

@ -0,0 +1,31 @@
"""Treebeard exceptions"""
class InvalidPosition(Exception):
"""Raised when passing an invalid pos value"""
class InvalidMoveToDescendant(Exception):
"""Raised when attemping to move a node to one of it's descendants."""
class NodeAlreadySaved(Exception):
"""
Raised when attempting to add a node which is already saved to the
database.
"""
class MissingNodeOrderBy(Exception):
"""
Raised when an operation needs a missing
:attr:`~treebeard.MP_Node.node_order_by` attribute
"""
class PathOverflow(Exception):
"""
Raised when trying to add or move a node to a position where no more nodes
can be added (see :attr:`~treebeard.MP_Node.path` and
:attr:`~treebeard.MP_Node.alphabet` for more info)
"""

229
vendor/treebeard/forms.py vendored Normal file
View file

@ -0,0 +1,229 @@
"""Forms for treebeard."""
from django import forms
from django.db.models.query import QuerySet
from django.forms.models import BaseModelForm, ErrorList, model_to_dict
from django.forms.models import modelform_factory as django_modelform_factory
from django.utils.html import escape
from django.utils.safestring import mark_safe
from django.utils.translation import gettext_lazy as _
from treebeard.al_tree import AL_Node
from treebeard.mp_tree import MP_Node
from treebeard.ns_tree import NS_Node
class MoveNodeForm(forms.ModelForm):
"""
Form to handle moving a node in a tree.
Handles sorted/unsorted trees.
It adds two fields to the form:
- Relative to: The target node where the current node will
be moved to.
- Position: The position relative to the target node that
will be used to move the node. These can be:
- For sorted trees: ``Child of`` and ``Sibling of``
- For unsorted trees: ``First child of``, ``Before`` and
``After``
.. warning::
Subclassing :py:class:`MoveNodeForm` directly is
discouraged, since special care is needed to handle
excluded fields, and these change depending on the
tree type.
It is recommended that the :py:func:`movenodeform_factory`
function is used instead.
"""
__position_choices_sorted = (
('sorted-child', _('Child of')),
('sorted-sibling', _('Sibling of')),
)
__position_choices_unsorted = (
('first-child', _('First child of')),
('left', _('Before')),
('right', _('After')),
)
_position = forms.ChoiceField(required=True, label=_("Position"))
_ref_node_id = forms.TypedChoiceField(required=False,
coerce=int,
label=_("Relative to"))
def _get_position_ref_node(self, instance):
if self.is_sorted:
position = 'sorted-child'
node_parent = instance.get_parent()
if node_parent:
ref_node_id = node_parent.pk
else:
ref_node_id = ''
else:
prev_sibling = instance.get_prev_sibling()
if prev_sibling:
position = 'right'
ref_node_id = prev_sibling.pk
else:
position = 'first-child'
if instance.is_root():
ref_node_id = ''
else:
ref_node_id = instance.get_parent().pk
return {'_ref_node_id': ref_node_id,
'_position': position}
def __init__(self, data=None, files=None, auto_id='id_%s', prefix=None,
initial=None, error_class=ErrorList, label_suffix=':',
empty_permitted=False, instance=None, **kwargs):
opts = self._meta
if opts.model is None:
raise ValueError('ModelForm has no model class specified')
# update the '_position' field choices
self.is_sorted = getattr(opts.model, 'node_order_by', False)
if self.is_sorted:
choices_sort_mode = self.__class__.__position_choices_sorted
else:
choices_sort_mode = self.__class__.__position_choices_unsorted
self.declared_fields['_position'].choices = choices_sort_mode
# update the '_ref_node_id' choices
choices = self.mk_dropdown_tree(opts.model, for_node=instance)
self.declared_fields['_ref_node_id'].choices = choices
# put initial data for these fields into a map, update the map with
# initial data, and pass this new map to the parent constructor as
# initial data
if instance is None:
initial_ = {}
else:
initial_ = self._get_position_ref_node(instance)
if initial is not None:
initial_.update(initial)
super(MoveNodeForm, self).__init__(
data=data, files=files, auto_id=auto_id, prefix=prefix,
initial=initial_, error_class=error_class,
label_suffix=label_suffix, empty_permitted=empty_permitted,
instance=instance, **kwargs)
def _clean_cleaned_data(self):
""" delete auxilary fields not belonging to node model """
reference_node_id = 0
if '_ref_node_id' in self.cleaned_data:
reference_node_id = self.cleaned_data['_ref_node_id']
del self.cleaned_data['_ref_node_id']
position_type = self.cleaned_data['_position']
del self.cleaned_data['_position']
return position_type, reference_node_id
def save(self, commit=True):
position_type, reference_node_id = self._clean_cleaned_data()
if self.instance.pk is None:
cl_data = {}
for field in self.cleaned_data:
if not isinstance(self.cleaned_data[field], (list, QuerySet)):
cl_data[field] = self.cleaned_data[field]
if reference_node_id:
reference_node = self._meta.model.objects.get(
pk=reference_node_id)
self.instance = reference_node.add_child(**cl_data)
self.instance.move(reference_node, pos=position_type)
else:
self.instance = self._meta.model.add_root(**cl_data)
else:
self.instance.save()
if reference_node_id:
reference_node = self._meta.model.objects.get(
pk=reference_node_id)
self.instance.move(reference_node, pos=position_type)
else:
if self.is_sorted:
pos = 'sorted-sibling'
else:
pos = 'first-sibling'
self.instance.move(self._meta.model.get_first_root_node(), pos)
# Reload the instance
self.instance = self._meta.model.objects.get(pk=self.instance.pk)
super(MoveNodeForm, self).save(commit=commit)
return self.instance
@staticmethod
def is_loop_safe(for_node, possible_parent):
if for_node is not None:
return not (
possible_parent == for_node
) or (possible_parent.is_descendant_of(for_node))
return True
@staticmethod
def mk_indent(level):
return '&nbsp;&nbsp;&nbsp;&nbsp;' * (level - 1)
@classmethod
def add_subtree(cls, for_node, node, options):
""" Recursively build options tree. """
if cls.is_loop_safe(for_node, node):
options.append(
(node.pk,
mark_safe(cls.mk_indent(node.get_depth()) + escape(node))))
for subnode in node.get_children():
cls.add_subtree(for_node, subnode, options)
@classmethod
def mk_dropdown_tree(cls, model, for_node=None):
""" Creates a tree-like list of choices """
options = [(0, _('-- root --'))]
for node in model.get_root_nodes():
cls.add_subtree(for_node, node, options)
return options
def movenodeform_factory(model, form=MoveNodeForm, fields=None, exclude=None,
formfield_callback=None, widgets=None):
"""Dynamically build a MoveNodeForm subclass with the proper Meta.
:param Node model:
The subclass of :py:class:`Node` that will be handled
by the form.
:param form:
The form class that will be used as a base. By
default, :py:class:`MoveNodeForm` will be used.
:return: A :py:class:`MoveNodeForm` subclass
"""
_exclude = _get_exclude_for_model(model, exclude)
return django_modelform_factory(
model, form, fields, _exclude, formfield_callback, widgets)
def _get_exclude_for_model(model, exclude):
if exclude:
_exclude = tuple(exclude)
else:
_exclude = ()
if issubclass(model, AL_Node):
_exclude += ('sib_order', 'parent')
elif issubclass(model, MP_Node):
_exclude += ('depth', 'numchild', 'path')
elif issubclass(model, NS_Node):
_exclude += ('depth', 'lft', 'rgt', 'tree_id')
return _exclude

Some files were not shown because too many files have changed in this diff Show more