6786892849

Committed 07 Nov 2023 03:44PM UTC coverage: 99.032%. First build

Build # 6786892849

Build Type

push

github

Committed by

tdetry

Commit Message

[0.1.23] fix accounting/invoices params

Run Details

14 of 16 new or added lines in 3 files covered. (87.5%)

2864 of 2892 relevant lines covered (99.03%)

0.99 hits per line

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

98.86

/tests/test_sync.py

import uuid

from chift.openapi.models import Sync


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",
        "trigger": {"type": trigger_type, "data": {}},
        "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["trigger"]["data"]["cronschedule"] = "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, {})


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, {})


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,
        {},
    )
    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]},
    )

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

chift-oneapi / chift-python-sdk / 6786892849

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