diff --git a/settings/common.py b/settings/common.py index 98524466..b0e00970 100644 --- a/settings/common.py +++ b/settings/common.py @@ -183,6 +183,7 @@ INSTALLED_APPS = [ "taiga.projects.history", "taiga.projects.notifications", "taiga.projects.votes", + "taiga.timeline", "south", "reversion", diff --git a/taiga/base/utils/db.py b/taiga/base/utils/db.py index 6e056fda..a748c6fd 100644 --- a/taiga/base/utils/db.py +++ b/taiga/base/utils/db.py @@ -1,3 +1,21 @@ +# Copyright (C) 2014 Andrey Antukh +# Copyright (C) 2014 Jesús Espino +# Copyright (C) 2014 David Barragán +# 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 . + +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) diff --git a/taiga/projects/history/services.py b/taiga/projects/history/services.py index 21837ed4..3a61ca5f 100644 --- a/taiga/projects/history/services.py +++ b/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. diff --git a/taiga/routers.py b/taiga/routers.py index 1cb0b2f6..f09b1d55 100644 --- a/taiga/routers.py +++ b/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 diff --git a/taiga/timeline/__init__.py b/taiga/timeline/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/taiga/timeline/api.py b/taiga/timeline/api.py new file mode 100644 index 00000000..96f5f7b0 --- /dev/null +++ b/taiga/timeline/api.py @@ -0,0 +1,73 @@ +# Copyright (C) 2014 Andrey Antukh +# Copyright (C) 2014 Jesús Espino +# Copyright (C) 2014 David Barragán +# 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 . + +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" diff --git a/taiga/timeline/migrations/0001_initial.py b/taiga/timeline/migrations/0001_initial.py new file mode 100644 index 00000000..7003e016 --- /dev/null +++ b/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'] \ No newline at end of file diff --git a/taiga/timeline/migrations/__init__.py b/taiga/timeline/migrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/taiga/timeline/models.py b/taiga/timeline/models.py new file mode 100644 index 00000000..7aff7217 --- /dev/null +++ b/taiga/timeline/models.py @@ -0,0 +1,48 @@ +# Copyright (C) 2014 Andrey Antukh +# Copyright (C) 2014 Jesús Espino +# Copyright (C) 2014 David Barragán +# 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 . + +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 * diff --git a/taiga/timeline/serializers.py b/taiga/timeline/serializers.py new file mode 100644 index 00000000..69c91ad1 --- /dev/null +++ b/taiga/timeline/serializers.py @@ -0,0 +1,27 @@ +# Copyright (C) 2014 Andrey Antukh +# Copyright (C) 2014 Jesús Espino +# Copyright (C) 2014 David Barragán +# 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 . + +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 diff --git a/taiga/timeline/service.py b/taiga/timeline/service.py new file mode 100644 index 00000000..4eb9a339 --- /dev/null +++ b/taiga/timeline/service.py @@ -0,0 +1,95 @@ +# Copyright (C) 2014 Andrey Antukh +# Copyright (C) 2014 Jesús Espino +# Copyright (C) 2014 David Barragán +# 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 . + +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 diff --git a/taiga/timeline/signals.py b/taiga/timeline/signals.py new file mode 100644 index 00000000..ead869d0 --- /dev/null +++ b/taiga/timeline/signals.py @@ -0,0 +1,65 @@ +# Copyright (C) 2014 Andrey Antukh +# Copyright (C) 2014 Jesús Espino +# Copyright (C) 2014 David Barragán +# 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 . + +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") diff --git a/taiga/timeline/timeline_implementations.py b/taiga/timeline/timeline_implementations.py new file mode 100644 index 00000000..d21a3e49 --- /dev/null +++ b/taiga/timeline/timeline_implementations.py @@ -0,0 +1,124 @@ +# Copyright (C) 2014 Andrey Antukh +# Copyright (C) 2014 Jesús Espino +# Copyright (C) 2014 David Barragán +# 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 . + +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()) diff --git a/tests/integration/test_timeline.py b/tests/integration/test_timeline.py new file mode 100644 index 00000000..d028e9df --- /dev/null +++ b/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 diff --git a/tests/unit/test_timeline.py b/tests/unit/test_timeline.py new file mode 100644 index 00000000..f09bd8ab --- /dev/null +++ b/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"