import unittest import sqlite3 import os import sys import tempfile import json import time from datetime import datetime sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from database import get_db_connection, create_tables, calculate_discount_for_new_order, calculate_order_total, get_db from app import app class TestIntegration(unittest.TestCase): """Интеграционное тестирование всего приложения""" @classmethod def setUpClass(cls): """Создаем тестовую базу данных""" cls.db_fd, cls.db_path = tempfile.mkstemp() import database database.DATABASE = cls.db_path create_tables() cls.conn = sqlite3.connect(cls.db_path) cls.conn.row_factory = sqlite3.Row cursor = cls.conn.cursor() # Добавляем типы товаров cursor.executemany(''' INSERT INTO product_types (type_name, description) VALUES (?, ?) ''', [ ('Ручки', 'Пишущие принадлежности'), ('Тетради', 'Школьные тетради'), ('Бумага', 'Офисная бумага'), ('Техника', 'Дорогая техника') ]) # Добавляем партнеров (БЕЗ rating) cursor.executemany(''' INSERT INTO partners (name, director, email, phone, legal_address, inn) VALUES (?, ?, ?, ?, ?, ?) ''', [ ('ООО КанцПром', 'Иванов И.И.', 'info@kancprom.ru', '+7(495)123-45-67', 'г. Москва, ул. Промышленная', '7701123456'), ('ИП Петров', 'Петров А.С.', 'petrov@mail.ru', '+7(495)765-43-21', 'г. Москва, ул. Торговая', '7701765432') ]) # Добавляем товары (БЕЗ quantity_in_stock) cursor.executemany(''' INSERT INTO products (product_type_id, product_name, article, min_partner_price) VALUES (?, ?, ?, ?) ''', [ (1, 'Ручка синяя', 'R001', 15.50), (1, 'Ручка красная', 'R002', 12.00), (2, 'Тетрадь 48л', 'T001', 45.00), (2, 'Тетрадь 96л', 'T002', 85.00), (3, 'Бумага А4', 'B001', 280.00), (4, 'Принтер лазерный', 'P001', 15000.00) ]) cls.conn.commit() def test_db_connection(): conn = sqlite3.connect(cls.db_path) conn.row_factory = sqlite3.Row return conn app.config['DATABASE'] = cls.db_path database.get_db_connection = test_db_connection @classmethod def tearDownClass(cls): cls.conn.close() os.close(cls.db_fd) time.sleep(0.5) try: os.unlink(cls.db_path) except: pass def setUp(self): self.app = app.test_client() self.app.testing = True with sqlite3.connect(self.db_path) as conn: conn.execute("DELETE FROM order_items") conn.execute("DELETE FROM orders") def test_1_full_order_flow(self): """Тест полного цикла создания заказа""" order_data = { 'partner_id': 1, 'order_date': '2026-03-04', 'notes': 'Тестовый заказ', 'items': [ {'product_id': 1, 'quantity': 10, 'price': 15.50}, {'product_id': 3, 'quantity': 5, 'price': 45.00} ] } response = self.app.post('/api/order', data=json.dumps(order_data), content_type='application/json') self.assertEqual(response.status_code, 200) data = json.loads(response.data) self.assertTrue(data['success']) order_id = data['order_id'] with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row order = conn.execute('SELECT * FROM orders WHERE id = ?', (order_id,)).fetchone() self.assertIsNotNone(order) self.assertEqual(order['partner_id'], 1) items = conn.execute('SELECT * FROM order_items WHERE order_id = ?', (order_id,)).fetchall() self.assertEqual(len(items), 2) def test_2_discount_accumulation(self): """Тест накопления скидки""" order1_data = { 'partner_id': 2, 'order_date': '2026-03-04', 'notes': 'Первый заказ', 'items': [ {'product_id': 6, 'quantity': 7, 'price': 15000.00} # 7 * 15000 = 105 000 ] } response = self.app.post('/api/order', data=json.dumps(order1_data), content_type='application/json') self.assertEqual(response.status_code, 200) data = json.loads(response.data) order1_id = data['order_id'] order1_total = calculate_order_total(order1_id) self.assertEqual(order1_total['discount_percent'], 0) order2_data = { 'partner_id': 2, 'order_date': '2026-03-05', 'notes': 'Второй заказ', 'items': [ {'product_id': 6, 'quantity': 4, 'price': 15000.00} # 4 * 15000 = 60 000 (всего 165 000) ] } response = self.app.post('/api/order', data=json.dumps(order2_data), content_type='application/json') self.assertEqual(response.status_code, 200) data = json.loads(response.data) order2_id = data['order_id'] order2_total = calculate_order_total(order2_id) self.assertEqual(order2_total['discount_percent'], 15) def test_3_discount_reset_next_year(self): """Тест сброса скидки в новом году""" order_2026_data = { 'partner_id': 1, 'order_date': '2026-12-01', 'notes': 'Заказ 2026', 'items': [ {'product_id': 6, 'quantity': 14, 'price': 15000.00} # 14 * 15000 = 210 000 ] } response = self.app.post('/api/order', data=json.dumps(order_2026_data), content_type='application/json') self.assertEqual(response.status_code, 200) data = json.loads(response.data) order_2026_id = data['order_id'] order_2026_total = calculate_order_total(order_2026_id) self.assertEqual(order_2026_total['discount_percent'], 30) order_2027_data = { 'partner_id': 1, 'order_date': '2027-01-15', 'notes': 'Заказ 2027', 'items': [ {'product_id': 6, 'quantity': 7, 'price': 15000.00} # 7 * 15000 = 105 000 ] } response = self.app.post('/api/order', data=json.dumps(order_2027_data), content_type='application/json') self.assertEqual(response.status_code, 200) data = json.loads(response.data) order_2027_id = data['order_id'] order_2027_total = calculate_order_total(order_2027_id) self.assertEqual(order_2027_total['discount_percent'], 0) def test_4_negative_quantity_prevention(self): """Тест предотвращения отрицательного количества""" order_data = { 'partner_id': 1, 'order_date': '2026-03-04', 'items': [ {'product_id': 1, 'quantity': -5, 'price': 15.50} ] } response = self.app.post('/api/order', data=json.dumps(order_data), content_type='application/json') self.assertEqual(response.status_code, 400) data = json.loads(response.data) self.assertFalse(data['success']) def test_5_no_product_deletion_endpoint(self): """Тест отсутствия возможности удаления товара""" # Сначала создаем заказ, чтобы товар был в заказах order_data = { 'partner_id': 1, 'order_date': '2026-03-04', 'notes': 'Заказ для теста', 'items': [ {'product_id': 1, 'quantity': 5, 'price': 15.50} ] } response = self.app.post('/api/order', data=json.dumps(order_data), content_type='application/json') self.assertEqual(response.status_code, 200) # Пытаемся получить доступ к несуществующему маршруту удаления # Вместо DELETE используем GET, чтобы проверить, что маршрут не существует response = self.app.get('/api/product/1/delete') # Должен быть 404 (не найдено) self.assertEqual(response.status_code, 404) # Проверяем, что товар все еще существует в базе from database import get_db with get_db() as conn: product = conn.execute('SELECT * FROM products WHERE id = 1').fetchone() self.assertIsNotNone(product) # Проверяем, что API не поддерживает метод DELETE для товаров # Отправляем DELETE запрос на существующий маршрут /api/product/1 response = self.app.delete('/api/product/1') # Должен быть либо 405 (Method Not Allowed), либо 404 (Not Found) self.assertIn(response.status_code, [404, 405]) # Проверяем, что товар все еще существует with get_db() as conn: product = conn.execute('SELECT * FROM products WHERE id = 1').fetchone() self.assertIsNotNone(product) if __name__ == '__main__': unittest.main()