frequenz-channels-python/tests/test_grouping_latest_value_cache_integration.py at ccd751e46e188f3b52c9d912d15b60416227e6df · shsms/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
# 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:
    """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.get(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}

    cache.clear(5)
    assert 5 not in cache
    assert 6 in cache

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