taiga
/
taiga-back
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

US #73: Create a user schemaless storage system

remotes/origin/enhancement/email-actions
David Barragán Merino 2014-05-14 14:01:31 +02:00
parent 4554c0919d
commit d786a1ba95
11 changed files with 452 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
settings/common.py
View File

@ -177,6 +177,7 @@ INSTALLED_APPS = [
"taiga.domains",
"taiga.front",
"taiga.users",
"taiga.userstorage",
"taiga.projects",
"taiga.projects.attachments",
"taiga.projects.milestones",

6
taiga/routers.py
View File

@ -28,6 +28,12 @@ router.register(r"permissions", PermissionsViewSet, base_name="permissions")
router.register(r"auth", AuthViewSet, base_name="auth")
#taiga.userstorage
from taiga.userstorage.api import StorageEntriesViewSet
router.register(r"user-storage", StorageEntriesViewSet, base_name="user-storage")
# Resolver & Search
from taiga.base.searches.api import SearchViewSet
from taiga.base.resolver.api import ResolverViewSet

0
taiga/userstorage/__init__.py Normal file
View File

48
taiga/userstorage/api.py Normal file
View File

@ -0,0 +1,48 @@
# Copyright (C) 2014 Andrey Antukh <niwi@niwi.be>
# Copyright (C) 2014 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014 David Barragán <bameda@dbarragan.com>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from django.utils.translation import ugettext_lazy as _
from django.db import IntegrityError
from rest_framework.permissions import IsAuthenticated
from taiga.base.api import ModelCrudViewSet
from taiga.base import exceptions as exc
from . import models
from . import serializers
from . import permissions
class StorageEntriesViewSet(ModelCrudViewSet):
model = models.StorageEntry
serializer_class = serializers.StorageEntrySerializer
permission_classes = (IsAuthenticated, permissions.StorageEntriesPermission)
lookup_field = "key"
def get_queryset(self):
return self.request.user.storage_entries.all()
def pre_save(self, obj):
obj.owner = self.request.user
def create(self, *args, **kwargs):
try:
return super().create(*args, **kwargs)
except IntegrityError:
key = self.request.DATA.get("key", None)
raise exc.IntegrityError(_("Duplicate key value violates unique constraint. "
"Key '{}' already exists.").format(key))

91
taiga/userstorage/migrations/0001_initial.py Normal file
View File

@ -0,0 +1,91 @@
# -*- coding: utf-8 -*-
from south.utils import datetime_utils as datetime
from south.db import db
from south.v2 import SchemaMigration
from django.db import models
class Migration(SchemaMigration):
def forwards(self, orm):
# Adding model 'StorageEntry'
db.create_table('userstorage_storageentry', (
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
('owner', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['users.User'], related_name='storage_entries')),
('created_date', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, blank=True)),
('modified_date', self.gf('django.db.models.fields.DateTimeField')(auto_now=True, blank=True)),
('key', self.gf('django.db.models.fields.CharField')(max_length=255)),
('value', self.gf('django_pgjson.fields.JsonField')(default=None)),
))
db.send_create_signal('userstorage', ['StorageEntry'])
# Adding unique constraint on 'StorageEntry', fields ['owner', 'key']
db.create_unique('userstorage_storageentry', ['owner_id', 'key'])
def backwards(self, orm):
# Removing unique constraint on 'StorageEntry', fields ['owner', 'key']
db.delete_unique('userstorage_storageentry', ['owner_id', 'key'])
# Deleting model 'StorageEntry'
db.delete_table('userstorage_storageentry')
models = {
'auth.group': {
'Meta': {'object_name': 'Group'},
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '80'}),
'permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'symmetrical': 'False', 'blank': 'True'})
},
'auth.permission': {
'Meta': {'object_name': 'Permission', 'unique_together': "(('content_type', 'codename'),)", 'ordering': "('content_type__app_label', 'content_type__model', 'codename')"},
'codename': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '50'})
},
'contenttypes.contenttype': {
'Meta': {'db_table': "'django_content_type'", 'object_name': 'ContentType', 'unique_together': "(('app_label', 'model'),)", 'ordering': "('name',)"},
'app_label': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'model': ('django.db.models.fields.CharField', [], {'max_length': '100'}),
'name': ('django.db.models.fields.CharField', [], {'max_length': '100'})
},
'users.user': {
'Meta': {'object_name': 'User', 'ordering': "['username']"},
'color': ('django.db.models.fields.CharField', [], {'default': "'#2ee685'", 'max_length': '9', 'blank': 'True'}),
'colorize_tags': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'date_joined': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'default_language': ('django.db.models.fields.CharField', [], {'default': "''", 'max_length': '20', 'blank': 'True'}),
'default_timezone': ('django.db.models.fields.CharField', [], {'default': "''", 'max_length': '20', 'blank': 'True'}),
'description': ('django.db.models.fields.TextField', [], {'blank': 'True'}),
'email': ('django.db.models.fields.EmailField', [], {'max_length': '75', 'blank': 'True'}),
'first_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'groups': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Group']", 'related_name': "'user_set'", 'symmetrical': 'False', 'blank': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'is_active': ('django.db.models.fields.BooleanField', [], {'default': 'True'}),
'is_staff': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'is_superuser': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'last_login': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime.now'}),
'last_name': ('django.db.models.fields.CharField', [], {'max_length': '30', 'blank': 'True'}),
'notify_changes_by_me': ('django.db.models.fields.BooleanField', [], {'default': 'False'}),
'notify_level': ('django.db.models.fields.CharField', [], {'default': "'all_owned_projects'", 'max_length': '32'}),
'password': ('django.db.models.fields.CharField', [], {'max_length': '128'}),
'photo': ('django.db.models.fields.files.FileField', [], {'null': 'True', 'max_length': '500', 'blank': 'True'}),
'token': ('django.db.models.fields.CharField', [], {'default': 'None', 'null': 'True', 'max_length': '200', 'blank': 'True'}),
'user_permissions': ('django.db.models.fields.related.ManyToManyField', [], {'to': "orm['auth.Permission']", 'related_name': "'user_set'", 'symmetrical': 'False', 'blank': 'True'}),
'username': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '30'})
},
'userstorage.storageentry': {
'Meta': {'object_name': 'StorageEntry', 'unique_together': "(('owner', 'key'),)", 'ordering': "['owner', 'key']"},
'created_date': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
'key': ('django.db.models.fields.CharField', [], {'max_length': '255'}),
'modified_date': ('django.db.models.fields.DateTimeField', [], {'auto_now': 'True', 'blank': 'True'}),
'owner': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['users.User']", 'related_name': "'storage_entries'"}),
'value': ('django_pgjson.fields.JsonField', [], {'default': 'None'})
}
}
complete_apps = ['userstorage']

0
taiga/userstorage/migrations/__init__.py Normal file
View File

37
taiga/userstorage/models.py Normal file
View File

@ -0,0 +1,37 @@
# Copyright (C) 2014 Andrey Antukh <niwi@niwi.be>
# Copyright (C) 2014 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014 David Barragán <bameda@dbarragan.com>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from django.db import models
from django.conf import settings
from django.utils.translation import ugettext_lazy as _
from django_pgjson.fields import JsonField
class StorageEntry(models.Model):
owner = models.ForeignKey(settings.AUTH_USER_MODEL, blank=False, null=False,
related_name="storage_entries", verbose_name=_("owner"))
created_date = models.DateTimeField(auto_now_add=True, null=False, blank=False,
verbose_name=_("created date"))
modified_date = models.DateTimeField(auto_now=True, null=False, blank=False,
verbose_name=_("modified date"))
key = models.CharField(max_length=255, null=False, blank=False, verbose_name=_("key"))
value = JsonField(blank=True, default=None, null=True, verbose_name=_("value"))
class Meta:
verbose_name = "storage entry"
verbose_name_plural = "storages entries"
unique_together = ("owner", "key")
ordering = ["owner", "key"]

22
taiga/userstorage/permissions.py Normal file
View File

@ -0,0 +1,22 @@
# Copyright (C) 2014 Andrey Antukh <niwi@niwi.be>
# Copyright (C) 2014 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014 David Barragán <bameda@dbarragan.com>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from taiga.base.permissions import Permission
class StorageEntriesPermission(Permission):
def has_object_permission(self, request, view, obj):
return request.user == obj.owner

29
taiga/userstorage/serializers.py Normal file
View File

@ -0,0 +1,29 @@
# Copyright (C) 2014 Andrey Antukh <niwi@niwi.be>
# Copyright (C) 2014 Jesús Espino <jespinog@gmail.com>
# Copyright (C) 2014 David Barragán <bameda@dbarragan.com>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from rest_framework import serializers
from taiga.base.serializers import JsonField
from . import models
class StorageEntrySerializer(serializers.ModelSerializer):
value = JsonField(label="value")
class Meta:
model = models.StorageEntry
fields = ("key", "value", "created_date", "modified_date")
read_only_fields = ("created_date", "modified_date")

10
tests/factories.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
import uuid
import factory
@ -7,6 +6,7 @@ from django.conf import settings
import taiga.domains.models
import taiga.projects.models
import taiga.users.models
import taiga.userstorage.models
class DomainFactory(factory.DjangoModelFactory):
@ -60,3 +60,11 @@ class MembershipFactory(factory.DjangoModelFactory):
project = factory.SubFactory("tests.factories.ProjectFactory")
role = factory.SubFactory("tests.factories.RoleFactory")
user = factory.SubFactory("tests.factories.UserFactory")
class StorageEntryFactory(factory.DjangoModelFactory):
FACTORY_FOR = taiga.userstorage.models.StorageEntry
owner = factory.SubFactory("tests.factories.UserFactory")
key = factory.Sequence(lambda n: "key-{}".format(n))
value = factory.Sequence(lambda n: "value {}".format(n))

209
tests/integration/test_userstorage_api.py Normal file
View File

@ -0,0 +1,209 @@
import pytest
from rest_framework.reverse import reverse
from .. import factories
import json
pytestmark = pytest.mark.django_db
class TestListStorageEntries:
def _load_initial_data(self):
self.user1 = factories.UserFactory()
self.user2 = factories.UserFactory()
self.storage11 = factories.StorageEntryFactory(owner=self.user1)
self.storage12 = factories.StorageEntryFactory(owner=self.user1)
self.storage21 = factories.StorageEntryFactory(owner=self.user2)
def test_list_by_anonymous_user(self, client):
self._load_initial_data()
response = client.get(reverse("user-storage-list"))
assert response.status_code == 401
def test_list_only_user1_entriees(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
response = client.get(reverse("user-storage-list"))
assert response.status_code == 200
entries = response.data
assert len(entries) == 2
response = client.logout()
def test_list_only_user2_entriees(self, client):
self._load_initial_data()
response = client.login(username=self.user2.username, password=self.user2.username)
response = client.get(reverse("user-storage-list"))
assert response.status_code == 200
entries = response.data
assert len(entries) == 1
response = client.logout()
class TestViewStorageEntries:
def _load_initial_data(self):
self.user1 = factories.UserFactory()
self.user2 = factories.UserFactory()
self.storage11 = factories.StorageEntryFactory(owner=self.user1)
def test_view_an_entry_by_anonymous_user(self, client):
self._load_initial_data()
response = client.get(reverse("user-storage-detail", args=[self.storage11.key]))
assert response.status_code == 401
def test_view_an_entry(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
response = client.get(reverse("user-storage-detail", args=[self.storage11.key]))
assert response.status_code == 200
entry = response.data
assert entry["key"] == self.storage11.key
assert entry["value"] == self.storage11.value
response = client.logout()
def test_view_an_entry_by_incorrect_user(self, client):
self._load_initial_data()
response = client.login(username=self.user2.username, password=self.user2.username)
response = client.get(reverse("user-storage-detail", args=[self.storage11.key]))
assert response.status_code == 404
response = client.logout()
def test_view_non_existent_entry(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
response = client.get(reverse("user-storage-detail", args=["foo"]))
assert response.status_code == 404
response = client.logout()
class TestCreateStorageEntries:
@classmethod
def setup_class(cls):
cls.form = {"key": "foo",
"value": "bar"}
cls.form_without_key = {"value": "bar"}
cls.form_without_value = {"key": "foo"}
def _load_initial_data(self):
self.user1 = factories.UserFactory()
self.user2 = factories.UserFactory()
self.storage11 = factories.StorageEntryFactory(owner=self.user1)
def test_create_entry_by_anonymous_user_with_error(self, client):
self._load_initial_data()
response = client.post(reverse("user-storage-list"), self.form)
assert response.status_code == 401
def test_create_entry_successfully(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
response = client.post(reverse("user-storage-list"), self.form)
assert response.status_code == 201
response = client.get(reverse("user-storage-detail", args=[self.form["key"]]))
assert response.status_code == 200
response = client.logout()
def test_create_entry_with_incorret_form_error(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
response = client.post(reverse("user-storage-list"), self.form_without_key)
assert response.status_code == 400
response = client.post(reverse("user-storage-list"), self.form_without_value)
assert response.status_code == 400
response = client.logout()
def test_create_entry_with_integrity_error(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
error_form = {"key": self.storage11.key,
"value": "bar"}
response = client.post(reverse("user-storage-list"), error_form)
assert response.status_code == 400
response = client.logout()
class TestUpdateStorageEntries:
@classmethod
def setup_class(cls):
cls.form = {"value": "bar"}
def _load_initial_data(self):
self.user1 = factories.UserFactory()
self.user2 = factories.UserFactory()
self.storage11 = factories.StorageEntryFactory(owner=self.user1)
def test_update_entry_by_anonymous_user(self, client):
self._load_initial_data()
self.form["key"] = self.storage11.key
response = client.put(reverse("user-storage-detail", args=[self.storage11.key]),
json.dumps(self.form),
content_type='application/json')
assert response.status_code == 401
def test_update_entry(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
self.form["key"] = self.storage11.key
response = client.put(reverse("user-storage-detail", args=[self.storage11.key]),
json.dumps(self.form),
content_type='application/json')
assert response.status_code == 200
response = client.get(reverse("user-storage-detail", args=[self.storage11.key]))
assert response.status_code == 200
entry = response.data
assert entry["value"] == self.form["value"]
response = client.logout()
def test_update_non_existent_entry(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
self.form["key"] = "foo"
response = client.get(reverse("user-storage-detail", args=[self.form["key"]]))
assert response.status_code == 404
response = client.put(reverse("user-storage-detail", args=[self.form["key"]]),
json.dumps(self.form),
content_type='application/json')
assert response.status_code == 201
response = client.get(reverse("user-storage-detail", args=[self.form["key"]]))
assert response.status_code == 200
entry = response.data
assert entry["value"] == self.form["value"]
response = client.logout()
class TestDeleteStorageEntries:
def _load_initial_data(self):
self.user1 = factories.UserFactory()
self.user2 = factories.UserFactory()
self.storage11 = factories.StorageEntryFactory(owner=self.user1)
def test_delete_entry_by_anonymous_user(self, client):
self._load_initial_data()
response = client.delete(reverse("user-storage-detail", args=[self.storage11.key]))
assert response.status_code == 401
def test_delete_entry(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
key = self.storage11.key
response = client.delete(reverse("user-storage-detail", args=[key]))
assert response.status_code == 204
response = client.get(reverse("user-storage-detail", args=[key]))
assert response.status_code == 404
response = client.logout()
def test_delete_entry_by_incorrect_user(self, client):
self._load_initial_data()
response = client.login(username=self.user2.username, password=self.user2.username)
response = client.delete(reverse("user-storage-detail", args=[self.storage11.key]))
assert response.status_code == 404
response = client.logout()
def test_delete_non_existent_entry(self, client):
self._load_initial_data()
response = client.login(username=self.user1.username, password=self.user1.username)
response = client.delete(reverse("user-storage-detail", args=["foo"]))
assert response.status_code == 404
response = client.logout()