7666915815

Committed 26 Jan 2024 10:18AM UTC coverage: 98.637%. First build

Build # 7666915815

Build Type

push

github

Committed by

hhertoghe

Commit Message

reformat

Run Details

3112 of 3155 relevant lines covered (98.64%)

0.99 hits per line

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

97.78

/tests/test_sync.py

import uuid
from datetime import datetime

from chift.openapi.models import Sync

today_date = datetime.now().strftime("%d/%m/%Y")


def _get_flow_data(
    flow_name=None, datastore_name=None, trigger_type="event", execution_type="code"
):
    if not flow_name:
        flow_name = str(uuid.uuid1())
    if not datastore_name:
        datastore_name = str(uuid.uuid1())

    data = {
        "name": flow_name,
        "description": "test",
        "triggers": [{"id": "trigger-1", "type": trigger_type}],
        "execution": {
            "type": execution_type,
            "data": {},
        },
        "config": {
            "datastores": [
                {
                    "name": "INVOICES",
                    "definition": {
                        "columns": [
                            {
                                "name": "source_id",
                                "type": "text",
                                "title": "Invoice source id",
                            },
                            {
                                "name": "target_id",
                                "type": "text",
                                "title": "Invoice target id",
                            },
                            {
                                "name": "reference",
                                "type": "text",
                                "title": "Invoice code/ref",
                            },
                        ],
                        "search_column": "source_id",
                    },
                },
                {
                    "name": "CONTACTS",
                    "definition": {
                        "columns": [
                            {
                                "name": "source_id",
                                "type": "text",
                                "title": "Contact source id",
                            },
                            {
                                "name": "target_id",
                                "type": "text",
                                "title": "Contact target id",
                            },
                        ],
                        "search_column": "source_id",
                    },
                },
                {
                    "name": "PRODUCTS",
                    "definition": {
                        "columns": [
                            {
                                "name": "source_id",
                                "type": "text",
                                "title": "Product source id",
                            },
                            {
                                "name": "target_id",
                                "type": "text",
                                "title": "Product target id",
                            },
                            {
                                "name": "reference",
                                "type": "text",
                                "title": "Product code/ref",
                            },
                        ],
                        "search_column": "source_id",
                    },
                },
                {
                    "name": "ERRORS",
                    "definition": {
                        "columns": [
                            {
                                "name": "message",
                                "type": "text",
                                "title": "Error message",
                            },
                            {
                                "name": "source_id",
                                "type": "text",
                                "title": "Invoice source id",
                            },
                        ]
                    },
                },
            ],
            "definitionFields": [
                {
                    "type": "date",
                    "title": "Début de la synchronisation? (dd/mm/yyyy)",
                    "name": "from_date",
                    "optional": False,
                },
                {
                    "type": "text",
                    "title": "Valeur par défault pour le pays (ISO2)",
                    "name": "default_country",
                    "default": "FR",
                    "optional": True,
                },
            ],
            "customFields": [
                {
                    "type": "source_node_id",
                    "value": 7012,
                },
                {
                    "type": "target_node_id",
                    "value": 7010,
                },
                {
                    "type": "tax_mapping_name",
                    "value": "Tax Rates Mapping",
                },
            ],
        },
    }

    if trigger_type == "timer":
        data["triggers"][0]["cronschedules"] = ["0/30 * * * *"]

    if execution_type == "code":
        data["execution"]["data"]["code"] = "console.log('This is fucking awesome!')"

    if execution_type == "module":
        data["execution"]["data"]["name"] = "Invoice to invoice"

    return data


def get_sync(chift):
    # TODO: improve this once sync creation endpoint is there
    syncs = chift.Sync.all()
    return syncs[0]


def test_sync(chift):
    syncs = chift.Sync.all()

    assert syncs

    expected_sync = syncs[0]

    actual_sync = chift.Sync.get(expected_sync.syncid)

    assert expected_sync.name == actual_sync.name


def test_flow_update(chift):
    syncs = chift.Sync.all()

    sync: Sync = syncs[0]

    # create flow
    datastore_name = str(uuid.uuid1())
    data = _get_flow_data(datastore_name=datastore_name)
    flow_created = chift.Flow.create(sync.syncid, data)

    assert flow_created.id in [flow.id for flow in chift.Sync.get(sync.syncid).flows]

    # update flow
    data["description"] = "test updated"
    data["config"]["datastores"] = [
        {
            "name": datastore_name,
            "definition": {
                "columns": [{"name": "test_x", "title": "title", "type": "text"}]
            },
        },
        {
            "name": str(uuid.uuid1()),
            "definition": {
                "columns": [{"name": "test_y", "title": "title", "type": "text"}]
            },
        },
    ]
    flow_updated = chift.Flow.create(sync.syncid, data)

    assert flow_updated.id == flow_created.id

    for flow in chift.Sync.get(sync.syncid).flows:
        if flow.id == flow_created.id:
            assert flow.description == "test updated"
            assert len(flow.config.datastores) == 2

    chift.Flow.delete(sync.syncid, flow_created.id)


def test_flow_create_trigger_code(chift):
    syncs = chift.Sync.all()

    sync: Sync = syncs[0]

    # create flow
    flow_created = chift.Flow.create(sync.syncid, _get_flow_data(execution_type="code"))

    # test trigger flow of type code (AWS lambda) should not raise
    chift.Flow.trigger(
        sync.syncid, flow_created.id, {"data": {"from_date": today_date}}
    )


def test_flow_create_trigger_module(chift):
    syncs = chift.Sync.all()
    sync: Sync = syncs[0]

    # create flow
    flow_created = chift.Flow.create(
        sync.syncid, _get_flow_data(execution_type="module")
    )

    # test trigger flow of type chain should not raise
    chift.Flow.trigger(
        sync.syncid, flow_created.id, {"data": {"from_date": today_date}}
    )


def test_flow_create_trigger_timer(chift):
    syncs = chift.Sync.all()

    sync: Sync = syncs[0]

    # create flow with timer should not raise
    flow_created = chift.Flow.create(sync.syncid, _get_flow_data(trigger_type="timer"))

    assert flow_created


def test_create_flow(chift):
    sync = get_sync(chift)

    datastore_name = str(uuid.uuid1())
    data = _get_flow_data(
        flow_name="Migration de factures",
        datastore_name=datastore_name,
        execution_type="module",
    )
    flow_created = chift.Flow.create(sync.syncid, data)

    assert flow_created.id in [flow.id for flow in chift.Sync.get(sync.syncid).flows]

    # let's trigger it
    res = chift.Flow.trigger(
        sync.syncid,
        flow_created.id,
        {"data": {"from_date": today_date}},
    )
    status = chift.Flow.chainexecution(
        sync.syncid, flow_created.id, res["data"]["executionid"]
    )
    assert status.get("status")

    # let's trigger it for one consumer
    chift.Flow.trigger(
        sync.syncid,
        flow_created.id,
        {"consumers": sync.consumers[:1], "data": {"from_date": today_date}},
    )

    # test getting sync for consumer info
    consumer = chift.Consumer.get(sync.consumers[0])
    info = consumer.Sync.get(sync.syncid)
    assert info


def test_datastore(chift):
    sync = get_sync(chift)

    consumer = chift.Consumer.get(sync.consumers[0])
    datastore = sync.flows[0].config.datastores[0]

    # test create
    value = {}

    str_column_name = None

    for column in datastore.definition.columns:
        if column.type == "json":
            value[column.name] = {"1": "1"}
        elif column.type == "number":
            value[column.name] = 1
        else:
            str_column_name = column.name
            value[column.name] = "1"

    data = consumer.Data.create(datastore.id, [{"data": value}])[0]

    # test update
    updated_value = data.data
    if str_column_name in updated_value:
        updated_value[str_column_name] = "2"

    updated_data = consumer.Data.update(datastore.id, data.id, {"data": updated_value})

    if str_column_name in updated_data.data:
        assert updated_data.data[str_column_name] == "2"

1	import uuid	1✔
2	from datetime import datetime	1✔
3
4	from chift.openapi.models import Sync	1✔
5
6	today_date = datetime.now().strftime("%d/%m/%Y")	1✔
7
8
9	def _get_flow_data(	1✔
10	flow_name=None, datastore_name=None, trigger_type="event", execution_type="code"
11	):
12	if not flow_name:	1✔
13	flow_name = str(uuid.uuid1())	1✔
14	if not datastore_name:	1✔
15	datastore_name = str(uuid.uuid1())	1✔
16
17	data = {	1✔
18	"name": flow_name,
19	"description": "test",
20	"triggers": [{"id": "trigger-1", "type": trigger_type}],
21	"execution": {
22	"type": execution_type,
23	"data": {},
24	},
25	"config": {
26	"datastores": [
27	{
28	"name": "INVOICES",
29	"definition": {
30	"columns": [
31	{
32	"name": "source_id",
33	"type": "text",
34	"title": "Invoice source id",
35	},
36	{
37	"name": "target_id",
38	"type": "text",
39	"title": "Invoice target id",
40	},
41	{
42	"name": "reference",
43	"type": "text",
44	"title": "Invoice code/ref",
45	},
46	],
47	"search_column": "source_id",
48	},
49	},
50	{
51	"name": "CONTACTS",
52	"definition": {
53	"columns": [
54	{
55	"name": "source_id",
56	"type": "text",
57	"title": "Contact source id",
58	},
59	{
60	"name": "target_id",
61	"type": "text",
62	"title": "Contact target id",
63	},
64	],
65	"search_column": "source_id",
66	},
67	},
68	{
69	"name": "PRODUCTS",
70	"definition": {
71	"columns": [
72	{
73	"name": "source_id",
74	"type": "text",
75	"title": "Product source id",
76	},
77	{
78	"name": "target_id",
79	"type": "text",
80	"title": "Product target id",
81	},
82	{
83	"name": "reference",
84	"type": "text",
85	"title": "Product code/ref",
86	},
87	],
88	"search_column": "source_id",
89	},
90	},
91	{
92	"name": "ERRORS",
93	"definition": {
94	"columns": [
95	{
96	"name": "message",
97	"type": "text",
98	"title": "Error message",
99	},
100	{
101	"name": "source_id",
102	"type": "text",
103	"title": "Invoice source id",
104	},
105	]
106	},
107	},
108	],
109	"definitionFields": [
110	{
111	"type": "date",
112	"title": "Début de la synchronisation? (dd/mm/yyyy)",
113	"name": "from_date",
114	"optional": False,
115	},
116	{
117	"type": "text",
118	"title": "Valeur par défault pour le pays (ISO2)",
119	"name": "default_country",
120	"default": "FR",
121	"optional": True,
122	},
123	],
124	"customFields": [
125	{
126	"type": "source_node_id",
127	"value": 7012,
128	},
129	{
130	"type": "target_node_id",
131	"value": 7010,
132	},
133	{
134	"type": "tax_mapping_name",
135	"value": "Tax Rates Mapping",
136	},
137	],
138	},
139	}
140
141	if trigger_type == "timer":	1✔
142	data["triggers"][0]["cronschedules"] = ["0/30 * * * *"]	1✔
143
144	if execution_type == "code":	1✔
145	data["execution"]["data"]["code"] = "console.log('This is fucking awesome!')"	1✔
146
147	if execution_type == "module":	1✔
148	data["execution"]["data"]["name"] = "Invoice to invoice"	1✔
149
150	return data	1✔
151
152
153	def get_sync(chift):	1✔
154	# TODO: improve this once sync creation endpoint is there
155	syncs = chift.Sync.all()	1✔
156	return syncs[0]	1✔
157
158
159	def test_sync(chift):	1✔
160	syncs = chift.Sync.all()	1✔
161
162	assert syncs	1✔
163
164	expected_sync = syncs[0]	1✔
165
166	actual_sync = chift.Sync.get(expected_sync.syncid)	1✔
167
168	assert expected_sync.name == actual_sync.name	1✔
169
170
171	def test_flow_update(chift):	1✔
172	syncs = chift.Sync.all()	1✔
173
174	sync: Sync = syncs[0]	1✔
175
176	# create flow
177	datastore_name = str(uuid.uuid1())	1✔
178	data = _get_flow_data(datastore_name=datastore_name)	1✔
179	flow_created = chift.Flow.create(sync.syncid, data)	1✔
180
181	assert flow_created.id in [flow.id for flow in chift.Sync.get(sync.syncid).flows]	1✔
182
183	# update flow
184	data["description"] = "test updated"	1✔
185	data["config"]["datastores"] = [	1✔
186	{
187	"name": datastore_name,
188	"definition": {
189	"columns": [{"name": "test_x", "title": "title", "type": "text"}]
190	},
191	},
192	{
193	"name": str(uuid.uuid1()),
194	"definition": {
195	"columns": [{"name": "test_y", "title": "title", "type": "text"}]
196	},
197	},
198	]
199	flow_updated = chift.Flow.create(sync.syncid, data)	1✔
200
201	assert flow_updated.id == flow_created.id	1✔
202
203	for flow in chift.Sync.get(sync.syncid).flows:	1✔
204	if flow.id == flow_created.id:	1✔
205	assert flow.description == "test updated"	1✔
206	assert len(flow.config.datastores) == 2	1✔
207
208	chift.Flow.delete(sync.syncid, flow_created.id)	1✔
209
210
211	def test_flow_create_trigger_code(chift):	1✔
212	syncs = chift.Sync.all()	1✔
213
214	sync: Sync = syncs[0]	1✔
215
216	# create flow
217	flow_created = chift.Flow.create(sync.syncid, _get_flow_data(execution_type="code"))	1✔
218
219	# test trigger flow of type code (AWS lambda) should not raise
220	chift.Flow.trigger(	1✔
221	sync.syncid, flow_created.id, {"data": {"from_date": today_date}}
222	)
223
224
225	def test_flow_create_trigger_module(chift):	1✔
226	syncs = chift.Sync.all()	1✔
227	sync: Sync = syncs[0]	1✔
228
229	# create flow
230	flow_created = chift.Flow.create(	1✔
231	sync.syncid, _get_flow_data(execution_type="module")
232	)
233
234	# test trigger flow of type chain should not raise
235	chift.Flow.trigger(	1✔
236	sync.syncid, flow_created.id, {"data": {"from_date": today_date}}
237	)
238
239
240	def test_flow_create_trigger_timer(chift):	1✔
241	syncs = chift.Sync.all()	1✔
242
243	sync: Sync = syncs[0]	1✔
244
245	# create flow with timer should not raise
246	flow_created = chift.Flow.create(sync.syncid, _get_flow_data(trigger_type="timer"))	1✔
247
248	assert flow_created	1✔
249
250
251	def test_create_flow(chift):	1✔
252	sync = get_sync(chift)	1✔
253
254	datastore_name = str(uuid.uuid1())	1✔
255	data = _get_flow_data(	1✔
256	flow_name="Migration de factures",
257	datastore_name=datastore_name,
258	execution_type="module",
259	)
260	flow_created = chift.Flow.create(sync.syncid, data)	1✔
261
262	assert flow_created.id in [flow.id for flow in chift.Sync.get(sync.syncid).flows]	1✔
263
264	# let's trigger it
265	res = chift.Flow.trigger(	1✔
266	sync.syncid,
267	flow_created.id,
268	{"data": {"from_date": today_date}},
269	)
270	status = chift.Flow.chainexecution(	1✔
271	sync.syncid, flow_created.id, res["data"]["executionid"]
272	)
273	assert status.get("status")	1✔
274
275	# let's trigger it for one consumer
276	chift.Flow.trigger(	1✔
277	sync.syncid,
278	flow_created.id,
279	{"consumers": sync.consumers[:1], "data": {"from_date": today_date}},
280	)
281
282	# test getting sync for consumer info
283	consumer = chift.Consumer.get(sync.consumers[0])	1✔
284	info = consumer.Sync.get(sync.syncid)	1✔
285	assert info	1✔
286
287
288	def test_datastore(chift):	1✔
289	sync = get_sync(chift)	1✔
290
291	consumer = chift.Consumer.get(sync.consumers[0])	1✔
292	datastore = sync.flows[0].config.datastores[0]	1✔
293
294	# test create
295	value = {}	1✔
296
297	str_column_name = None	1✔
298
299	for column in datastore.definition.columns:	1✔
300	if column.type == "json":	1✔
301	value[column.name] = {"1": "1"}	×
302	elif column.type == "number":	1✔
303	value[column.name] = 1	×
304	else:
305	str_column_name = column.name	1✔
306	value[column.name] = "1"	1✔
307
308	data = consumer.Data.create(datastore.id, [{"data": value}])[0]	1✔
309
310	# test update
311	updated_value = data.data	1✔
312	if str_column_name in updated_value:	1✔
313	updated_value[str_column_name] = "2"	1✔
314
315	updated_data = consumer.Data.update(datastore.id, data.id, {"data": updated_value})	1✔
316
317	if str_column_name in updated_data.data:	1✔
318	assert updated_data.data[str_column_name] == "2"	1✔

chift-oneapi / chift-python-sdk / 7666915815

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