admin_api + tests

backend/admin_api.py

@@ -0,0 +1,62 @@
+import psycopg2
+
+from const import DB_CONFIG
+
+def add_product_to_shop(
+    name: str,
+    cost: str,           # строка, например "100.00" или "50"
+    count: int,
+    reserved: int = 0,   # по умолчанию 0
+    picture_url: str = "",
+    description: str = "",
+    type: str = ""
+) -> int:
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        # Вставка с автоматической генерацией id и возвратом id
+        cursor.execute("""
+            INSERT INTO shop (name, cost, count, reserved, picture_url, description, type)
+            VALUES (%s, %s::NUMERIC, %s, %s, %s, %s, %s)
+            RETURNING id
+        """, (name, cost, count, reserved, picture_url, description, type))
+
+        product_id = cursor.fetchone()[0]
+        connection.commit()
+        return product_id
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        connection.rollback()
+        raise
+    except Exception as e:
+        print("Ошибка при добавлении товара:", e)
+        connection.rollback()
+        raise
+    finally:
+        cursor.close()
+        connection.close()
+
+def delete_product_from_shop(product_id: int) -> bool:
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        # Удаляем товар по id
+        cursor.execute("DELETE FROM shop WHERE id = %s", (product_id,))
+
+        if cursor.rowcount == 0:
+            raise ValueError(f"Товар с id={product_id} не найден")
+
+        connection.commit()
+        return True
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        connection.rollback()
+        raise
+    except ValueError:
+        connection.rollback()
+        raise
+    finally:
+        cursor.close()
+        connection.close()

backend/api.py

@@ -5,6 +5,7 @@ import json
 from datetime import timedelta
 from psycopg2 import IntegrityError
 from collections import Counter
+from decimal import Decimal, InvalidOperation
 
 from utils import *
 from type import *
@@ -93,7 +94,7 @@ def registration(nickname: str, password: str, email: str) -> str:
     token_expiry_date = date.today() + token_live_time
     token_expiry_date_str = token_expiry_date.strftime("%Y.%m.%d")
 
-    money = "100"
+    money = "0"
     histories_id = "{}"
 
     connection = psycopg2.connect(**DB_CONFIG)
@@ -118,6 +119,51 @@ def registration(nickname: str, password: str, email: str) -> str:
     finally:
         cursor.close()
 
+def unregister(token: str) -> None:
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        if not check_token(cursor, token):
+            raise AuthError("Пользователь не авторизован")
+
+        user_id: int = get_user_id(cursor, token)
+
+        # Получаем массив histories_id пользователя
+        cursor.execute("SELECT histories_id FROM users WHERE id = %s", (user_id,))
+        row = cursor.fetchone()
+        if row is None:
+            raise ValueError("Пользователь не найден")
+
+        histories_id = row[0]  # Может быть None или список
+
+        # Удаляем все связанные истории, если они есть
+        if histories_id and len(histories_id) > 0:
+            # Используем ANY для удаления по массиву ID
+            cursor.execute("""
+                DELETE FROM history 
+                WHERE id = ANY(%s::BIGINT[])
+            """, (histories_id,))
+
+        # Удаляем пользователя (корзина удалится автоматически через CASCADE)
+        cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
+        if cursor.rowcount == 0:
+            raise ValueError("Не удалось удалить пользователя")
+
+        connection.commit()
+        print("Пользователь успешно удалён")
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        connection.rollback()
+        raise
+    except ValueError as e:
+        connection.rollback()
+        raise
+
+    finally:
+        cursor.close()
+        connection.close()
+
 def get_products() -> list[dict]:
     connection = psycopg2.connect(**DB_CONFIG)
     cursor = connection.cursor()
@@ -170,6 +216,9 @@ def get_basket(token: str):
 
         products_id = res[0]  # Это список, например: [1, 1, 3]
 
+        if len(products_id) == 0:
+            return json.dumps([])
+
         # Распаковываем массив и соединяем каждый элемент с таблицей shop
         cursor.execute("""
             SELECT json_agg(row_to_json(s)) AS result
@@ -419,17 +468,34 @@ def buy_products(token: str) -> int:
 
         # 2. Получаем стоимость товаров и проверяем их наличие
         product_counts = Counter(products_id)
-        total_cost = 0
+        total_cost = Decimal('0.00')
 
         for product_id in product_counts:
             cursor.execute("SELECT cost::NUMERIC FROM shop WHERE id = %s", (product_id,))
             shop_row = cursor.fetchone()
             if shop_row is None:
                 raise ValueError(f"Товар с id={product_id} не найден")
-            cost = shop_row[0]
+            cost = shop_row[0]  # Decimal
             total_cost += cost * product_counts[product_id]
 
-        # 3. Вставляем запись в history и получаем её id
+        # 3. Проверяем баланс пользователя
+        cursor.execute("SELECT money::NUMERIC FROM users WHERE id = %s", (user_id,))
+        user_row = cursor.fetchone()
+        if user_row is None:
+            raise ValueError("Пользователь не найден")
+        user_money = user_row[0]  # Decimal
+
+        if user_money < total_cost:
+            raise ValueError("Недостаточно средств на счету")
+
+        # 4. Списываем деньги
+        new_balance = user_money - total_cost
+        cursor.execute(
+            "UPDATE users SET money = %s WHERE id = %s",
+            (new_balance, user_id)
+        )
+
+        # 5. Вставляем запись в history и получаем её id
         cursor.execute("""
             INSERT INTO history (user_id, products_id, products_cost, date)
             VALUES (%s, %s, %s, CURRENT_DATE)
@@ -438,7 +504,7 @@ def buy_products(token: str) -> int:
 
         history_id = cursor.fetchone()[0]  # Получаем сгенерированный id
 
-        # 4. Добавляем history_id в users.histories_id
+        # 6. Добавляем history_id в users.histories_id
         cursor.execute("""
             UPDATE users 
             SET histories_id = array_append(COALESCE(histories_id, '{}'), %s)
@@ -448,7 +514,7 @@ def buy_products(token: str) -> int:
         if cursor.rowcount == 0:
             raise ValueError("Не удалось обновить историю пользователя")
 
-        # 5. Уменьшаем reserved в shop для каждого купленного товара
+        # 7. Уменьшаем reserved в shop для каждого купленного товара
         for product_id, quantity in product_counts.items():
             cursor.execute("""
                 UPDATE shop 
@@ -457,7 +523,6 @@ def buy_products(token: str) -> int:
             """, (quantity, product_id, quantity))
 
             if cursor.rowcount == 0:
-                # Проверяем, существует ли товар
                 cursor.execute("SELECT id FROM shop WHERE id = %s", (product_id,))
                 if cursor.fetchone() is None:
                     raise ValueError(f"Товар с id={product_id} не найден")
@@ -467,11 +532,54 @@ def buy_products(token: str) -> int:
                         f"попытка списать {quantity}, но reserved < {quantity}"
                     )
 
-        # 6. Очищаем корзину (products_id = пустой массив)
+        # 8. Очищаем корзину (products_id = пустой массив)
         cursor.execute("UPDATE basket SET products_id = '{}' WHERE user_id = %s", (user_id,))
 
         connection.commit()
-        return history_id  # Можно вернуть id заказа, если нужно
+        return history_id
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        connection.rollback()
+        raise
+    except ValueError:
+        connection.rollback()
+        raise
+
+    finally:
+        cursor.close()
+        connection.close()
+
+def add_money(token: str, money: str) -> None:
+    # Валидация и преобразование входной суммы
+    try:
+        # Удаляем возможные пробелы и заменяем запятую на точку (для удобства)
+        money_clean = money.strip().replace(',', '.')
+        amount = Decimal(money_clean)
+        if amount < 0:
+            raise ValueError("Сумма должна быть неотрицательной")
+    except (InvalidOperation, ValueError) as e:
+        raise ValueError("Некорректный формат суммы") from e
+
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        if not check_token(cursor, token):
+            raise AuthError("Пользователь не авторизован")
+
+        user_id: int = get_user_id(cursor, token)
+
+        # Обновляем баланс: money = money + amount
+        cursor.execute("""
+            UPDATE users 
+            SET money = money + %s
+            WHERE id = %s
+        """, (amount, user_id))
+
+        if cursor.rowcount == 0:
+            raise ValueError("Пользователь не найден")
+
+        connection.commit()
 
     except psycopg2.Error as e:
         print("Ошибка при работе с PostgreSQL:", e)

backend/main.py

@@ -21,8 +21,11 @@ if __name__ == "__main__":
     # print(get_history(token))
 
     token = login("vasili_pupkin@gmail.com", "Vasya2005")
-    # add_product_to_basket(token, 1)
-    # add_product_to_basket(token, 1)
-    # add_product_to_basket(token, 1)
-    # add_product_to_basket(token, 2)
-    print(get_histories(token))
+    # add_product_to_basket(token, 3)
+    # add_product_to_basket(token, 3)
+    # add_product_to_basket(token, 5)
+    # add_product_to_basket(token, 5)
+    add_money()
+    # buy_products(token)
+    # print(get_histories(token))
+    # unregister(token)

backend/tests.py

@@ -0,0 +1,222 @@
+import unittest
+import psycopg2
+import json  # Используем json.loads вместо eval
+from api import (
+    registration,
+    add_product_to_basket,
+    buy_products,
+    get_basket,
+    get_histories,
+    unregister,
+    add_money  # Импортируем напрямую
+)
+from admin_api import add_product_to_shop, delete_product_from_shop
+from const import DB_CONFIG
+
+
+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_check_basket_is_empty(self):
+        """10. Проверка, что корзина пуста"""
+        basket = get_basket(self.token)
+        self.assertEqual(basket, '[]')
+        print("✅ Корзина пуста")
+
+    def test_11_unregister_user(self):
+        """11. Удаление пользователя"""
+        unregister(self.token)
+        print("✅ Пользователь удалён")
+
+    def test_12_check_user_and_related_data_deleted(self):
+        """12. Проверка, что пользователь, корзина и история удалены"""
+        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_13_delete_test_products(self):
+        """13. Удаление тестовых товаров"""
+        for pid in self.product_ids:
+            delete_product_from_shop(pid)
+        print("✅ Тестовые товары удалены")
+
+    def test_14_check_products_deleted(self):
+        """14. Проверка, что товары удалены"""
+        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 _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()