3249283896

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/HewlettPackard/squest/commit/<a class=hub.com/HewlettPackard/squest/commit/9a599511d5892256d376ec8b5c78fbe9f439e9ea">9a599511d<a href="https://github.com/HewlettPackard/squest/commit/9a599511d5892256d376ec8b5c78fbe9f439e9ea">">Merge </a><a class="double-link" href="https://github.com/HewlettPackard/squest/commit/<a class="double-link" href="https://github.com/HewlettPackard/squest/commit/8ffacd8ebdeb9c21bff1c3ea0d088fa59fd68552">8ffacd8eb</a>">8ffacd8eb</a><a href="https://github.com/HewlettPackard/squest/commit/9a599511d5892256d376ec8b5c78fbe9f439e9ea"> into a5101fcc9">a5101fcc9</a>

Pull Request Pull Request #555: [Fix] extra_vars JSON field display str instead of dict

Run Details

19 of 19 new or added lines in 6 files covered. (100.0%)

8071 of 8681 relevant lines covered (92.97%)

0.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.67

/service_catalog/models/state_hooks.py

import logging

from django.db.models import TextChoices, Model, ForeignKey, CASCADE, CharField, JSONField, SET_NULL
from django.utils.translation import gettext_lazy as _
from django.core.exceptions import ValidationError

from service_catalog.models.services import Service
from service_catalog.models.job_templates import JobTemplate
from service_catalog.models.operations import Operation

logger = logging.getLogger(__name__)


class HookModel(TextChoices):
    Request = 'Request', _('Request')
    Instance = 'Instance', _('Instance')


class ServiceStateHook(Model):
    instance = ForeignKey(Service,
                                 on_delete=CASCADE,
                                 related_name='instances',
                                 related_query_name='instance',
                                 null=True)
    model = CharField(max_length=100, choices=HookModel.choices)
    state = CharField(max_length=100)
    job_template = ForeignKey(JobTemplate, on_delete=CASCADE)
    extra_vars = JSONField(default=dict, blank=True)

    def clean(self):
        if self.extra_vars is None or not isinstance(self.extra_vars, dict):
            raise ValidationError({'extra_vars': _("Please enter a valid JSON. Empty value is {} for JSON.")})


class GlobalHook(Model):
    name = CharField(unique=True, max_length=100)
    model = CharField(max_length=100, choices=HookModel.choices)
    state = CharField(max_length=100)
    service = ForeignKey(Service, on_delete=SET_NULL, null=True, blank=True, default=None)
    operation = ForeignKey(Operation, on_delete=SET_NULL, null=True, blank=True, default=None)
    job_template = ForeignKey(JobTemplate, on_delete=CASCADE)
    extra_vars = JSONField(default=dict, blank=True)

    def clean(self):
        if self.operation and not self.service:
            raise ValidationError({'service': _("Service cannot be null if operation is set.")})
        if self.service and self.operation and not self.service.operations.filter(id=self.operation.id).exists():
            raise ValidationError({'operation': _(f"Operation must be in the service({','.join(operation.name for operation in self.service.operations.all())}).")})
        if self.extra_vars is None or not isinstance(self.extra_vars, dict):
            raise ValidationError({'extra_vars': _("Please enter a valid JSON. Empty value is {} for JSON.")})


class HookManager(object):

    @classmethod
    def trigger_hook(cls, sender, instance, name, source, target, *args, **kwargs):
        """
        Method called when Instance or Request change state
        :param sender: Class that call the signal (Instance or Request)
        :param instance: Instance object
        :param name: name of the FSM method
        :param source: source state
        :param target: target state (current)
        :return:
        """
        logger.debug(f"[HookManager] trigger_hook executed with "
                     f"sender model '{sender.__name__}', "
                     f"instance ID'{instance.id}', "
                     f"transition name '{name}', "
                     f"source '{source}', "
                     f"target '{target}'")
        # check if global hooks exist for this object sender model and state
        global_hook_set = GlobalHook.objects.filter(model=sender.__name__, state=target)
        from service_catalog.api.serializers import InstanceReadSerializer, RequestSerializer
        from service_catalog.models import Instance, Request
        if global_hook_set:
            # serialize the instance
            serialized_data = dict()
            service = None
            operations = list()
            if isinstance(instance, Instance):
                service = instance.service
                operations = instance.service.operations.all()
                serialized_data = InstanceReadSerializer(instance).data
            if isinstance(instance, Request):
                service = instance.instance.service
                operations = [instance.operation]
                serialized_data = RequestSerializer(instance).data
            for global_hook in global_hook_set.all():
                if global_hook.service is None or service == global_hook.service:
                    if global_hook.operation is None or global_hook.operation in operations:
                        extra_vars = global_hook.extra_vars
                        extra_vars["squest"] = serialized_data
                        global_hook.job_template.execute(extra_vars=extra_vars)

1	import logging	1✔
2
3	from django.db.models import TextChoices, Model, ForeignKey, CASCADE, CharField, JSONField, SET_NULL	1✔
4	from django.utils.translation import gettext_lazy as _	1✔
5	from django.core.exceptions import ValidationError	1✔
6
7	from service_catalog.models.services import Service	1✔
8	from service_catalog.models.job_templates import JobTemplate	1✔
9	from service_catalog.models.operations import Operation	1✔
10
11	logger = logging.getLogger(__name__)	1✔
12
13
14	class HookModel(TextChoices):	1✔
15	Request = 'Request', _('Request')	1✔
16	Instance = 'Instance', _('Instance')	1✔
17
18
19	class ServiceStateHook(Model):	1✔
20	instance = ForeignKey(Service,	1✔
21	on_delete=CASCADE,
22	related_name='instances',
23	related_query_name='instance',
24	null=True)
25	model = CharField(max_length=100, choices=HookModel.choices)	1✔
26	state = CharField(max_length=100)	1✔
27	job_template = ForeignKey(JobTemplate, on_delete=CASCADE)	1✔
28	extra_vars = JSONField(default=dict, blank=True)	1✔
29
30	def clean(self):	1✔
31	if self.extra_vars is None or not isinstance(self.extra_vars, dict):	×
32	raise ValidationError({'extra_vars': _("Please enter a valid JSON. Empty value is {} for JSON.")})	×
33
34
35	class GlobalHook(Model):	1✔
36	name = CharField(unique=True, max_length=100)	1✔
37	model = CharField(max_length=100, choices=HookModel.choices)	1✔
38	state = CharField(max_length=100)	1✔
39	service = ForeignKey(Service, on_delete=SET_NULL, null=True, blank=True, default=None)	1✔
40	operation = ForeignKey(Operation, on_delete=SET_NULL, null=True, blank=True, default=None)	1✔
41	job_template = ForeignKey(JobTemplate, on_delete=CASCADE)	1✔
42	extra_vars = JSONField(default=dict, blank=True)	1✔
43
44	def clean(self):	1✔
45	if self.operation and not self.service:	1✔
46	raise ValidationError({'service': _("Service cannot be null if operation is set.")})	×
47	if self.service and self.operation and not self.service.operations.filter(id=self.operation.id).exists():	1✔
48	raise ValidationError({'operation': _(f"Operation must be in the service({','.join(operation.name for operation in self.service.operations.all())}).")})	×
49	if self.extra_vars is None or not isinstance(self.extra_vars, dict):	1✔
50	raise ValidationError({'extra_vars': _("Please enter a valid JSON. Empty value is {} for JSON.")})	×
51
52
53	class HookManager(object):	1✔
54
55	@classmethod	1✔
56	def trigger_hook(cls, sender, instance, name, source, target, args, *kwargs):	1✔
57	"""
58	Method called when Instance or Request change state
59	:param sender: Class that call the signal (Instance or Request)
60	:param instance: Instance object
61	:param name: name of the FSM method
62	:param source: source state
63	:param target: target state (current)
64	:return:
65	"""
66	logger.debug(f"[HookManager] trigger_hook executed with "	1✔
67	f"sender model '{sender.__name__}', "
68	f"instance ID'{instance.id}', "
69	f"transition name '{name}', "
70	f"source '{source}', "
71	f"target '{target}'")
72	# check if global hooks exist for this object sender model and state
73	global_hook_set = GlobalHook.objects.filter(model=sender.__name__, state=target)	1✔
74	from service_catalog.api.serializers import InstanceReadSerializer, RequestSerializer	1✔
75	from service_catalog.models import Instance, Request	1✔
76	if global_hook_set:	1✔
77	# serialize the instance
78	serialized_data = dict()	1✔
79	service = None	1✔
80	operations = list()	1✔
81	if isinstance(instance, Instance):	1✔
82	service = instance.service	1✔
83	operations = instance.service.operations.all()	1✔
84	serialized_data = InstanceReadSerializer(instance).data	1✔
85	if isinstance(instance, Request):	1✔
86	service = instance.instance.service	1✔
87	operations = [instance.operation]	1✔
88	serialized_data = RequestSerializer(instance).data	1✔
89	for global_hook in global_hook_set.all():	1✔
90	if global_hook.service is None or service == global_hook.service:	1✔
91	if global_hook.operation is None or global_hook.operation in operations:	1✔
92	extra_vars = global_hook.extra_vars	1✔
93	extra_vars["squest"] = serialized_data	1✔
94	global_hook.job_template.execute(extra_vars=extra_vars)	1✔

HewlettPackard / squest / 3249283896

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous