metaflow/test/unit/test_argo_ttl_strategy.py at 21bb155e1ea4763209acf547c10b577b06c748c7 · Netflix/metaflow · 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
"""Tests for WorkflowSpec.ttl_strategy()."""

from metaflow.plugins.argo.argo_workflows import WorkflowSpec


def test_positive_value_sets_ttl():
    ws = WorkflowSpec()
    ws.ttl_strategy(604800)
    assert ws.payload["ttlStrategy"] == {"secondsAfterCompletion": 604800}


def test_zero_disables_ttl():
    ws = WorkflowSpec()
    ws.ttl_strategy(0)
    assert "ttlStrategy" not in ws.payload


def test_none_disables_ttl():
    ws = WorkflowSpec()
    ws.ttl_strategy(None)
    assert "ttlStrategy" not in ws.payload


def test_negative_disables_ttl():
    ws = WorkflowSpec()
    ws.ttl_strategy(-1)
    assert "ttlStrategy" not in ws.payload


def test_string_value_converted_to_int():
    ws = WorkflowSpec()
    ws.ttl_strategy("86400")
    assert ws.payload["ttlStrategy"] == {"secondsAfterCompletion": 86400}


def test_chaining():
    """ttl_strategy returns self for method chaining."""
    ws = WorkflowSpec()
    result = ws.ttl_strategy(3600)
    assert result is ws