OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2024-11-24 08:06:26 +01:00
Code Issues Projects Releases Packages Wiki Activity
python-telegram-bot/tests/ext/test_dictpersistence.py
Harshil 8f9db63f4f
Bump ruff and Add New Rules (#4416) 
Co-authored-by: Bibo-Joshi <22366557+Bibo-Joshi@users.noreply.github.com>
2024-08-07 21:56:46 +02:00

388 lines
16 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python
#
# A library that provides a Python interface to the Telegram Bot API
# Copyright (C) 2015-2024
# Leandro Toledo de Souza <devs@python-telegram-bot.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser Public License for more details.
#
# You should have received a copy of the GNU Lesser Public License
# along with this program. If not, see [http://www.gnu.org/licenses/].
import json
import pytest
from telegram.ext import DictPersistence
from tests.auxil.slots import mro_slots
@pytest.fixture(autouse=True)
def _reset_callback_data_cache(cdc_bot):
yield
cdc_bot.callback_data_cache.clear_callback_data()
cdc_bot.callback_data_cache.clear_callback_queries()
@pytest.fixture
def bot_data():
return {"test1": "test2", "test3": {"test4": "test5"}}
@pytest.fixture
def chat_data():
return {-12345: {"test1": "test2", "test3": {"test4": "test5"}}, -67890: {3: "test4"}}
@pytest.fixture
def user_data():
return {12345: {"test1": "test2", "test3": {"test4": "test5"}}, 67890: {3: "test4"}}
@pytest.fixture
def callback_data():
return [("test1", 1000, {"button1": "test0", "button2": "test1"})], {"test1": "test2"}
@pytest.fixture
def conversations():
return {
"name1": {(123, 123): 3, (456, 654): 4},
"name2": {(123, 321): 1, (890, 890): 2},
"name3": {(123, 321): 1, (890, 890): 2},
}
@pytest.fixture
def user_data_json(user_data):
return json.dumps(user_data)
@pytest.fixture
def chat_data_json(chat_data):
return json.dumps(chat_data)
@pytest.fixture
def bot_data_json(bot_data):
return json.dumps(bot_data)
@pytest.fixture
def callback_data_json(callback_data):
return json.dumps(callback_data)
@pytest.fixture
def conversations_json(conversations):
return """{"name1": {"[123, 123]": 3, "[456, 654]": 4}, "name2":
{"[123, 321]": 1, "[890, 890]": 2}, "name3":
{"[123, 321]": 1, "[890, 890]": 2}}"""
class TestDictPersistence:
"""Just tests the DictPersistence interface. Integration of persistence into Applictation
is tested in TestBasePersistence!"""
async def test_slot_behaviour(self):
inst = DictPersistence()
for attr in inst.__slots__:
assert getattr(inst, attr, "err") != "err", f"got extra slot '{attr}'"
assert len(mro_slots(inst)) == len(set(mro_slots(inst))), "duplicate slot"
async def test_no_json_given(self):
dict_persistence = DictPersistence()
assert await dict_persistence.get_user_data() == {}
assert await dict_persistence.get_chat_data() == {}
assert await dict_persistence.get_bot_data() == {}
assert await dict_persistence.get_callback_data() is None
assert await dict_persistence.get_conversations("noname") == {}
async def test_bad_json_string_given(self):
bad_user_data = "thisisnojson99900()))("
bad_chat_data = "thisisnojson99900()))("
bad_bot_data = "thisisnojson99900()))("
bad_callback_data = "thisisnojson99900()))("
bad_conversations = "thisisnojson99900()))("
with pytest.raises(TypeError, match="user_data"):
DictPersistence(user_data_json=bad_user_data)
with pytest.raises(TypeError, match="chat_data"):
DictPersistence(chat_data_json=bad_chat_data)
with pytest.raises(TypeError, match="bot_data"):
DictPersistence(bot_data_json=bad_bot_data)
with pytest.raises(TypeError, match="callback_data"):
DictPersistence(callback_data_json=bad_callback_data)
with pytest.raises(TypeError, match="conversations"):
DictPersistence(conversations_json=bad_conversations)
async def test_invalid_json_string_given(self):
bad_user_data = '["this", "is", "json"]'
bad_chat_data = '["this", "is", "json"]'
bad_bot_data = '["this", "is", "json"]'
bad_conversations = '["this", "is", "json"]'
bad_callback_data_1 = '[[["str", 3.14, {"di": "ct"}]], "is"]'
bad_callback_data_2 = '[[["str", "non-float", {"di": "ct"}]], {"di": "ct"}]'
bad_callback_data_3 = '[[[{"not": "a str"}, 3.14, {"di": "ct"}]], {"di": "ct"}]'
bad_callback_data_4 = '[[["wrong", "length"]], {"di": "ct"}]'
bad_callback_data_5 = '["this", "is", "json"]'
with pytest.raises(TypeError, match="user_data"):
DictPersistence(user_data_json=bad_user_data)
with pytest.raises(TypeError, match="chat_data"):
DictPersistence(chat_data_json=bad_chat_data)
with pytest.raises(TypeError, match="bot_data"):
DictPersistence(bot_data_json=bad_bot_data)
for bad_callback_data in [
bad_callback_data_1,
bad_callback_data_2,
bad_callback_data_3,
bad_callback_data_4,
bad_callback_data_5,
]:
with pytest.raises(TypeError, match="callback_data"):
DictPersistence(callback_data_json=bad_callback_data)
with pytest.raises(TypeError, match="conversations"):
DictPersistence(conversations_json=bad_conversations)
async def test_good_json_input(
self, user_data_json, chat_data_json, bot_data_json, conversations_json, callback_data_json
):
dict_persistence = DictPersistence(
user_data_json=user_data_json,
chat_data_json=chat_data_json,
bot_data_json=bot_data_json,
conversations_json=conversations_json,
callback_data_json=callback_data_json,
)
user_data = await dict_persistence.get_user_data()
assert isinstance(user_data, dict)
assert user_data[12345]["test1"] == "test2"
assert user_data[67890][3] == "test4"
chat_data = await dict_persistence.get_chat_data()
assert isinstance(chat_data, dict)
assert chat_data[-12345]["test1"] == "test2"
assert chat_data[-67890][3] == "test4"
bot_data = await dict_persistence.get_bot_data()
assert isinstance(bot_data, dict)
assert bot_data["test1"] == "test2"
assert bot_data["test3"]["test4"] == "test5"
assert "test6" not in bot_data
callback_data = await dict_persistence.get_callback_data()
assert isinstance(callback_data, tuple)
assert callback_data[0] == [("test1", 1000, {"button1": "test0", "button2": "test1"})]
assert callback_data[1] == {"test1": "test2"}
conversation1 = await dict_persistence.get_conversations("name1")
assert isinstance(conversation1, dict)
assert conversation1[(123, 123)] == 3
assert conversation1[(456, 654)] == 4
with pytest.raises(KeyError):
conversation1[(890, 890)]
conversation2 = await dict_persistence.get_conversations("name2")
assert isinstance(conversation1, dict)
assert conversation2[(123, 321)] == 1
assert conversation2[(890, 890)] == 2
with pytest.raises(KeyError):
conversation2[(123, 123)]
async def test_good_json_input_callback_data_none(self):
dict_persistence = DictPersistence(callback_data_json="null")
assert dict_persistence.callback_data is None
assert dict_persistence.callback_data_json == "null"
async def test_dict_outputs(
self,
user_data,
user_data_json,
chat_data,
chat_data_json,
bot_data,
bot_data_json,
callback_data_json,
conversations,
conversations_json,
):
dict_persistence = DictPersistence(
user_data_json=user_data_json,
chat_data_json=chat_data_json,
bot_data_json=bot_data_json,
callback_data_json=callback_data_json,
conversations_json=conversations_json,
)
assert dict_persistence.user_data == user_data
assert dict_persistence.chat_data == chat_data
assert dict_persistence.bot_data == bot_data
assert dict_persistence.bot_data == bot_data
assert dict_persistence.conversations == conversations
async def test_json_outputs(
self, user_data_json, chat_data_json, bot_data_json, callback_data_json, conversations_json
):
dict_persistence = DictPersistence(
user_data_json=user_data_json,
chat_data_json=chat_data_json,
bot_data_json=bot_data_json,
callback_data_json=callback_data_json,
conversations_json=conversations_json,
)
assert dict_persistence.user_data_json == user_data_json
assert dict_persistence.chat_data_json == chat_data_json
assert dict_persistence.callback_data_json == callback_data_json
assert dict_persistence.conversations_json == conversations_json
async def test_updating(
self,
user_data_json,
chat_data_json,
bot_data_json,
callback_data,
callback_data_json,
conversations,
conversations_json,
):
dict_persistence = DictPersistence(
user_data_json=user_data_json,
chat_data_json=chat_data_json,
bot_data_json=bot_data_json,
callback_data_json=callback_data_json,
conversations_json=conversations_json,
)
user_data = await dict_persistence.get_user_data()
user_data[12345]["test3"]["test4"] = "test6"
assert dict_persistence.user_data != user_data
assert dict_persistence.user_data_json != json.dumps(user_data)
await dict_persistence.update_user_data(12345, user_data[12345])
assert dict_persistence.user_data == user_data
assert dict_persistence.user_data_json == json.dumps(user_data)
await dict_persistence.drop_user_data(67890)
assert 67890 not in dict_persistence.user_data
dict_persistence._user_data = None
await dict_persistence.drop_user_data(123)
assert isinstance(await dict_persistence.get_user_data(), dict)
chat_data = await dict_persistence.get_chat_data()
chat_data[-12345]["test3"]["test4"] = "test6"
assert dict_persistence.chat_data != chat_data
assert dict_persistence.chat_data_json != json.dumps(chat_data)
await dict_persistence.update_chat_data(-12345, chat_data[-12345])
assert dict_persistence.chat_data == chat_data
assert dict_persistence.chat_data_json == json.dumps(chat_data)
await dict_persistence.drop_chat_data(-67890)
assert -67890 not in dict_persistence.chat_data
dict_persistence._chat_data = None
await dict_persistence.drop_chat_data(123)
assert isinstance(await dict_persistence.get_chat_data(), dict)
bot_data = await dict_persistence.get_bot_data()
bot_data["test3"]["test4"] = "test6"
assert dict_persistence.bot_data != bot_data
assert dict_persistence.bot_data_json != json.dumps(bot_data)
await dict_persistence.update_bot_data(bot_data)
assert dict_persistence.bot_data == bot_data
assert dict_persistence.bot_data_json == json.dumps(bot_data)
callback_data = await dict_persistence.get_callback_data()
callback_data[1]["test3"] = "test4"
callback_data[0][0][2]["button2"] = "test41"
assert dict_persistence.callback_data != callback_data
assert dict_persistence.callback_data_json != json.dumps(callback_data)
await dict_persistence.update_callback_data(callback_data)
assert dict_persistence.callback_data == callback_data
assert dict_persistence.callback_data_json == json.dumps(callback_data)
conversation1 = await dict_persistence.get_conversations("name1")
conversation1[(123, 123)] = 5
assert dict_persistence.conversations["name1"] != conversation1
await dict_persistence.update_conversation("name1", (123, 123), 5)
assert dict_persistence.conversations["name1"] == conversation1
conversations["name1"][(123, 123)] = 5
assert (
dict_persistence.conversations_json
== DictPersistence._encode_conversations_to_json(conversations)
)
assert await dict_persistence.get_conversations("name1") == conversation1
dict_persistence._conversations = None
await dict_persistence.update_conversation("name1", (123, 123), 5)
assert dict_persistence.conversations["name1"] == {(123, 123): 5}
assert await dict_persistence.get_conversations("name1") == {(123, 123): 5}
assert (
dict_persistence.conversations_json
== DictPersistence._encode_conversations_to_json({"name1": {(123, 123): 5}})
)
async def test_no_data_on_init(
self, bot_data, user_data, chat_data, conversations, callback_data
):
dict_persistence = DictPersistence()
assert dict_persistence.user_data is None
assert dict_persistence.chat_data is None
assert dict_persistence.bot_data is None
assert dict_persistence.conversations is None
assert dict_persistence.callback_data is None
assert dict_persistence.user_data_json == "null"
assert dict_persistence.chat_data_json == "null"
assert dict_persistence.bot_data_json == "null"
assert dict_persistence.conversations_json == "null"
assert dict_persistence.callback_data_json == "null"
await dict_persistence.update_bot_data(bot_data)
await dict_persistence.update_user_data(12345, user_data[12345])
await dict_persistence.update_chat_data(-12345, chat_data[-12345])
await dict_persistence.update_conversation("name", (1, 1), "new_state")
await dict_persistence.update_callback_data(callback_data)
assert dict_persistence.user_data[12345] == user_data[12345]
assert dict_persistence.chat_data[-12345] == chat_data[-12345]
assert dict_persistence.bot_data == bot_data
assert dict_persistence.conversations["name"] == {(1, 1): "new_state"}
assert dict_persistence.callback_data == callback_data
async def test_no_json_dumping_if_data_did_not_change(
self, bot_data, user_data, chat_data, conversations, callback_data, monkeypatch
):
dict_persistence = DictPersistence()
await dict_persistence.update_bot_data(bot_data)
await dict_persistence.update_user_data(12345, user_data[12345])
await dict_persistence.update_chat_data(-12345, chat_data[-12345])
await dict_persistence.update_conversation("name", (1, 1), "new_state")
await dict_persistence.update_callback_data(callback_data)
assert dict_persistence.user_data_json == json.dumps({12345: user_data[12345]})
assert dict_persistence.chat_data_json == json.dumps({-12345: chat_data[-12345]})
assert dict_persistence.bot_data_json == json.dumps(bot_data)
assert (
dict_persistence.conversations_json
== DictPersistence._encode_conversations_to_json({"name": {(1, 1): "new_state"}})
)
assert dict_persistence.callback_data_json == json.dumps(callback_data)
flag = False
def dumps(*args, **kwargs):
nonlocal flag
flag = True
# Since the data doesn't change, json.dumps shoduln't be called beyond this point!
monkeypatch.setattr(json, "dumps", dumps)
await dict_persistence.update_bot_data(bot_data)
await dict_persistence.update_user_data(12345, user_data[12345])
await dict_persistence.update_chat_data(-12345, chat_data[-12345])
await dict_persistence.update_conversation("name", (1, 1), "new_state")
await dict_persistence.update_callback_data(callback_data)
assert not flag
Reference in a new issue View git blame Copy permalink