frequenz-channels-python/tests/test_grouping_latest_value_cache_integration.py at 7097b3d8dcf401fa606d9ac9f270dbe630cfabdd · llucax/frequenz-channels-python · 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
# License: MIT
# Copyright © 2024 Frequenz Energy-as-a-Service GmbH

"""Tests for the LatestValueCache implementation."""

import asyncio

import pytest

from frequenz.channels import Broadcast
from frequenz.channels.experimental import GroupingLatestValueCache


@pytest.mark.integration
async def test_latest_value_cache_key() -> None:  # pylint: disable=too-many-statements
    """Ensure LatestValueCache works with keys."""
    channel = Broadcast[tuple[int, str]](name="lvc_test")

    cache: GroupingLatestValueCache[int, tuple[int, str]] = GroupingLatestValueCache(
        channel.new_receiver(), key=lambda x: x[0]
    )
    sender = channel.new_sender()

    assert 5 not in cache
    assert cache.get(0) is None

    assert cache.keys() == set()

    await sender.send((5, "a"))
    await sender.send((6, "b"))
    await sender.send((5, "c"))
    await asyncio.sleep(0)

    assert 5 in cache
    assert 6 in cache
    assert 7 not in cache

    assert cache.get(5) == (5, "c")
    assert cache[5] == (5, "c")
    assert cache.get(6) == (6, "b")
    assert cache[6] == (6, "b")

    assert cache.keys() == {5, 6}

    assert cache.get(7, default=(7, "default")) == (7, "default")

    await sender.send((12, "d"))
    await asyncio.sleep(0)

    assert cache.get(12) == (12, "d")
    assert cache.get(12) == (12, "d")
    assert cache.get(5) == (5, "c")
    assert cache.get(6) == (6, "b")

    await sender.send((6, "e"))
    await sender.send((6, "f"))
    await sender.send((6, "g"))
    await asyncio.sleep(0)

    assert cache.get(6) == (6, "g")

    assert cache.keys() == {5, 6, 12}

    del cache[5]
    assert 5 not in cache
    assert 6 in cache

    assert cache.get(5) is None
    assert cache.keys() == {6, 12}

    assert cache.pop(6) == (6, "g")
    assert 6 not in cache
    assert cache.keys() == {12}

    assert cache.pop(8, default=True) is True
    with pytest.raises(KeyError):
        cache.pop(8)

    assert cache.popitem() == (12, (12, "d"))
    assert 12 not in cache
    assert not cache

    await sender.send((1, "h"))
    await sender.send((2, "i"))
    await asyncio.sleep(0)

    expected = {1: (1, "h"), 2: (2, "i")}
    assert cache.keys() == expected.keys()
    assert list(cache.values()) == list(expected.values())
    assert list(cache.items()) == list(expected.items())
    assert cache == expected
    assert list(cache) == list(expected)

    cache.clear()
    assert not cache
    assert cache.keys() == set()

    await cache.stop()


@pytest.mark.integration
async def test_equality() -> None:
    """Test that two caches with the same content are equal."""
    channel = Broadcast[tuple[int, str]](name="lvc_test")

    cache1: GroupingLatestValueCache[int, tuple[int, str]] = GroupingLatestValueCache(
        channel.new_receiver(), key=lambda x: x[0]
    )
    cache2: GroupingLatestValueCache[int, tuple[int, str]] = GroupingLatestValueCache(
        channel.new_receiver(), key=lambda x: x[0]
    )

    sender = channel.new_sender()
    await sender.send((1, "one"))
    await sender.send((2, "two"))
    await asyncio.sleep(0)

    assert cache1 == cache2

    del cache1[1]
    assert cache1 != cache2

    await cache1.stop()
    await cache2.stop()