830 lines
30 KiB
Python
830 lines
30 KiB
Python
import os
|
||
import sqlite3
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
|
||
from database.db import BotDB
|
||
|
||
|
||
@pytest.fixture
|
||
def bot():
|
||
"""Фикстура для создания объекта BotDB."""
|
||
current_dir = os.getcwd()
|
||
return BotDB(current_dir, "database/test.db")
|
||
|
||
|
||
@pytest.fixture(autouse=True, )
|
||
def setup_db():
|
||
"""Фикстура для создания всей базы перед каждым тестом."""
|
||
# Mock data 1st user
|
||
user_id = 12345
|
||
first_name = "Иван"
|
||
full_name = "Иван Иванович"
|
||
username = "@iban"
|
||
message_text = 'Hello, planet'
|
||
message_id = 1
|
||
message_for_user = "LOL"
|
||
has_stickers = 0
|
||
# Mock data 2nd user
|
||
user_id_2 = 14278
|
||
first_name_2 = "Борис"
|
||
full_name_2 = "Борис Петрович"
|
||
username_2 = "@boris"
|
||
message_text_2 = 'Hello, world'
|
||
message_id_2 = 2
|
||
message_for_user_2 = "LOL2"
|
||
has_stickers_2 = 1
|
||
# Other data
|
||
date = "2024-07-10"
|
||
next_date = "2024-07-11"
|
||
conn = sqlite3.connect("database/test.db")
|
||
cursor = conn.cursor()
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS "admins" (
|
||
user_id INTEGER NOT NULL,
|
||
"role" TEXT
|
||
);
|
||
""")
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS "audio_message_reference"
|
||
(
|
||
"id" INTEGER NOT NULL UNIQUE,
|
||
"file_name" TEXT NOT NULL UNIQUE,
|
||
"author_id" INTEGER NOT NULL,
|
||
"date_added" DATE NOT NULL,
|
||
"listen_count" INTEGER NOT NULL,
|
||
"file_id" INTEGER NOT NULL,
|
||
PRIMARY KEY ("id")
|
||
);
|
||
""")
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS "blacklist"
|
||
(
|
||
"user_id" INTEGER NOT NULL UNIQUE,
|
||
"user_name" INTEGER,
|
||
"message_for_user" INTEGER,
|
||
"date_to_unban" INTEGER
|
||
);
|
||
""")
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS "messages" (
|
||
"ID" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
|
||
"Message" TEXT NOT NULL,
|
||
"type" INTEGER
|
||
);
|
||
""")
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS "our_users" (
|
||
"id" INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT UNIQUE,
|
||
"user_id" INTEGER NOT NULL UNIQUE,
|
||
"first_name" STRING,
|
||
"full_name" STRING,
|
||
"username" STRING,
|
||
"is_bot" BOOLEAN,
|
||
"language_code" STRING,
|
||
"has_stickers" INTEGER NOT NULL DEFAULT 0,
|
||
"date_added" DATE NOT NULL,
|
||
"date_changed" DATE NOT NULL
|
||
, state_user TEXT(20));
|
||
""")
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS user_messages (
|
||
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
|
||
message_text TEXT,
|
||
user_id INTEGER,
|
||
message_id INTEGER NOT NULL,
|
||
date TEXT
|
||
);
|
||
""")
|
||
cursor.execute("""
|
||
CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT);
|
||
""")
|
||
cursor.execute("""
|
||
CREATE TABLE migrations (
|
||
version INTEGER PRIMARY KEY NOT NULL,
|
||
script_name TEXT NOT NULL,
|
||
created_at TEXT
|
||
);
|
||
""")
|
||
|
||
# blacklist mock data
|
||
cursor.execute("INSERT INTO blacklist (user_id, user_name, message_for_user, date_to_unban) VALUES (?, ?, ?, ?)",
|
||
(user_id, username, message_for_user, next_date))
|
||
cursor.execute("INSERT INTO blacklist (user_id, user_name, message_for_user, date_to_unban) VALUES (?, ?, ?, ?)",
|
||
(user_id_2, username_2, message_for_user_2, date))
|
||
# our_users mock data
|
||
cursor.execute(
|
||
"INSERT INTO our_users (user_id, first_name, full_name, username, date_added, date_changed, has_stickers)"
|
||
" VALUES (?, ?, ?, ?, ?, ?, ?)", (user_id, first_name, full_name, username, date, date, has_stickers)
|
||
)
|
||
cursor.execute(
|
||
"INSERT INTO our_users (user_id, first_name, full_name, username, date_added, date_changed, has_stickers)"
|
||
" VALUES (?, ?, ?, ?, ?, ?, ?)", (user_id_2, first_name_2, full_name_2, username_2, date, date, has_stickers_2)
|
||
)
|
||
# messages mock data
|
||
cursor.execute(
|
||
"INSERT INTO user_messages (message_text, user_id, message_id, date) "
|
||
"VALUES (?, ?, ?, ?)",
|
||
(message_text, user_id, message_id, date))
|
||
cursor.execute(
|
||
"INSERT INTO user_messages (message_text, user_id, message_id, date) "
|
||
"VALUES (?, ?, ?, ?)",
|
||
(message_text_2, user_id_2, message_id_2, date))
|
||
# mock admins
|
||
cursor.execute(
|
||
"INSERT INTO admins (user_id, role) "
|
||
"VALUES (?, ?)",
|
||
(user_id, 'creator'))
|
||
conn.commit()
|
||
conn.close()
|
||
yield
|
||
os.remove('database/test.db')
|
||
|
||
|
||
def test_bot_init(bot):
|
||
"""Проверяет, что объект BotDB инициализируется с правильным именем файла."""
|
||
assert bot.db_file == os.path.join(os.getcwd(), "database", "test.db")
|
||
# Проверьте, что соединения с базой данных нет, так как оно не устанавливается в init
|
||
assert bot.conn is None
|
||
assert bot.cursor is None
|
||
|
||
|
||
def test_bot_connect(bot):
|
||
"""Проверяет, что метод connect создает подключение к базе данных."""
|
||
bot.connect()
|
||
assert bot.conn is not None
|
||
assert bot.cursor is not None
|
||
bot.close()
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_bot_close(bot):
|
||
"""Проверяет, что метод close закрывает подключение к базе данных."""
|
||
bot.connect()
|
||
assert bot.conn is not None
|
||
assert bot.cursor is not None
|
||
bot.close()
|
||
assert bot.conn is None
|
||
assert bot.cursor is None
|
||
|
||
|
||
def test_create_table_success(bot):
|
||
sql_script = 'CREATE TABLE test_table (id INTEGER PRIMARY KEY);'
|
||
bot.create_table(sql_script)
|
||
|
||
# Проверяем, что таблица создана
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test_table'")
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
|
||
assert result is not None
|
||
|
||
|
||
def test_create_table_error(bot):
|
||
sql_script = 'CREATE TABLE test_table (id INTEGER PRIMARY KEY);'
|
||
bot.create_table(sql_script)
|
||
|
||
with pytest.raises(sqlite3.OperationalError):
|
||
bot.create_table(sql_script)
|
||
|
||
|
||
def test_get_current_version_success(bot):
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("INSERT INTO migrations (version, script_name) VALUES (123, 'test')")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
# Вызываем функцию и проверяем результат
|
||
version = bot.get_current_version()
|
||
assert version == 123
|
||
|
||
|
||
def test_get_current_version_error(bot):
|
||
__drop_table('migrations')
|
||
with pytest.raises(sqlite3.OperationalError):
|
||
bot.get_current_version()
|
||
|
||
|
||
def test_update_version_success(bot):
|
||
# Вызываем функцию update_version
|
||
new_version = 124
|
||
script_name = "migration_script.sql"
|
||
bot.update_version(new_version, script_name)
|
||
|
||
# Проверяем, что данные записаны в таблицу
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT * FROM migrations WHERE version = ?", (new_version,))
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
assert result is not None
|
||
assert result[0] == new_version
|
||
assert result[1] == script_name
|
||
assert result[2] == datetime.now().strftime("%d-%m-%Y %H:%M:%S")
|
||
|
||
|
||
def test_update_version_integrity_error(bot):
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("INSERT INTO migrations (version, script_name) VALUES (123, 'test')")
|
||
conn.commit()
|
||
conn.close()
|
||
# Пытаемся обновить версию с уже существующим значением
|
||
with pytest.raises(sqlite3.IntegrityError):
|
||
bot.update_version(123, "script_2.sql")
|
||
|
||
|
||
def test_update_version_error(bot):
|
||
__drop_table('migrations')
|
||
with pytest.raises(sqlite3.OperationalError):
|
||
bot.update_version(123, "script_2.sql")()
|
||
|
||
|
||
def test_add_new_user_in_db(bot):
|
||
"""Проверяет добавление нового пользователя в базу данных."""
|
||
user_id = 50
|
||
first_name = "Петр"
|
||
full_name = "Петр Иванов"
|
||
username = "@petr_ivanov"
|
||
is_bot = False
|
||
language_code = "ru"
|
||
date_added = "2024-07-09"
|
||
date_changed = "2024-07-09"
|
||
|
||
# Вызываем функцию add_new_user_in_db
|
||
bot.add_new_user_in_db(
|
||
user_id, first_name, full_name, username, is_bot, language_code, date_added, date_changed
|
||
)
|
||
|
||
# Проверяем наличие записи в базе данных
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT * FROM our_users WHERE user_id = ?", (user_id,))
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
|
||
assert result is not None
|
||
assert result[1] == user_id
|
||
assert result[2] == first_name
|
||
assert result[3] == full_name
|
||
assert result[4] == username
|
||
assert result[5] == is_bot
|
||
assert result[6] == language_code
|
||
assert result[8] == date_added
|
||
assert result[9] == date_changed
|
||
|
||
|
||
def test_add_new_user_in_db_duplicate_user_id(bot, setup_db):
|
||
"""Проверяет поведение при попытке добавить пользователя с уже существующим user_id."""
|
||
user_id = 12345
|
||
|
||
# Попытка добавить пользователя с тем же user_id
|
||
with pytest.raises(sqlite3.IntegrityError):
|
||
bot.add_new_user_in_db(
|
||
user_id, "Марина", "Марина Альфредовна", "marina", False, "bg", "2024-07-09", "2024-07-09"
|
||
)
|
||
|
||
|
||
def test_add_new_user_in_db_empty_first_name(bot):
|
||
""" Проверяет добавление пользователя с пустым именем (first_name) """
|
||
user_id = 43
|
||
first_name = "" # Пустое имя
|
||
full_name = "Boris Petrov"
|
||
username = "@boris"
|
||
is_bot = False
|
||
language_code = "fr"
|
||
date_added = "2024-07-09"
|
||
date_changed = "2024-07-09"
|
||
|
||
# Вызываем функцию add_new_user_in_db
|
||
bot.add_new_user_in_db(
|
||
user_id, first_name, full_name, username, is_bot, language_code, date_added, date_changed
|
||
)
|
||
|
||
# Проверяем наличие записи в базе данных
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute(f"SELECT * FROM our_users WHERE user_id = ?", (user_id,))
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
assert result is not None
|
||
assert result[1] == user_id
|
||
assert result[2] == first_name
|
||
assert result[3] == full_name
|
||
assert result[4] == username
|
||
assert result[5] == is_bot
|
||
assert result[6] == language_code
|
||
assert result[8] == date_added
|
||
assert result[9] == date_changed
|
||
|
||
|
||
def test_user_exists_found(bot):
|
||
"""Проверяет, что функция возвращает True, если пользователь найден."""
|
||
user_id = 12345
|
||
|
||
# Проверяем наличие записи в базе данных
|
||
assert bot.user_exists(user_id) is True
|
||
|
||
|
||
def test_user_exists_not_found(bot):
|
||
"""Проверяет, что функция возвращает False, если пользователь не найден."""
|
||
user_id = 99999
|
||
assert bot.user_exists(user_id) is False
|
||
|
||
|
||
def test_user_exists_error(bot):
|
||
"""Проверяет, что функция возвращает ошибки"""
|
||
__drop_table('our_users')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.user_exists(12345)
|
||
|
||
|
||
def test_get_user_id_found(bot):
|
||
"""Проверяет, что функция возвращает ID пользователя, если он найден."""
|
||
user_id = 12345
|
||
# Проверяем, что возвращается правильный ID из базы
|
||
user_id_db = bot.get_user_id(user_id)
|
||
assert user_id_db == 1
|
||
|
||
|
||
def test_get_user_id_not_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает None, если пользователь не найден."""
|
||
user_id = 99999
|
||
assert bot.get_user_id(user_id) is None
|
||
|
||
|
||
def test_get_user_id_error(bot):
|
||
"""Проверяет, что функция обрабатывает некорректный user_id."""
|
||
__drop_table('our_users')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_user_id(12345)
|
||
|
||
|
||
def test_get_username_found(bot):
|
||
"""Проверяет, что функция возвращает username пользователя, если он найден."""
|
||
user_id = 12345
|
||
username = "@iban"
|
||
# Проверяем, что возвращается правильный username из базы
|
||
username_db = bot.get_username(user_id)
|
||
assert username_db == username
|
||
|
||
|
||
def test_get_username_not_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает None, если пользователь не найден."""
|
||
user_id = 99999
|
||
assert bot.get_username(user_id) is None
|
||
|
||
|
||
def test_get_username_error(bot):
|
||
"""Проверяет, что функция возвращает ошибку"""
|
||
__drop_table('our_users')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_username(12345)
|
||
|
||
|
||
def test_get_all_user_id_empty(bot):
|
||
"""Проверяет, что функция возвращает пустой список, если в базе нет пользователей."""
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("DELETE FROM our_users")
|
||
conn.commit()
|
||
conn.close()
|
||
# Проверяем наличие записей в базе данных
|
||
user_ids = bot.get_all_user_id()
|
||
assert user_ids == []
|
||
|
||
|
||
def test_get_all_user_id_non_empty(bot, setup_db):
|
||
"""Проверяет, что функция возвращает список всех user_id из базы данных."""
|
||
# Проверяем наличие записи в базе данных
|
||
user_ids = bot.get_all_user_id()
|
||
assert user_ids == [12345, 14278] # Проверяем, что в списке два ожидаемых user_id
|
||
|
||
|
||
def test_get_all_user_id_error(bot):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('our_users')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_all_user_id()
|
||
|
||
|
||
def test_get_user_first_name_found(bot):
|
||
"""Проверяет, что функция возвращает имя пользователя, если он найден."""
|
||
user_id = 12345
|
||
first_name = bot.get_user_first_name(user_id)
|
||
assert first_name == "Иван"
|
||
|
||
|
||
def test_get_user_first_name_not_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает None, если пользователь не найден."""
|
||
user_id = 99999
|
||
assert bot.get_user_first_name(user_id) is None
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_get_user_first_name_invalid_user_id(bot):
|
||
"""Проверяет, что функция обрабатывает некорректный user_id."""
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_user_first_name("invalid_user_id") # Передача строки
|
||
|
||
|
||
def test_get_user_first_name_error(bot):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('our_users')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_user_first_name(12345)
|
||
|
||
|
||
def test_get_info_about_stickers_found_received(bot):
|
||
"""Проверяет, что функция возвращает True, если пользователь получил стикеры."""
|
||
user_id = 14278
|
||
assert bot.get_info_about_stickers(user_id) is True
|
||
|
||
|
||
def test_get_info_about_stickers_found_not_received(bot, setup_db):
|
||
"""Проверяет, что функция возвращает False, если пользователь не получил стикеры."""
|
||
user_id = 12345
|
||
assert bot.get_info_about_stickers(user_id) is False
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_get_info_about_stickers_not_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает None, если пользователь не найден."""
|
||
user_id = 99999
|
||
assert bot.get_info_about_stickers(user_id) is None
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_get_info_about_stickers_invalid_user_id(bot):
|
||
"""Проверяет, что функция обрабатывает некорректный user_id."""
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_info_about_stickers("invalid_user_id")
|
||
|
||
|
||
def test_get_info_about_stickers_error(bot):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('our_users')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_info_about_stickers(12345)
|
||
|
||
|
||
def test_update_info_about_stickers_success(bot):
|
||
"""Проверяет, что функция успешно обновляет информацию о получении стикеров."""
|
||
user_id = 12345
|
||
bot.update_info_about_stickers(user_id)
|
||
|
||
# Проверяем, что информация обновлена
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT has_stickers FROM our_users WHERE user_id = ?", (user_id,))
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
assert result[0] == 1
|
||
|
||
|
||
def test_update_info_about_stickers_not_found(bot):
|
||
"""Проверяет, что функция не вызывает ошибки, если пользователь не найден."""
|
||
user_id = 99999
|
||
bot.update_info_about_stickers(user_id)
|
||
|
||
# Проверяем, что база данных не изменилась
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT COUNT(*) FROM our_users WHERE user_id = ?", (user_id,))
|
||
result = cursor.fetchone()
|
||
conn.close()
|
||
assert result[0] == 0
|
||
|
||
|
||
def test_update_info_about_stickers_error(bot):
|
||
"""Проверяет, что функция вызывает ошибки"""
|
||
__drop_table('our_users')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.update_info_about_stickers(12345)
|
||
|
||
|
||
def test_get_users_blacklist_empty(bot):
|
||
"""Проверяет, что функция возвращает пустой словарь, если в черном списке нет пользователей."""
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("DELETE FROM blacklist")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
blacklist = bot.get_users_blacklist()
|
||
assert blacklist == {}
|
||
|
||
|
||
def test_get_users_blacklist_non_empty(bot):
|
||
"""Проверяет, что функция возвращает словарь с пользователями из черного списка."""
|
||
blacklist = bot.get_users_blacklist()
|
||
assert blacklist == {12345: "@iban", 14278: "@boris"}
|
||
|
||
|
||
def test_get_users_blacklist_error(bot):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('blacklist')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_users_blacklist()
|
||
|
||
|
||
def test_get_blacklist_users_by_id_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает информацию о пользователе, если он найден в черном списке."""
|
||
user_id = 12345
|
||
|
||
result = bot.get_blacklist_users_by_id(user_id)
|
||
assert result == (12345, "@iban", "LOL", "2024-07-11")
|
||
|
||
|
||
def test_get_blacklist_users_by_id_not_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает None, если пользователь не найден в черном списке."""
|
||
user_id = 99999
|
||
assert bot.get_blacklist_users_by_id(user_id) is None
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_get_blacklist_users_by_id_invalid_user_id(bot):
|
||
"""Проверяет, что функция обрабатывает некорректный user_id."""
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_blacklist_users_by_id("invalid_user_id") # Передача строки
|
||
|
||
|
||
def test_get_blacklist_users_by_id_error(bot):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('blacklist')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_blacklist_users_by_id(12345)
|
||
|
||
|
||
def test_get_users_for_unblock_today_found(bot):
|
||
"""Проверяет, что функция возвращает словарь с пользователями, у которых истекает блокировка сегодня."""
|
||
date_to_unban = "2024-07-11"
|
||
result = bot.get_users_for_unblock_today(date_to_unban)
|
||
assert result == {12345: "@iban"}
|
||
|
||
|
||
def test_get_users_for_unblock_today_not_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает пустой словарь, если сегодня нет пользователей, у которых истекает блокировка."""
|
||
date_to_unban = "2024-07-12"
|
||
result = bot.get_users_for_unblock_today(date_to_unban)
|
||
assert result == {}
|
||
|
||
|
||
def test_get_users_for_unblock_today_error(bot):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('blacklist')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_users_for_unblock_today("2023-12-26")
|
||
|
||
|
||
def test_check_user_in_blacklist_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает True, если пользователь найден в черном списке."""
|
||
user_id = 12345
|
||
bot.set_user_blacklist(user_id, "JohnDoe") # Добавляем пользователя в черный список
|
||
|
||
assert bot.check_user_in_blacklist(user_id) is True
|
||
|
||
|
||
def test_check_user_in_blacklist_not_found(bot, setup_db):
|
||
"""Проверяет, что функция возвращает False, если пользователь не найден в черном списке."""
|
||
user_id = 99999
|
||
assert bot.check_user_in_blacklist(user_id) is False
|
||
|
||
|
||
def test_check_user_in_blacklist_error(bot, setup_db):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('blacklist')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.check_user_in_blacklist(12345)
|
||
|
||
|
||
def test_set_user_blacklist_success(bot):
|
||
"""Проверяет, что функция успешно добавляет пользователя в черный список."""
|
||
user_id = 11
|
||
user_name = "Гриша"
|
||
message_for_user = "Лови бан!"
|
||
date_to_unban = datetime.now().strftime("%Y-%m-%d") # Текущая дата
|
||
|
||
assert bot.set_user_blacklist(user_id, user_name, message_for_user, date_to_unban) is None
|
||
|
||
# Проверяем, что запись добавлена в базу
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT * FROM blacklist WHERE user_id = ?", (user_id,))
|
||
result = cursor.fetchone()
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
assert result is not None
|
||
assert result[1] == user_name
|
||
assert result[2] == message_for_user
|
||
assert result[3] == date_to_unban
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_set_user_blacklist_duplicate_user_id(bot, setup_db):
|
||
"""Проверяет, что функция не добавляет дубликат user_id в черный список."""
|
||
user_id = 12345
|
||
bot.set_user_blacklist(user_id, "JohnDoe")
|
||
|
||
with pytest.raises(sqlite3.IntegrityError):
|
||
bot.set_user_blacklist(user_id, "JaneSmith") # Попытка добавить дубликат
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_set_user_blacklist_error(bot, setup_db):
|
||
"""Проверяет, что функция вызывает sqlite3. Error при ошибке запроса."""
|
||
__drop_table('blacklist')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.set_user_blacklist(12345, "JohnDoe", "You are banned!", "2024-01-01")
|
||
|
||
|
||
def test_delete_user_blacklist_success(bot):
|
||
bot.delete_user_blacklist(12345)
|
||
assert bot.check_user_in_blacklist(12345) is False
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_delete_user_blacklist_not_found(bot):
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("INSERT INTO blacklist (user_id, user_name, date_to_unban) VALUES (?, ?, ?)",
|
||
(12345, "JohnDoe", "2023-12-26"))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
result = bot.delete_user_blacklist(514)
|
||
assert result is False
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_delete_user_blacklist_error(bot):
|
||
__drop_table('blacklist')
|
||
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.delete_user_blacklist(12345)
|
||
|
||
|
||
def test_add_new_message_in_db_success(bot):
|
||
result = bot.add_new_message_in_db('hello', 4232187, 5, '2024-01-01')
|
||
assert result is None
|
||
|
||
|
||
def test_add_new_message_in_db_error(bot):
|
||
__drop_table('user_messages')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.add_new_message_in_db('hello', 12345, 1, '2024-01-01')
|
||
|
||
|
||
def test_update_date_for_user_success(bot):
|
||
bot.update_date_for_user('2024-07-15', 12345)
|
||
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("SELECT date_changed FROM our_users WHERE user_id = ?", (12345,))
|
||
new_date = cursor.fetchone()[0]
|
||
conn.close()
|
||
assert new_date == '2024-07-15'
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_update_date_for_user_error(bot):
|
||
__drop_table('our_users')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.update_date_for_user('2024-07-15', 12345)
|
||
|
||
|
||
def test_is_admin_success(bot):
|
||
assert bot.is_admin(12345) is True
|
||
|
||
|
||
def test_is_admin_not_found(bot):
|
||
assert bot.is_admin(1) is False
|
||
|
||
|
||
def test_is_admin_error(bot):
|
||
__drop_table('admins')
|
||
assert bot.is_admin(1) is None
|
||
|
||
|
||
def test_get_user_by_message_id_success(bot):
|
||
assert bot.get_user_by_message_id(1) == 12345
|
||
|
||
|
||
@pytest.mark.xfail
|
||
def test_get_user_by_message_id_not_found(bot):
|
||
assert bot.get_user_by_message_id(124) == None
|
||
|
||
|
||
def test_get_user_by_message_id_error(bot):
|
||
__drop_table('user_messages')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_user_by_message_id(14)
|
||
|
||
|
||
def test_get_last_users_from_db_success(bot):
|
||
users = bot.get_last_users_from_db()
|
||
assert users is not None
|
||
assert len(users) == 2
|
||
|
||
|
||
def test_get_last_users_from_db_empty(bot):
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("DELETE FROM our_users")
|
||
conn.commit()
|
||
conn.close()
|
||
users = bot.get_last_users_from_db()
|
||
assert users == []
|
||
assert len(users) == 0
|
||
|
||
|
||
def test_get_user_by_message_id_error(bot):
|
||
__drop_table('our_users')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_last_users_from_db()
|
||
|
||
|
||
def test_get_banned_users_from_db_success(bot):
|
||
users = bot.get_banned_users_from_db()
|
||
assert users[0][0] == '@iban'
|
||
assert users[0][1] == 12345
|
||
assert users[0][2] == 'LOL'
|
||
assert users[1][0] == '@boris'
|
||
assert users[1][1] == 14278
|
||
assert users[1][2] == 'LOL2'
|
||
|
||
|
||
def test_get_banned_users_from_db_empty(bot):
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("DELETE FROM blacklist")
|
||
conn.commit()
|
||
conn.close()
|
||
users = bot.get_banned_users_from_db()
|
||
assert users == []
|
||
assert len(users) == 0
|
||
|
||
|
||
def test_get_banned_users_from_db_error(bot):
|
||
__drop_table('blacklist')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_banned_users_from_db()
|
||
|
||
|
||
def test_get_banned_users_from_db_with_limits_success_limit(bot):
|
||
users = bot.get_banned_users_from_db_with_limits(0, 1)
|
||
assert users[0][0] == '@iban'
|
||
assert users[0][1] == 12345
|
||
assert users[0][2] == 'LOL'
|
||
assert len(users) == 1
|
||
|
||
|
||
def test_get_banned_users_from_db_with_limits_success_offset(bot):
|
||
users = bot.get_banned_users_from_db_with_limits(1, 2)
|
||
assert users[0][0] == '@boris'
|
||
assert users[0][1] == 14278
|
||
assert users[0][2] == 'LOL2'
|
||
assert len(users) == 1
|
||
|
||
|
||
def test_get_banned_users_from_db_with_limits_empty(bot):
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute("DELETE FROM blacklist")
|
||
conn.commit()
|
||
conn.close()
|
||
users = bot.get_banned_users_from_db_with_limits(0, 2)
|
||
assert users == []
|
||
assert len(users) == 0
|
||
|
||
|
||
def test_get_banned_users_from_db_with_limits_error(bot):
|
||
__drop_table('blacklist')
|
||
with pytest.raises(sqlite3.Error):
|
||
bot.get_banned_users_from_db_with_limits(0, 2)
|
||
|
||
|
||
def __drop_table(table_name: str):
|
||
conn = sqlite3.connect('database/test.db')
|
||
cursor = conn.cursor()
|
||
cursor.execute(f"DROP TABLE {table_name}")
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main()
|