harmont-cli/crates/hm-dsl-engine/harmont-py/tests/test_cache_predicate.py at cdab6edf48e9fe90fcf5ec66dd422a1bef0edb50 · harmont-dev/harmont-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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
"""Tests for the ``predicate`` cache policy."""

from __future__ import annotations

import hashlib
from pathlib import Path

import harmont as hm
from harmont._pipeline import _cache_to_dict
from harmont.cache import CachePredicate
from harmont.keygen import resolve_pipeline_keys

NUL = "\x00"


def _sha256_hex(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()


def _make_graph(nodes, edges=None):
    return {
        "nodes": nodes,
        "node_holes": [],
        "edge_property": "directed",
        "edges": edges or [],
    }


# -- factory -----------------------------------------------------------------


def test_predicate_factory_returns_cache_predicate():
    policy = hm.predicate(lambda: "v1")
    assert isinstance(policy, CachePredicate)


# -- lowering ----------------------------------------------------------------


def test_predicate_lowering_calls_fn_and_produces_dict():
    policy = hm.predicate(lambda: "v1")
    d = _cache_to_dict(policy)
    assert d == {"policy": "predicate", "value": "v1"}


def test_predicate_lowering_stringifies_return_value():
    policy = hm.predicate(lambda: 42)
    d = _cache_to_dict(policy)
    assert d == {"policy": "predicate", "value": "42"}


def test_predicate_fn_is_called_at_lowering_time():
    calls: list[int] = []

    def counter():
        calls.append(1)
        return "x"

    policy = hm.predicate(counter)
    assert len(calls) == 0
    _cache_to_dict(policy)
    assert len(calls) == 1


# -- keygen ------------------------------------------------------------------


def test_predicate_keygen_produces_deterministic_key():
    graph = _make_graph(
        [
            {
                "step": {
                    "key": "a",
                    "cmd": "echo",
                    "cache": {"policy": "predicate", "value": "v1"},
                },
                "env": {},
            },
        ]
    )
    out = resolve_pipeline_keys(
        graph,
        pipeline_org="default",
        pipeline_slug="default",
        now=0,
        base_path=Path("/tmp"),  # noqa: S108
        env={},
    )
    policy_res = "predicate-" + _sha256_hex("v1")
    expected = _sha256_hex(
        "default" + NUL + "default" + NUL + "a" + NUL + "scratch" + NUL + policy_res
    )
    assert out["nodes"][0]["step"]["cache"]["key"] == expected


def test_different_predicate_values_produce_different_keys():
    def make_graph(value):
        return _make_graph(
            [
                {
                    "step": {
                        "key": "a",
                        "cmd": "echo",
                        "cache": {"policy": "predicate", "value": value},
                    },
                    "env": {},
                },
            ]
        )

    g1 = make_graph("v1")
    g2 = make_graph("v2")
    resolve_pipeline_keys(
        g1,
        pipeline_org="o",
        pipeline_slug="s",
        now=0,
        base_path=Path("/tmp"),  # noqa: S108
        env={},
    )
    resolve_pipeline_keys(
        g2,
        pipeline_org="o",
        pipeline_slug="s",
        now=0,
        base_path=Path("/tmp"),  # noqa: S108
        env={},
    )
    assert g1["nodes"][0]["step"]["cache"]["key"] != g2["nodes"][0]["step"]["cache"]["key"]