272 lines
10 KiB
Python
272 lines
10 KiB
Python
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() |