las-cli/tests/test_workflow_executions.py at a6eb3c74d1c47814579685ae16d25928ba86301c · LucidtechAI/las-cli · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
import pytest
from tests import service, util


def test_executions_create(parser, client):
    args = [
        'workflows',
        'execute',
        service.create_workflow_id(),
        str(util.metadata_path()),
    ]
    util.main_parser(parser, client, args)


def test_executions_create_all(parser, client):
    args = [
        'workflows',
        'execute-all',
        service.create_workflow_id(),
        service.create_dataset_id(),
    ]
    util.main_parser(parser, client, args)


@pytest.mark.parametrize('sort_by', [
    ('--sort-by', 'startTime'),
    ('--sort-by', 'endTime'),
    (),
])
@pytest.mark.parametrize('status', [
    ('--status', 'succeeded'),
    ('--status', 'failed'),
    ('--status', 'rejected'),
    ('--status', 'running'),
    ('--status', 'retry'),
    ('--status', 'error'),
    ('--status', 'error', 'retry'),
    (),
])
@pytest.mark.parametrize('order', [
    ('--order', 'ascending'),
    ('--order', 'descending'),
    (),
])
@pytest.mark.parametrize('from_start_time', [
    ('--from-start-time', '2022-01-01'),
    ('--from-start-time', '3d'),
    (),
])
@pytest.mark.parametrize('to_start_time', [
    ('--to-start-time', '2022-01-01'),
    ('--to-start-time', '3d'),
    (),
])
def test_executions_list(parser, client, list_defaults, sort_by, order, status, from_start_time, to_start_time):
    args = [
        'workflows',
        'list-executions',
        service.create_workflow_id(),
        *list_defaults,
        *sort_by,
        *order,
        *from_start_time,
        *to_start_time,
    ]
    util.main_parser(parser, client, args)


def test_executions_get(parser, client):
    args = [
        'workflows',
        'get-execution',
        service.create_workflow_id(),
        service.create_workflow_execution_id(),
    ]
    util.main_parser(parser, client, args)


@pytest.mark.parametrize('optional_args', [
    ('--next-transition-id', service.create_transition_id()),
    ('--status', 'completed'),
])
def test_executions_update(parser, client, optional_args):
    args = [
        'workflows',
        'update-execution',
        service.create_workflow_id(),
        service.create_workflow_execution_id(),
        *optional_args,
    ]
    util.main_parser(parser, client, args)