import unittest import psycopg2 from backend.api import ( registration, add_product_to_basket, buy_products, get_basket, get_histories, unregister, get_products, get_histories_with_products, add_money # Импортируем напрямую ) from backend.api import add_product_to_shop, delete_product_from_shop from backend.consts import * class TestFullUserFlow(unittest.TestCase): @classmethod def setUpClass(cls): cls.test_email = "test_user@example.com" cls.test_password = "secure_password" cls.test_nickname = "testuser" cls.token = None cls.product_ids = [] def test_01_register_user(self): """1. Регистрация тестового пользователя""" self.__class__.token = registration( nickname=self.test_nickname, password=self.test_password, email=self.test_email ) self.assertIsNotNone(self.token) print(f"✅ Пользователь зарегистрирован, токен: {self.token}") def test_02_add_test_products(self): """2. Добавление тестовых товаров в shop""" pid1 = add_product_to_shop( name="Тестовый товар 1", cost="50.00", count=10, reserved=0, description="Описание 1", type="test" ) pid2 = add_product_to_shop( name="Тестовый товар 2", cost="30.00", count=5, reserved=0, description="Описание 2", type="test" ) self.__class__.product_ids = [pid1, pid2] print(f"✅ Товары добавлены: {self.product_ids}") def test_03_add_to_basket(self): """3. Добавление товаров в корзину""" add_product_to_basket(self.token, self.product_ids[0]) add_product_to_basket(self.token, self.product_ids[0]) # ещё раз add_product_to_basket(self.token, self.product_ids[1]) print("✅ Товары добавлены в корзину") def test_04_check_shop_state_after_add_to_basket(self): """4. Проверка, что count и reserved изменились корректно""" conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: cur.execute("SELECT count, reserved FROM shop WHERE id = %s", (self.product_ids[0],)) count1, reserved1 = cur.fetchone() self.assertEqual(count1, 8) self.assertEqual(reserved1, 2) cur.execute("SELECT count, reserved FROM shop WHERE id = %s", (self.product_ids[1],)) count2, reserved2 = cur.fetchone() self.assertEqual(count2, 4) self.assertEqual(reserved2, 1) print("✅ Состояние магазина корректно после добавления в корзину") finally: cur.close() conn.close() def test_05_add_money_and_verify(self): """5. Добавляем деньги и проверяем баланс""" # Начальный баланс = 0 (из регистрации) initial_balance = self._get_user_balance() self.assertEqual(initial_balance, 0.0) # Добавляем 50.00 add_money(self.token, "150.00") # Проверяем новый баланс new_balance = self._get_user_balance() self.assertEqual(new_balance, 150.0) print("✅ Деньги добавлены, баланс = 150.00") def test_06_buy_products(self): """6. Покупка товаров (130.00)""" total_cost = 50.00 * 2 + 30.00 # 130.00 history_id = buy_products(self.token) self.assertIsInstance(history_id, int) print(f"✅ Покупка завершена, history_id: {history_id}") def test_07_check_user_money_deducted(self): """7. Проверка списания денег: 150 - 130 = 20""" balance = self._get_user_balance() self.assertAlmostEqual(balance, 20.0, places=2) print("✅ Деньги списаны корректно, остаток = 20.00") def test_08_check_shop_after_buy(self): """8. Проверка, что reserved = 0 после покупки""" conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: for pid in self.product_ids: cur.execute("SELECT reserved FROM shop WHERE id = %s", (pid,)) reserved = cur.fetchone()[0] self.assertEqual(reserved, 0) print("✅ reserved корректно сброшен после покупки") finally: cur.close() conn.close() def test_09_check_history_created(self): """9. Проверка создания истории покупок""" histories = get_histories(self.token) self.assertEqual(len(histories), 1) history = histories[0] self.assertEqual(history['user_id'], self._get_user_id()) self.assertEqual(history['products_cost'], 130.0) print("✅ История покупки создана") def test_10_get_history_with_products(self): """10. test_16_get_history_with_products""" conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: products = get_histories_with_products(self.token) print(products) finally: cur.close() conn.close() def test_11_check_basket_is_empty(self): """11. Проверка, что корзина пуста""" basket = get_basket(self.token) self.assertEqual(basket, '[]') print("✅ Корзина пуста") def test_12_unregister_user(self): """12. Удаление пользователя""" unregister(self.token) print("✅ Пользователь удалён") def test_13_check_user_and_related_data_deleted(self): """13. Проверка, что пользователь, корзина и история удалены""" conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: # Пользователь cur.execute("SELECT id FROM users WHERE token = %s", (self.token,)) self.assertIsNone(cur.fetchone()) user_id = self._get_user_id_before_delete() if user_id: # Корзина (CASCADE) cur.execute("SELECT user_id FROM basket WHERE user_id = %s", (user_id,)) self.assertIsNone(cur.fetchone()) # История (удалена вручную) cur.execute("SELECT id FROM history WHERE user_id = %s", (user_id,)) self.assertIsNone(cur.fetchone()) print("✅ Пользователь, корзина и история успешно удалены") finally: cur.close() conn.close() def test_14_delete_test_products(self): """14. Удаление тестовых товаров""" for pid in self.product_ids: delete_product_from_shop(pid) print("✅ Тестовые товары удалены") def test_15_check_products_deleted(self): """15. Проверка, что товары удалены""" conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: for pid in self.product_ids: cur.execute("SELECT id FROM shop WHERE id = %s", (pid,)) self.assertIsNone(cur.fetchone()) print("✅ Товары действительно удалены") finally: cur.close() conn.close() def test_16_check_products(self): """16. Проверка, что товары есть""" conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: products = get_products() print(products) self.assertIsNotNone(products) print("✅ Товары действительно есть") finally: cur.close() conn.close() # Вспомогательные методы def _get_user_balance(self) -> float: conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: cur.execute("SELECT money::NUMERIC FROM users WHERE token = %s", (self.token,)) return float(cur.fetchone()[0]) finally: cur.close() conn.close() def _get_user_id(self) -> int: conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: cur.execute("SELECT id FROM users WHERE token = %s", (self.token,)) row = cur.fetchone() return row[0] if row else None finally: cur.close() conn.close() def _get_user_id_before_delete(self) -> int: # После unregister токен недействителен, но мы можем попробовать найти по email conn = psycopg2.connect(**DB_CONFIG) cur = conn.cursor() try: cur.execute("SELECT id FROM users WHERE email = %s", (self.test_email,)) row = cur.fetchone() return row[0] if row else None finally: cur.close() conn.close() if __name__ == '__main__': unittest.main()