25400571889

Committed 05 May 2026 08:29PM UTC coverage: 90.612% (-0.4%) from 90.986%

Build # 25400571889

Build Type

Pull #643

github

Committed by

web-flow

Commit Message

Merge 4e812ec20 into 15d676845

Pull Request Pull Request #643: Adding new backfill option

Coverage Stats

8291 of 9150 relevant lines covered (90.61%)

0.91 hits per line

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

98.0

back/admin/integrations/serializers.py

from django.core.exceptions import ValidationError
from django_q.models import validate_cron
from rest_framework import serializers


# Credit: https://stackoverflow.com/a/42432240
class ValidateMixin:
    def validate(self, data):
        if hasattr(self, "initial_data"):
            unknown_keys = set(self.initial_data.keys()) - set(self.fields.keys())
            if unknown_keys:
                raise ValidationError("Got unknown fields: {}".format(unknown_keys))
        return data


class ManifestFormSerializer(ValidateMixin, serializers.Serializer):
    id = serializers.CharField()
    url = serializers.CharField(required=False)
    name = serializers.CharField()
    type = serializers.ChoiceField(
        [
            ("multiple_choice", "Multiple choice"),
            ("choice", "Choice"),
            ("input", "Input field"),
            ("generate", "Generate random text"),
        ]
    )
    items = serializers.JSONField(required=False)
    data_from = serializers.CharField(required=False)
    choice_value = serializers.CharField(required=False)
    choice_name = serializers.CharField(required=False)
    options_source = serializers.ChoiceField(
        [
            ("fixed list", "fixed list"),
            ("fetch url", "fetch url"),
        ],
        required=False,
    )


class ManifestConditionSerializer(ValidateMixin, serializers.Serializer):
    response_notation = serializers.CharField()
    value = serializers.CharField()


class ManifestPollingSerializer(ValidateMixin, serializers.Serializer):
    interval = serializers.IntegerField(min_value=1, max_value=3600)
    amount = serializers.IntegerField(min_value=1, max_value=100)


class ManifestExistSerializer(ValidateMixin, serializers.Serializer):
    url = serializers.CharField()
    expected = serializers.CharField()
    store_data = serializers.DictField(child=serializers.CharField(), default=dict)
    status_code = serializers.ListField(
        child=serializers.IntegerField(), required=False
    )
    extra_headers = serializers.ChoiceField(
        [
            ("pritunl", "pritunl"),
        ],
        required=False,
    )
    method = serializers.ChoiceField(
        [
            ("GET", "GET"),
            ("POST", "POST"),
            ("PUT", "PUT"),
        ]
    )


class ManifestExecuteSerializer(ValidateMixin, serializers.Serializer):
    url = serializers.CharField()
    data = serializers.JSONField(required=False, default=dict)
    headers = serializers.DictField(child=serializers.CharField(), default=dict)
    store_data = serializers.DictField(child=serializers.CharField(), default=dict)
    method = serializers.ChoiceField(
        [
            ("HEAD", "HEAD"),
            ("GET", "GET"),
            ("POST", "POST"),
            ("PUT", "PUT"),
            ("PATCH", "PATCH"),
            ("DELETE", "DELETE"),
        ]
    )
    extra_headers = serializers.ChoiceField(
        [
            ("pritunl", "pritunl"),
        ],
        required=False,
    )
    status_code = serializers.ListField(
        child=serializers.IntegerField(), required=False
    )
    files = serializers.DictField(child=serializers.CharField(), default=dict)
    save_as_file = serializers.CharField(required=False)
    polling = ManifestPollingSerializer(required=False)
    continue_if = ManifestConditionSerializer(required=False)

    def validate(self, data):
        # Check that if polling has been filled, that continue_if is also filled
        polling = data.get("polling", False)
        continue_if = data.get("continue_if", False)
        if polling and not continue_if:
            raise serializers.ValidationError(
                "continue_if must be filled if you use polling"
            )
        return data


class ManifestRevokeSerializer(ValidateMixin, serializers.Serializer):
    url = serializers.CharField()
    data = serializers.JSONField(required=False, default=dict)
    status_code = serializers.ListField(
        child=serializers.IntegerField(), required=False
    )
    extra_headers = serializers.ChoiceField(
        [
            ("pritunl", "pritunl"),
        ],
        required=False,
    )
    headers = serializers.DictField(child=serializers.CharField(), default=dict)
    method = serializers.ChoiceField(
        [
            ("HEAD", "HEAD"),
            ("GET", "GET"),
            ("POST", "POST"),
            ("PUT", "PUT"),
            ("PATCH", "PATCH"),
            ("DELETE", "DELETE"),
        ]
    )


class ManifestPostExecuteNotificationSerializer(ValidateMixin, serializers.Serializer):
    type = serializers.ChoiceField(
        [
            ("email", "Email message"),
            ("text", "Text message"),
        ]
    )
    to = serializers.CharField()
    subject = serializers.CharField()
    message = serializers.CharField()


class ManifestInitialDataFormSerializer(ValidateMixin, serializers.Serializer):
    id = serializers.CharField()
    name = serializers.CharField()
    secret = serializers.BooleanField(default=True)
    description = serializers.CharField()


class ManifestExtraUserInfoFormSerializer(ValidateMixin, serializers.Serializer):
    id = serializers.CharField()
    name = serializers.CharField()
    description = serializers.CharField()


class ManifestOauthSerializer(ValidateMixin, serializers.Serializer):
    authenticate_url = serializers.CharField()
    access_token = ManifestExecuteSerializer(required=True)
    refresh = ManifestExecuteSerializer(required=True)
    without_code = serializers.BooleanField(required=False)


class WebhookManifestSerializer(ValidateMixin, serializers.Serializer):
    form = ManifestFormSerializer(required=False, many=True)
    exists = ManifestExistSerializer(required=False)
    revoke = ManifestRevokeSerializer(required=False, many=True)
    execute = ManifestExecuteSerializer(many=True)
    post_execute_notification = ManifestPostExecuteNotificationSerializer(
        many=True, required=False
    )
    initial_data_form = ManifestInitialDataFormSerializer(many=True, required=False)
    extra_user_info = ManifestExtraUserInfoFormSerializer(many=True, required=False)
    headers = serializers.DictField(child=serializers.CharField(), default=dict)
    oauth = ManifestOauthSerializer(required=False)


class SyncUsersManifestSerializer(ValidateMixin, serializers.Serializer):
    data_from = serializers.CharField(required=False)
    data_structure = serializers.DictField(child=serializers.CharField())
    execute = ManifestExecuteSerializer(many=True)
    next_page_token_from = serializers.CharField(required=False)
    next_page = serializers.CharField(required=False)
    next_page_from = serializers.CharField(required=False)
    schedule = serializers.CharField(required=False, validators=[validate_cron])
    action = serializers.ChoiceField([("create", "create"), ("update", "update")])
    amount_pages_to_fetch = serializers.IntegerField(required=False)
    initial_data_form = ManifestInitialDataFormSerializer(many=True, required=False)
    headers = serializers.DictField(child=serializers.CharField(), default=dict)
    oauth = ManifestOauthSerializer(required=False)

1	from django.core.exceptions import ValidationError	1✔
2	from django_q.models import validate_cron	1✔
3	from rest_framework import serializers	1✔
4
5
6	# Credit: https://stackoverflow.com/a/42432240
7	class ValidateMixin:	1✔
8	def validate(self, data):	1✔
9	if hasattr(self, "initial_data"):	1✔
10	unknown_keys = set(self.initial_data.keys()) - set(self.fields.keys())	1✔
11	if unknown_keys:	1✔
12	raise ValidationError("Got unknown fields: {}".format(unknown_keys))	×
13	return data	1✔
14
15
16	class ManifestFormSerializer(ValidateMixin, serializers.Serializer):	1✔
17	id = serializers.CharField()	1✔
18	url = serializers.CharField(required=False)	1✔
19	name = serializers.CharField()	1✔
20	type = serializers.ChoiceField(	1✔
21	[
22	("multiple_choice", "Multiple choice"),
23	("choice", "Choice"),
24	("input", "Input field"),
25	("generate", "Generate random text"),
26	]
27	)
28	items = serializers.JSONField(required=False)	1✔
29	data_from = serializers.CharField(required=False)	1✔
30	choice_value = serializers.CharField(required=False)	1✔
31	choice_name = serializers.CharField(required=False)	1✔
32	options_source = serializers.ChoiceField(	1✔
33	[
34	("fixed list", "fixed list"),
35	("fetch url", "fetch url"),
36	],
37	required=False,
38	)
39
40
41	class ManifestConditionSerializer(ValidateMixin, serializers.Serializer):	1✔
42	response_notation = serializers.CharField()	1✔
43	value = serializers.CharField()	1✔
44
45
46	class ManifestPollingSerializer(ValidateMixin, serializers.Serializer):	1✔
47	interval = serializers.IntegerField(min_value=1, max_value=3600)	1✔
48	amount = serializers.IntegerField(min_value=1, max_value=100)	1✔
49
50
51	class ManifestExistSerializer(ValidateMixin, serializers.Serializer):	1✔
52	url = serializers.CharField()	1✔
53	expected = serializers.CharField()	1✔
54	store_data = serializers.DictField(child=serializers.CharField(), default=dict)	1✔
55	status_code = serializers.ListField(	1✔
56	child=serializers.IntegerField(), required=False
57	)
58	extra_headers = serializers.ChoiceField(	1✔
59	[
60	("pritunl", "pritunl"),
61	],
62	required=False,
63	)
64	method = serializers.ChoiceField(	1✔
65	[
66	("GET", "GET"),
67	("POST", "POST"),
68	("PUT", "PUT"),
69	]
70	)
71
72
73	class ManifestExecuteSerializer(ValidateMixin, serializers.Serializer):	1✔
74	url = serializers.CharField()	1✔
75	data = serializers.JSONField(required=False, default=dict)	1✔
76	headers = serializers.DictField(child=serializers.CharField(), default=dict)	1✔
77	store_data = serializers.DictField(child=serializers.CharField(), default=dict)	1✔
78	method = serializers.ChoiceField(	1✔
79	[
80	("HEAD", "HEAD"),
81	("GET", "GET"),
82	("POST", "POST"),
83	("PUT", "PUT"),
84	("PATCH", "PATCH"),
85	("DELETE", "DELETE"),
86	]
87	)
88	extra_headers = serializers.ChoiceField(	1✔
89	[
90	("pritunl", "pritunl"),
91	],
92	required=False,
93	)
94	status_code = serializers.ListField(	1✔
95	child=serializers.IntegerField(), required=False
96	)
97	files = serializers.DictField(child=serializers.CharField(), default=dict)	1✔
98	save_as_file = serializers.CharField(required=False)	1✔
99	polling = ManifestPollingSerializer(required=False)	1✔
100	continue_if = ManifestConditionSerializer(required=False)	1✔
101
102	def validate(self, data):	1✔
103	# Check that if polling has been filled, that continue_if is also filled
104	polling = data.get("polling", False)	1✔
105	continue_if = data.get("continue_if", False)	1✔
106	if polling and not continue_if:	1✔
107	raise serializers.ValidationError(	×
108	"continue_if must be filled if you use polling"
109	)
110	return data	1✔
111
112
113	class ManifestRevokeSerializer(ValidateMixin, serializers.Serializer):	1✔
114	url = serializers.CharField()	1✔
115	data = serializers.JSONField(required=False, default=dict)	1✔
116	status_code = serializers.ListField(	1✔
117	child=serializers.IntegerField(), required=False
118	)
119	extra_headers = serializers.ChoiceField(	1✔
120	[
121	("pritunl", "pritunl"),
122	],
123	required=False,
124	)
125	headers = serializers.DictField(child=serializers.CharField(), default=dict)	1✔
126	method = serializers.ChoiceField(	1✔
127	[
128	("HEAD", "HEAD"),
129	("GET", "GET"),
130	("POST", "POST"),
131	("PUT", "PUT"),
132	("PATCH", "PATCH"),
133	("DELETE", "DELETE"),
134	]
135	)
136
137
138	class ManifestPostExecuteNotificationSerializer(ValidateMixin, serializers.Serializer):	1✔
139	type = serializers.ChoiceField(	1✔
140	[
141	("email", "Email message"),
142	("text", "Text message"),
143	]
144	)
145	to = serializers.CharField()	1✔
146	subject = serializers.CharField()	1✔
147	message = serializers.CharField()	1✔
148
149
150	class ManifestInitialDataFormSerializer(ValidateMixin, serializers.Serializer):	1✔
151	id = serializers.CharField()	1✔
152	name = serializers.CharField()	1✔
153	secret = serializers.BooleanField(default=True)	1✔
154	description = serializers.CharField()	1✔
155
156
157	class ManifestExtraUserInfoFormSerializer(ValidateMixin, serializers.Serializer):	1✔
158	id = serializers.CharField()	1✔
159	name = serializers.CharField()	1✔
160	description = serializers.CharField()	1✔
161
162
163	class ManifestOauthSerializer(ValidateMixin, serializers.Serializer):	1✔
164	authenticate_url = serializers.CharField()	1✔
165	access_token = ManifestExecuteSerializer(required=True)	1✔
166	refresh = ManifestExecuteSerializer(required=True)	1✔
167	without_code = serializers.BooleanField(required=False)	1✔
168
169
170	class WebhookManifestSerializer(ValidateMixin, serializers.Serializer):	1✔
171	form = ManifestFormSerializer(required=False, many=True)	1✔
172	exists = ManifestExistSerializer(required=False)	1✔
173	revoke = ManifestRevokeSerializer(required=False, many=True)	1✔
174	execute = ManifestExecuteSerializer(many=True)	1✔
175	post_execute_notification = ManifestPostExecuteNotificationSerializer(	1✔
176	many=True, required=False
177	)
178	initial_data_form = ManifestInitialDataFormSerializer(many=True, required=False)	1✔
179	extra_user_info = ManifestExtraUserInfoFormSerializer(many=True, required=False)	1✔
180	headers = serializers.DictField(child=serializers.CharField(), default=dict)	1✔
181	oauth = ManifestOauthSerializer(required=False)	1✔
182
183
184	class SyncUsersManifestSerializer(ValidateMixin, serializers.Serializer):	1✔
185	data_from = serializers.CharField(required=False)	1✔
186	data_structure = serializers.DictField(child=serializers.CharField())	1✔
187	execute = ManifestExecuteSerializer(many=True)	1✔
188	next_page_token_from = serializers.CharField(required=False)	1✔
189	next_page = serializers.CharField(required=False)	1✔
190	next_page_from = serializers.CharField(required=False)	1✔
191	schedule = serializers.CharField(required=False, validators=[validate_cron])	1✔
192	action = serializers.ChoiceField([("create", "create"), ("update", "update")])	1✔
193	amount_pages_to_fetch = serializers.IntegerField(required=False)	1✔
194	initial_data_form = ManifestInitialDataFormSerializer(many=True, required=False)	1✔
195	headers = serializers.DictField(child=serializers.CharField(), default=dict)	1✔
196	oauth = ManifestOauthSerializer(required=False)	1✔

chiefonboarding / ChiefOnboarding / 25400571889

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