err-storage-redis/test_redis.py at master · errbotio/err-storage-redis · 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
import logging

import fakeredis
import pytest

import redisstorage


class MockBotConfig:
    def __init__(self, storage_config=None):
        self.STORAGE_CONFIG = storage_config if storage_config is not None else {}
        self.BOT_LOG_LEVEL = logging.WARNING


def test_storage_operations(monkeypatch):
    fake_redis = fakeredis.FakeStrictRedis()
    monkeypatch.setattr(redisstorage.redis, "StrictRedis", lambda **kwargs: fake_redis)

    config = MockBotConfig()
    plugin = redisstorage.RedisPlugin(config)
    storage = plugin.open("test_namespace")

    # Initial state
    assert storage.len() == 0
    assert list(storage.keys()) == []

    # Get non-existent
    with pytest.raises(KeyError):
        storage.get("key1")

    # Set and get basic values
    storage.set("key1", "value1")
    assert storage.len() == 1
    assert storage.get("key1") == "value1"
    assert list(storage.keys()) == ["key1"]

    # Set and get complex objects (list, dict, int)
    storage.set("key2", {"a": 1, "b": [2, 3]})
    assert storage.len() == 2
    assert storage.get("key2") == {"a": 1, "b": [2, 3]}
    assert set(storage.keys()) == {"key1", "key2"}

    # Overwrite value
    storage.set("key1", "new_value1")
    assert storage.get("key1") == "new_value1"

    # Remove non-existent key
    with pytest.raises(KeyError):
        storage.remove("nonexistent")

    # Remove existing key
    storage.remove("key1")
    assert storage.len() == 1
    with pytest.raises(KeyError):
        storage.get("key1")

    assert list(storage.keys()) == ["key2"]

    # Close storage
    storage.close()


def test_multiple_namespaces(monkeypatch):
    fake_redis = fakeredis.FakeStrictRedis()
    monkeypatch.setattr(redisstorage.redis, "StrictRedis", lambda **kwargs: fake_redis)

    config = MockBotConfig()
    plugin = redisstorage.RedisPlugin(config)

    storage_a = plugin.open("namespace_a")
    storage_b = plugin.open("namespace_b")

    storage_a.set("shared_key", "value_a")
    storage_b.set("shared_key", "value_b")

    assert storage_a.get("shared_key") == "value_a"
    assert storage_b.get("shared_key") == "value_b"

    assert storage_a.len() == 1
    assert storage_b.len() == 1

    storage_a.close()
    storage_b.close()


def test_simple_store_retrieve(monkeypatch):
    from errbot.storage import StoreMixin

    fake_redis = fakeredis.FakeStrictRedis()
    monkeypatch.setattr(redisstorage.redis, "StrictRedis", lambda **kwargs: fake_redis)

    config = MockBotConfig()
    plugin = redisstorage.RedisPlugin(config)
    sm = StoreMixin()
    sm.open_storage(plugin, "ns")
    sm["toto"] = "titui"
    assert sm["toto"] == "titui"
    sm.close_storage()


def test_mutable(monkeypatch):
    from errbot.storage import StoreMixin

    fake_redis = fakeredis.FakeStrictRedis()
    monkeypatch.setattr(redisstorage.redis, "StrictRedis", lambda **kwargs: fake_redis)

    config = MockBotConfig()
    plugin = redisstorage.RedisPlugin(config)
    sm = StoreMixin()
    sm.open_storage(plugin, "ns")
    sm["toto"] = [1, 3]

    with sm.mutable("toto") as titi:
        titi[1] = 5

    assert sm["toto"] == [1, 5]
    sm.close_storage()