Timeline service implementation

settings/common.py

@@ -183,6 +183,7 @@ INSTALLED_APPS = [
     "taiga.projects.history",
     "taiga.projects.notifications",
     "taiga.projects.votes",
+    "taiga.timeline",
 
     "south",
     "reversion",

taiga/base/utils/db.py

@@ -1,3 +1,21 @@
+# 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.contrib.contenttypes.models import ContentType
+
 FILTER_TAGS_SQL = "unpickle({table}.tags) && %s"
 
 
@@ -7,3 +25,15 @@ def filter_by_tags(tags, queryset):
     where_sql = FILTER_TAGS_SQL.format(table=table_name)
 
     return queryset.extra(where=[where_sql], params=[tags])
+
+
+def get_typename_for_model_class(model:object, for_concrete_model=True) -> str:
+    """
+    Get typename for model instance.
+    """
+    if for_concrete_model:
+        model = model._meta.concrete_model
+    else:
+        model = model._meta.proxy_for_model
+
+    return "{0}.{1}".format(model._meta.app_label, model._meta.model_name)

taiga/projects/history/services.py

@@ -40,6 +40,7 @@ from .models import HistoryType
 
 from taiga.mdrender.service import render as mdrender
 from taiga.mdrender.service import get_diff_of_htmls
+from taiga.base.utils.db import get_typename_for_model_class
 
 # Type that represents a freezed object
 FrozenObj = namedtuple("FrozenObj", ["key", "snapshot"])
@@ -60,14 +61,6 @@ def make_key_from_model_object(obj:object) -> str:
     return "{0}:{1}".format(tn, obj.pk)
 
 
-def get_typename_for_model_class(model:object) -> str:
-    """
-    Get typename for model instance.
-    """
-    ct = ContentType.objects.get_for_model(model)
-    return "{0}.{1}".format(ct.app_label, ct.model)
-
-
 def register_values_implementation(typename:str, fn=None):
     """
     Register values implementation for specified typename.

taiga/routers.py

@@ -102,6 +102,14 @@ router.register(r"history/issue", IssueHistory, base_name="issue-history")
 router.register(r"history/wiki", WikiHistory, base_name="wiki-history")
 
 
+# Timelines
+from taiga.timeline.api import UserTimeline
+from taiga.timeline.api import ProjectTimeline
+
+router.register(r"timeline/user", UserTimeline, base_name="user-timeline")
+router.register(r"timeline/project", ProjectTimeline, base_name="project-timeline")
+
+
 # Project components
 from taiga.projects.milestones.api import MilestoneViewSet
 from taiga.projects.userstories.api import UserStoryViewSet

taiga/timeline/api.py

@@ -0,0 +1,73 @@
+# 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.contrib.contenttypes.models import ContentType
+from django.shortcuts import get_object_or_404
+
+from rest_framework.response import Response
+
+from taiga.base.api import GenericViewSet
+
+from . import serializers
+from . import service
+
+# TODO: Set Timelines permissions
+
+
+class TimelineViewSet(GenericViewSet):
+    serializer_class = serializers.TimelineSerializer
+
+    content_type = None
+
+    def get_content_type(self):
+        app_name, model = self.content_type.split(".", 1)
+        return get_object_or_404(ContentType, app_label=app_name, model=model)
+
+    def get_object(self):
+        ct = self.get_content_type()
+        model_cls = ct.model_class()
+
+        qs = model_cls.objects.all()
+        filtered_qs = self.filter_queryset(qs)
+        return super().get_object(queryset=filtered_qs)
+
+    def response_for_queryset(self, queryset):
+        # Switch between paginated or standard style responses
+        page = self.paginate_queryset(queryset)
+        if page is not None:
+            serializer = self.get_pagination_serializer(page)
+        else:
+            serializer = self.get_serializer(queryset, many=True)
+
+        return Response(serializer.data)
+
+    # Just for restframework! Because it raises
+    # 404 on main api root if this method not exists.
+    def list(self, request):
+        return Response({})
+
+    def retrieve(self, request, pk):
+        obj = self.get_object()
+        qs = service.get_timeline(obj)
+        return self.response_for_queryset(qs)
+
+
+class UserTimeline(TimelineViewSet):
+    content_type = "users.user"
+
+
+class ProjectTimeline(TimelineViewSet):
+    content_type = "projects.project"

taiga/timeline/migrations/0001_initial.py

@@ -0,0 +1,55 @@
+# -*- 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 'Timeline'
+        db.create_table('timeline_timeline', (
+            ('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
+            ('content_type', self.gf('django.db.models.fields.related.ForeignKey')(to=orm['contenttypes.ContentType'])),
+            ('object_id', self.gf('django.db.models.fields.PositiveIntegerField')()),
+            ('namespace', self.gf('django.db.models.fields.SlugField')(default='default', max_length=50)),
+            ('event_type', self.gf('django.db.models.fields.SlugField')(max_length=50)),
+            ('data', self.gf('django_pgjson.fields.JsonField')(null=False, blank=False)),
+            ('created', self.gf('django.db.models.fields.DateTimeField')(auto_now_add=True, blank=True)),
+        ))
+        db.send_create_signal('timeline', ['Timeline'])
+
+        # Adding index on 'Timeline', fields ['content_type', 'object_id', 'namespace']
+        db.create_index('timeline_timeline', ['content_type_id', 'object_id', 'namespace'])
+
+
+    def backwards(self, orm):
+        # Removing index on 'Timeline', fields ['content_type', 'object_id', 'namespace']
+        db.delete_index('timeline_timeline', ['content_type_id', 'object_id', 'namespace'])
+
+        # Deleting model 'Timeline'
+        db.delete_table('timeline_timeline')
+
+
+    models = {
+        'contenttypes.contenttype': {
+            'Meta': {'unique_together': "(('app_label', 'model'),)", 'ordering': "('name',)", 'db_table': "'django_content_type'", 'object_name': 'ContentType'},
+            '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'})
+        },
+        'timeline.timeline': {
+            'Meta': {'object_name': 'Timeline', 'index_together': "[('content_type', 'object_id', 'namespace')]"},
+            'content_type': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['contenttypes.ContentType']"}),
+            'created': ('django.db.models.fields.DateTimeField', [], {'auto_now_add': 'True', 'blank': 'True'}),
+            'data': ('django_pgjson.fields.JsonField', [], {'null': 'False', 'blank': 'False'}),
+            'event_type': ('django.db.models.fields.SlugField', [], {'max_length': '50'}),
+            'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
+            'namespace': ('django.db.models.fields.SlugField', [], {'default': "'default'", 'max_length': '50'}),
+            'object_id': ('django.db.models.fields.PositiveIntegerField', [], {})
+        }
+    }
+
+    complete_apps = ['timeline']

taiga/timeline/models.py

@@ -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.db import models
+from django_pgjson.fields import JsonField
+
+from django.core.exceptions import ValidationError
+
+from django.contrib.contenttypes.models import ContentType
+from django.contrib.contenttypes.generic import GenericForeignKey
+
+
+class Timeline(models.Model):
+    content_type = models.ForeignKey(ContentType)
+    object_id = models.PositiveIntegerField()
+    content_object = GenericForeignKey('content_type', 'object_id')
+    namespace = models.SlugField(default="default")
+    event_type = models.SlugField()
+    data = JsonField()
+    created = models.DateTimeField(auto_now_add=True)
+
+    def save(self, *args, **kwargs):
+        if self.id:
+            raise ValidationError("Not modify allowed for timeline entries")
+        return super().save(*args, **kwargs)
+
+    class Meta:
+        index_together = [('content_type', 'object_id', 'namespace'), ]
+
+
+# Register all implementations
+from .timeline_implementations import *
+
+# Register all signals
+from .signals import *

taiga/timeline/serializers.py

@@ -0,0 +1,27 @@
+# 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 TimelineSerializer(serializers.ModelSerializer):
+    data = JsonField()
+
+    class Meta:
+        model = models.Timeline

taiga/timeline/service.py

@@ -0,0 +1,95 @@
+# 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.models import Model
+from django.db.models.query import QuerySet
+from functools import partial, wraps
+from django.contrib.contenttypes.models import ContentType
+
+from taiga.base.utils.db import get_typename_for_model_class
+
+_timeline_impl_map = {}
+
+
+def _get_impl_key_from_model(model:Model, event_type:str):
+    if issubclass(model, Model):
+        typename = get_typename_for_model_class(model)
+        return _get_impl_key_from_typename(typename, event_type)
+    raise Exception("Not valid model parameter")
+
+
+def _get_impl_key_from_typename(typename:str, event_type:str):
+    if isinstance(typename, str):
+        return "{0}.{1}".format(typename, event_type)
+    raise Exception("Not valid typename parameter")
+
+
+def _get_class_implementation(model:Model, event_type:str):
+    key = _get_impl_key_from_model(model, event_type)
+    return _timeline_impl_map.get(key, None)
+
+
+def _add_to_object_timeline(obj:object, instance:object, event_type:str, namespace:str="default", extra_data:dict={}):
+    assert isinstance(obj, Model), "obj must be a instance of Model"
+    assert isinstance(instance, Model), "instance must be a instance of Model"
+    from .models import Timeline
+
+    impl = _get_class_implementation(instance.__class__, event_type)
+    Timeline.objects.create(
+        content_object=obj,
+        namespace=namespace,
+        event_type=event_type,
+        data=impl(instance, extra_data=extra_data)
+    )
+
+
+def _add_to_objects_timeline(objects, instance:object, event_type:str, namespace:str="default", extra_data:dict={}):
+    for obj in objects:
+        _add_to_object_timeline(obj, instance, event_type, namespace, extra_data)
+
+
+def push_to_timeline(objects, instance:object, event_type:str, namespace:str="default", extra_data:dict={}):
+    if isinstance(objects, Model):
+        _add_to_object_timeline(objects, instance, event_type, namespace, extra_data)
+    elif isinstance(objects, QuerySet) or isinstance(objects, list):
+        _add_to_objects_timeline(objects, instance, event_type, namespace, extra_data)
+    else:
+        raise Exception("Invalid objects parameter")
+
+
+def get_timeline(obj, namespace="default"):
+    assert isinstance(obj, Model), "obj must be a instance of Model"
+    from .models import Timeline
+
+    ct = ContentType.objects.get_for_model(obj.__class__)
+    return Timeline.objects.filter(content_type=ct, object_id=obj.pk, namespace=namespace)
+
+
+def register_timeline_implementation(typename:str, event_type:str, fn=None):
+    assert isinstance(typename, str), "typename must be a string"
+    assert isinstance(event_type, str), "event_type must be a string"
+
+    if fn is None:
+        return partial(register_timeline_implementation, typename, event_type)
+
+    @wraps(fn)
+    def _wrapper(*args, **kwargs):
+        return fn(*args, **kwargs)
+
+    key = _get_impl_key_from_typename(typename, event_type)
+
+    _timeline_impl_map[key] = _wrapper
+    return _wrapper

taiga/timeline/signals.py

@@ -0,0 +1,65 @@
+# 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.models.loading import get_model
+from django.db.models import signals
+from django.dispatch import receiver
+
+from taiga.timeline.service import push_to_timeline
+
+# TODO: Add events to followers timeline when followers are implemented.
+# TODO: Add events to project watchers timeline when project watchers are implemented.
+
+
+@receiver(signals.post_save, sender=get_model("projects", "Project"))
+def create_project_push_to_timeline(sender, instance, created, **kwargs):
+    if created:
+        push_to_timeline(instance, instance, "create")
+
+
+@receiver(signals.post_save, sender=get_model("userstories", "UserStory"))
+def create_user_story_push_to_timeline(sender, instance, created, **kwargs):
+    if created:
+        push_to_timeline(instance.project, instance, "create")
+
+
+@receiver(signals.post_save, sender=get_model("issues", "Issue"))
+def create_issue_push_to_timeline(sender, instance, created, **kwargs):
+    if created:
+        push_to_timeline(instance.project, instance, "create")
+
+
+@receiver(signals.pre_save, sender=get_model("projects", "Membership"))
+def create_membership_push_to_timeline(sender, instance, **kwargs):
+    if not instance.pk and instance.user:
+        push_to_timeline(instance.project, instance, "create")
+    elif instance.pk:
+        prev_instance = sender.objects.get(pk=instance.pk)
+        if prev_instance.user != prev_instance.user:
+            push_to_timeline(instance.project, instance, "create")
+        elif prev_instance.role != prev_instance.role:
+            extra_data = {
+                "prev_role": {
+                    "id": prev_instance.role.pk,
+                    "name": prev_instance.role.name,
+                }
+            }
+            push_to_timeline(instance.project, instance, "role-changed", extra_data=extra_data)
+
+
+@receiver(signals.post_delete, sender=get_model("projects", "Membership"))
+def delete_membership_push_to_timeline(sender, instance, **kwargs):
+    push_to_timeline(instance.project, instance, "delete")

taiga/timeline/timeline_implementations.py

@@ -0,0 +1,124 @@
+# 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.timeline.service import register_timeline_implementation
+
+
+@register_timeline_implementation("projects.project", "create")
+def project_create_timeline(instance, extra_data={}):
+    return {
+        "project": {
+            "id": instance.pk,
+            "slug": instance.slug,
+            "name": instance.name,
+        },
+        "creator": {
+            "id": instance.owner.pk,
+            "name": instance.owner.get_full_name(),
+        }
+    }
+
+
+@register_timeline_implementation("userstories.userstory", "create")
+def userstory_create_timeline(instance, extra_data={}):
+    return {
+        "userstory": {
+            "id": instance.pk,
+            "subject": instance.subject,
+        },
+        "project": {
+            "id": instance.project.pk,
+            "slug": instance.project.slug,
+            "name": instance.project.name,
+        },
+        "creator": {
+            "id": instance.owner.pk,
+            "name": instance.owner.get_full_name(),
+        }
+    }
+
+
+@register_timeline_implementation("issues.issue", "create")
+def issue_create_timeline(instance, extra_data={}):
+    return {
+        "issue": {
+            "id": instance.pk,
+            "subject": instance.subject,
+        },
+        "project": {
+            "id": instance.project.pk,
+            "slug": instance.project.slug,
+            "name": instance.project.name,
+        },
+        "creator": {
+            "id": instance.owner.pk,
+            "name": instance.owner.get_full_name(),
+        }
+    }
+
+
+@register_timeline_implementation("projects.membership", "create")
+def membership_create_timeline(instance, extra_data={}):
+    return {
+        "user": {
+            "id": instance.user.pk,
+            "name": instance.user.get_full_name(),
+        },
+        "project": {
+            "id": instance.project.pk,
+            "slug": instance.project.slug,
+            "name": instance.project.name,
+        },
+        "role": {
+            "id": instance.role.pk,
+            "name": instance.role.name,
+        }
+    }
+
+
+@register_timeline_implementation("projects.membership", "delete")
+def membership_delete_timeline(instance, extra_data={}):
+    return {
+        "user": {
+            "id": instance.user.pk,
+            "name": instance.user.get_full_name(),
+        },
+        "project": {
+            "id": instance.project.pk,
+            "slug": instance.project.slug,
+            "name": instance.project.name,
+        },
+    }
+
+
+@register_timeline_implementation("projects.membership", "role-changed")
+def membership_role_changed_timeline(instance, extra_data={}):
+    result = {
+        "user": {
+            "id": instance.user.pk,
+            "name": instance.user.get_full_name(),
+        },
+        "project": {
+            "id": instance.project.pk,
+            "slug": instance.project.slug,
+            "name": instance.project.name,
+        },
+        "role": {
+            "id": instance.role.pk,
+            "name": instance.role.name,
+        }
+    }
+    return dict(result.items() + extra_data.items())

tests/integration/test_timeline.py

@@ -0,0 +1,44 @@
+import json
+import pytest
+
+from .. import factories
+
+from taiga.timeline import service
+from taiga.timeline.models import Timeline
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_add_to_object_timeline():
+    Timeline.objects.all().delete()
+    user1 = factories.UserFactory()
+    user2 = factories.UserFactory()
+
+    service.register_timeline_implementation("users.user", "test", lambda x, extra_data=None: str(id(x)))
+
+    service._add_to_object_timeline(user1, user2, "test")
+
+    assert Timeline.objects.filter(object_id=user1.id).count() == 1
+    assert Timeline.objects.order_by("-id")[0].data == id(user2)
+
+def test_get_timeline():
+    Timeline.objects.all().delete()
+
+    user1 = factories.UserFactory()
+    user2 = factories.UserFactory()
+    user3 = factories.UserFactory()
+    user4 = factories.UserFactory()
+
+    service.register_timeline_implementation("users.user", "test", lambda x, extra_data=None: str(id(x)))
+
+    service._add_to_object_timeline(user1, user1, "test")
+    service._add_to_object_timeline(user1, user2, "test")
+    service._add_to_object_timeline(user1, user3, "test")
+    service._add_to_object_timeline(user1, user4, "test")
+
+    service._add_to_object_timeline(user2, user1, "test")
+
+    assert service.get_timeline(user1).count() == 4
+    assert service.get_timeline(user2).count() == 1
+    assert service.get_timeline(user3).count() == 0

tests/unit/test_timeline.py

@@ -0,0 +1,76 @@
+from unittest.mock import patch, MagicMock, call
+
+from django.core.exceptions import ValidationError
+
+from taiga.timeline import service
+from taiga.timeline.models import Timeline
+from taiga.projects.models import Project
+from taiga.users.models import User
+
+import pytest
+
+
+def test_push_to_timeline_many_objects():
+    with patch("taiga.timeline.service._add_to_object_timeline") as mock:
+        users = [User(), User(), User()]
+        project = Project()
+        service.push_to_timeline(users, project, "test")
+        assert mock.call_count == 3
+        assert mock.mock_calls == [
+            call(users[0], project, "test", "default", {}),
+            call(users[1], project, "test", "default", {}),
+            call(users[2], project, "test", "default", {}),
+        ]
+        with pytest.raises(Exception):
+            service.push_to_timeline(None, project, "test")
+
+def test_add_to_objects_timeline():
+    with patch("taiga.timeline.service._add_to_object_timeline") as mock:
+        users = [User(), User(), User()]
+        project = Project()
+        service._add_to_objects_timeline(users, project, "test")
+        assert mock.call_count == 3
+        assert mock.mock_calls == [
+            call(users[0], project, "test", "default", {}),
+            call(users[1], project, "test", "default", {}),
+            call(users[2], project, "test", "default", {}),
+        ]
+        with pytest.raises(Exception):
+            service.push_to_timeline(None, project, "test")
+
+
+def test_modify_created_timeline_entry():
+    timeline = Timeline()
+    timeline.pk = 3
+    with pytest.raises(ValidationError):
+        timeline.save()
+
+
+def test_get_impl_key_from_model():
+    assert service._get_impl_key_from_model(Timeline, "test") == "timeline.timeline.test"
+    with pytest.raises(Exception):
+        service._get_impl_key(None)
+
+
+def test_get_impl_key_from_typename():
+    assert service._get_impl_key_from_typename("timeline.timeline", "test") == "timeline.timeline.test"
+    with pytest.raises(Exception):
+        service._get_impl_key(None)
+
+
+def test_get_class_implementation():
+    service._timeline_impl_map["timeline.timeline.test"] = "test"
+    assert service._get_class_implementation(Timeline, "test") == "test"
+    assert service._get_class_implementation(Timeline, "other") == None
+
+
+def test_register_timeline_implementation():
+    test_func = lambda x: "test-func-result"
+    service.register_timeline_implementation("timeline.timeline", "test", test_func)
+    assert service._timeline_impl_map["timeline.timeline.test"](None) == "test-func-result"
+
+    @service.register_timeline_implementation("timeline.timeline", "test-decorator")
+    def decorated_test_function(x):
+        return "test-decorated-func-result"
+
+    assert service._timeline_impl_map["timeline.timeline.test-decorator"](None) == "test-decorated-func-result"