diff --git a/api_v1/tasks.py b/api_v1/tasks.py index 2c3898a..5ce3e15 100644 --- a/api_v1/tasks.py +++ b/api_v1/tasks.py @@ -193,9 +193,9 @@ def _upsert_product(product_data, merchant, category): pieces_int = _ensure_int(product_data.pieces) if pieces_int is not None: defaults['pieces'] = pieces_int - segment_int = _ensure_int(product_data.segment_size) - if segment_int is not None: - defaults['segment_size'] = segment_int + segment_decimal = _ensure_decimal(product_data.segment_size) + if segment_decimal is not None: + defaults['segment_size'] = segment_decimal product_obj, created = basic_models.Product.objects.get_or_create( merchant=merchant, diff --git a/api_v1/test_mingdaoyun_utils.py b/api_v1/test_mingdaoyun_utils.py index 29c2521..4a982e5 100644 --- a/api_v1/test_mingdaoyun_utils.py +++ b/api_v1/test_mingdaoyun_utils.py @@ -1,4 +1,5 @@ import asyncio +from decimal import Decimal from unittest.mock import patch from django.test import SimpleTestCase @@ -99,7 +100,7 @@ class MingDaoYunParsersTestCase(SimpleTestCase): self.assertEqual(product.created_at, "2025-12-01 00:00:00") self.assertEqual(product.name, "产品A") self.assertEqual(product.pieces, 12) - self.assertEqual(product.segment_size, 3) + self.assertEqual(product.segment_size, Decimal("3")) # width 与 segment_size 当前共用同一个 controlId;保持同步代码现状 self.assertEqual(product.width, "3") self.assertEqual(product.unit, "米") diff --git a/basic_info/migrations/0023_alter_product_segment_size.py b/basic_info/migrations/0023_alter_product_segment_size.py new file mode 100644 index 0000000..69fdbda --- /dev/null +++ b/basic_info/migrations/0023_alter_product_segment_size.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.8 on 2025-12-27 06:16 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('basic_info', '0022_product_pieces_product_segment_size'), + ] + + operations = [ + migrations.AlterField( + model_name='product', + name='segment_size', + field=models.DecimalField(blank=True, decimal_places=2, max_digits=8, null=True, verbose_name='一段尺寸'), + ), + ] diff --git a/basic_info/models.py b/basic_info/models.py index 478d0eb..1fc640c 100644 --- a/basic_info/models.py +++ b/basic_info/models.py @@ -229,7 +229,7 @@ class Product(ModelBase): color = models.CharField(max_length=50, blank=True, null=True, verbose_name='颜色') width_size = models.DecimalField(max_digits=10, decimal_places=2, blank=True, null=True, verbose_name='宽幅') pieces = models.IntegerField(blank=True, null=True, verbose_name='件数') - segment_size = models.IntegerField(blank=True, null=True, verbose_name='一段尺寸') + segment_size = models.DecimalField(blank=True, null=True, verbose_name='一段尺寸', decimal_places=2, max_digits=8) single_price_in = models.DecimalField( max_digits=10, decimal_places=2, diff --git a/business/tests.py b/business/tests.py index 25c2e51..4501169 100644 --- a/business/tests.py +++ b/business/tests.py @@ -4,7 +4,7 @@ from concurrent.futures import ThreadPoolExecutor from django.contrib.auth import get_user_model from django.test import TestCase, TransactionTestCase, override_settings -from django.db import connections +from django.db import connections, close_old_connections from django.utils import timezone from unittest.mock import patch, MagicMock @@ -1240,11 +1240,17 @@ class SalesOrderConcurrencyTestCase(TransactionTestCase): def test_concurrent_sales_order_approval_updates_balance_once(self): def approve(): - services.review_sales_order( - sales_order_id=self.sales_order.id, - target_status=business_models.SalesOrderStatusEnum.APPROVED, - reviewed_by=self.user, - ) + # ThreadPoolExecutor 会复用线程;Django 的 DB connection 是线程局部的, + # 若不显式关闭,可能导致测试 DB 在 teardown 时仍被占用,无法 DROP。 + close_old_connections() + try: + services.review_sales_order( + sales_order_id=self.sales_order.id, + target_status=business_models.SalesOrderStatusEnum.APPROVED, + reviewed_by=self.user, + ) + finally: + connections.close_all() with ThreadPoolExecutor(max_workers=2) as executor: futures = [executor.submit(approve) for _ in range(2)] diff --git a/flower/utils/mingdaoyun/models.py b/flower/utils/mingdaoyun/models.py index 1e844d3..a725c2f 100644 --- a/flower/utils/mingdaoyun/models.py +++ b/flower/utils/mingdaoyun/models.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +from decimal import Decimal from typing import Any, List from pydantic import BaseModel, Field, computed_field @@ -13,7 +14,7 @@ class Product(BaseModel): rowid: str name: str pieces: int | None - segment_size: int | None + segment_size: Decimal | None unit: str | None color: str | None width: str | None = None diff --git a/flower/utils/mingdaoyun/parsers.py b/flower/utils/mingdaoyun/parsers.py index b780f1c..7575e8c 100644 --- a/flower/utils/mingdaoyun/parsers.py +++ b/flower/utils/mingdaoyun/parsers.py @@ -1,5 +1,6 @@ from __future__ import annotations +from decimal import Decimal, InvalidOperation from typing import Any from .mappings import customer_type_map, fabric_type_map, product_type_map @@ -26,8 +27,16 @@ def pick_product(fields: dict[str, Any]) -> Product: except (TypeError, ValueError): return None + def _to_decimal(value): + if value in ("", None): + return None + try: + return Decimal(str(value)) + except (InvalidOperation, TypeError, ValueError): + return None + data["pieces"] = _to_int(data.get("pieces")) - data["segment_size"] = _to_int(data.get("segment_size")) + data["segment_size"] = _to_decimal(data.get("segment_size")) return Product(**data) diff --git a/sse/tests.py b/sse/tests.py deleted file mode 100644 index b97250c..0000000 --- a/sse/tests.py +++ /dev/null @@ -1,413 +0,0 @@ -import json -import time -import threading -from queue import Queue -from unittest.mock import patch, MagicMock - -from django.test import TestCase -from django.contrib.auth import get_user_model -from django.urls import reverse -from rest_framework import status -from rest_framework.test import APIClient -from rest_framework_simplejwt.tokens import RefreshToken - -from basic_info import models as basic_models -from . import services - -User = get_user_model() - - -class SSEAPITestCase(TestCase): - """SSE API测试用例""" - - def setUp(self): - """设置测试数据""" - # 创建测试商户 - self.merchant1 = basic_models.Merchant.objects.create( - name='测试商户1', - type=basic_models.MerchantTypeEnum.STORE, - ) - self.merchant2 = basic_models.Merchant.objects.create( - name='测试商户2', - type=basic_models.MerchantTypeEnum.FACTORY, - ) - - # 创建测试用户 - self.user1 = User.objects.create_user(username='user1', password='testpass') - self.user2 = User.objects.create_user(username='user2', password='testpass') - self.user_no_employee = User.objects.create_user(username='no_employee', password='testpass') - - # 创建员工并关联商户 - self.employee1 = basic_models.Employee.objects.create( - merchant=self.merchant1, - sys_user=self.user1, - name='员工1', - mobile='13800138001', - ) - self.employee2 = basic_models.Employee.objects.create( - merchant=self.merchant2, - sys_user=self.user2, - name='员工2', - mobile='13800138002', - ) - - # 生成JWT token - refresh1 = RefreshToken.for_user(self.user1) - self.token1 = str(refresh1.access_token) - - refresh2 = RefreshToken.for_user(self.user2) - self.token2 = str(refresh2.access_token) - - refresh_no_employee = RefreshToken.for_user(self.user_no_employee) - self.token_no_employee = str(refresh_no_employee.access_token) - - # 设置客户端 - self.client = APIClient() - - # 清理所有连接 - services._connections.clear() - - def test_sse_connection_without_token(self): - """测试没有JWT token的SSE连接""" - response = self.client.get('/sse/') - - self.assertEqual(response.status_code, 403) - self.assertEqual(response.content, b'Authentication failed or user has no associated merchant') - - # 检查连接未被添加到服务中 - self.assertEqual(len(services._connections), 0) - - def test_sse_connection_with_invalid_token(self): - """测试无效JWT token的SSE连接""" - response = self.client.get( - '/sse/', - HTTP_AUTHORIZATION='Bearer invalid_token' - ) - - self.assertEqual(response.status_code, 403) - self.assertEqual(response.content, b'Authentication failed or user has no associated merchant') - - # 检查连接未被添加到服务中 - self.assertEqual(len(services._connections), 0) - - def test_sse_connection_user_no_employee(self): - """测试用户无关联员工的情况""" - response = self.client.get( - '/sse/', - HTTP_AUTHORIZATION=f'Bearer {self.token_no_employee}' - ) - - self.assertEqual(response.status_code, 403) - self.assertEqual(response.content, b'Authentication failed or user has no associated merchant') - - # 检查连接未被添加到服务中 - self.assertEqual(len(services._connections), 0) - - def test_push_test_event_to_merchant(self): - """测试向特定商户推送测试事件""" - # 手动建立连接 - queue1 = Queue() - services.push_connection(self.merchant1.id, queue1) - - # 推送事件 - response = self.client.post( - '/sse/push/', - HTTP_AUTHORIZATION=f'Bearer {self.token1}' - ) - - self.assertEqual(response.status_code, 200) - data = response.json() - self.assertEqual(data['status'], 'ok') - self.assertEqual(data['message'], 'Test event broadcasted to your merchant') - self.assertEqual(data['merchant_id'], self.merchant1.id) - self.assertEqual(data['clients'], 1) - - # 清理连接 - services.remove_connection(self.merchant1.id, queue1) - - def test_push_test_event_unauthenticated(self): - """测试未认证用户推送事件""" - response = self.client.post('/sse/push/') - - self.assertEqual(response.status_code, 401) - - def test_push_test_event_user_no_employee(self): - """测试无关联员工用户推送事件""" - response = self.client.post( - '/sse/push/', - HTTP_AUTHORIZATION=f'Bearer {self.token_no_employee}' - ) - - self.assertEqual(response.status_code, 403) - data = response.json() - self.assertEqual(data['error'], 'User has no associated merchant') - - def test_get_sse_status(self): - """测试获取SSE状态""" - # 手动建立两个商户的连接 - queue1 = Queue() - queue2 = Queue() - services.push_connection(self.merchant1.id, queue1) - services.push_connection(self.merchant2.id, queue2) - - # 获取状态 - response = self.client.get( - '/sse/status/', - HTTP_AUTHORIZATION=f'Bearer {self.token1}' - ) - - self.assertEqual(response.status_code, 200) - data = response.json() - self.assertEqual(data['status'], 'running') - self.assertEqual(data['total_clients'], 2) # 所有连接数 - self.assertEqual(data['merchant_clients'], 1) # 当前商户的连接数 - self.assertEqual(data['merchant_id'], self.merchant1.id) - - # 清理连接 - services.remove_connection(self.merchant1.id, queue1) - services.remove_connection(self.merchant2.id, queue2) - - def test_get_sse_status_unauthenticated(self): - """测试未认证用户获取状态""" - response = self.client.get('/sse/status/') - - self.assertEqual(response.status_code, 401) - - def test_shutdown_merchant_connections(self): - """测试关闭商户连接""" - # 手动建立两个商户的连接 - queue1 = Queue() - queue2 = Queue() - services.push_connection(self.merchant1.id, queue1) - services.push_connection(self.merchant2.id, queue2) - - # 关闭商户1的连接 - response = self.client.post( - '/sse/shutdown/', - HTTP_AUTHORIZATION=f'Bearer {self.token1}' - ) - - self.assertEqual(response.status_code, 200) - data = response.json() - self.assertEqual(data['status'], 'ok') - self.assertEqual(data['message'], 'Your merchant\'s SSE connections have been closed') - self.assertEqual(data['merchant_id'], self.merchant1.id) - self.assertEqual(data['clients'], 1) - - # 检查连接状态 - merchant1_connections = services.get_merchant_connections(self.merchant1.id) - merchant2_connections = services.get_merchant_connections(self.merchant2.id) - self.assertEqual(len(merchant1_connections), 0) - self.assertEqual(len(merchant2_connections), 1) - - # 清理连接 - services.remove_connection(self.merchant2.id, queue2) - - def test_shutdown_merchant_connections_unauthenticated(self): - """测试未认证用户关闭连接""" - response = self.client.post('/sse/shutdown/') - - self.assertEqual(response.status_code, 401) - - def test_merchant_isolation(self): - """测试商户隔离""" - # 为两个商户分别建立连接 - queue1 = Queue() - queue2 = Queue() - queue3 = Queue() - queue4 = Queue() - - services.push_connection(self.merchant1.id, queue1) - services.push_connection(self.merchant1.id, queue2) - services.push_connection(self.merchant2.id, queue3) - services.push_connection(self.merchant2.id, queue4) - - # 向商户1推送事件 - response = self.client.post( - '/sse/push/', - HTTP_AUTHORIZATION=f'Bearer {self.token1}' - ) - - self.assertEqual(response.status_code, 200) - data = response.json() - self.assertEqual(data['merchant_id'], self.merchant1.id) - self.assertEqual(data['clients'], 2) # 商户1有2个客户端 - - # 向商户2推送事件 - response = self.client.post( - '/sse/push/', - HTTP_AUTHORIZATION=f'Bearer {self.token2}' - ) - - self.assertEqual(response.status_code, 200) - data = response.json() - self.assertEqual(data['merchant_id'], self.merchant2.id) - self.assertEqual(data['clients'], 2) # 商户2有2个客户端 - - # 确认商户隔离 - merchant1_connections = services.get_merchant_connections(self.merchant1.id) - merchant2_connections = services.get_merchant_connections(self.merchant2.id) - self.assertEqual(len(merchant1_connections), 2) - self.assertEqual(len(merchant2_connections), 2) - - # 清理连接 - services.remove_connection(self.merchant1.id, queue1) - services.remove_connection(self.merchant1.id, queue2) - services.remove_connection(self.merchant2.id, queue3) - services.remove_connection(self.merchant2.id, queue4) - - def test_sse_connection_with_options_request(self): - """测试OPTIONS预检请求""" - response = self.client.options('/sse/') - - # OPTIONS请求应该成功 - self.assertEqual(response.status_code, 200) - - # 带Origin头的OPTIONS请求应该返回CORS头 - response_with_origin = self.client.options( - '/sse/', - HTTP_ORIGIN='https://example.com' - ) - self.assertEqual(response_with_origin.status_code, 200) - self.assertIn('Access-Control-Allow-Origin', response_with_origin) - self.assertIn('Access-Control-Allow-Methods', response_with_origin) - self.assertIn('Access-Control-Allow-Headers', response_with_origin) - - -class SSEServicesTestCase(TestCase): - """SSE服务层测试用例""" - - def setUp(self): - """设置测试数据""" - self.merchant1_id = 1 - self.merchant2_id = 2 - - # 清理所有连接 - services._connections.clear() - - def test_push_connection(self): - """测试添加连接""" - queue1 = Queue() - queue2 = Queue() - - # 添加连接 - services.push_connection(self.merchant1_id, queue1) - services.push_connection(self.merchant1_id, queue2) - services.push_connection(self.merchant2_id, Queue()) - - # 检查连接 - merchant1_connections = services.get_merchant_connections(self.merchant1_id) - merchant2_connections = services.get_merchant_connections(self.merchant2_id) - - self.assertEqual(len(merchant1_connections), 2) - self.assertEqual(len(merchant2_connections), 1) - self.assertIn(queue1, merchant1_connections) - self.assertIn(queue2, merchant1_connections) - - def test_remove_connection(self): - """测试移除连接""" - queue1 = Queue() - queue2 = Queue() - - # 添加连接 - services.push_connection(self.merchant1_id, queue1) - services.push_connection(self.merchant1_id, queue2) - - # 移除一个连接 - services.remove_connection(self.merchant1_id, queue1) - - # 检查连接 - merchant1_connections = services.get_merchant_connections(self.merchant1_id) - self.assertEqual(len(merchant1_connections), 1) - self.assertNotIn(queue1, merchant1_connections) - self.assertIn(queue2, merchant1_connections) - - def test_remove_all_merchant_connections(self): - """测试移除商户所有连接""" - queue1 = Queue() - queue2 = Queue() - - # 添加连接 - services.push_connection(self.merchant1_id, queue1) - services.push_connection(self.merchant1_id, queue2) - - # 移除所有连接 - services.remove_connection(self.merchant1_id, queue1) - services.remove_connection(self.merchant1_id, queue2) - - # 检查连接 - merchant1_connections = services.get_merchant_connections(self.merchant1_id) - self.assertEqual(len(merchant1_connections), 0) - - # 商户记录应该被移除 - self.assertNotIn(self.merchant1_id, services._connections) - - def test_get_all_connections_count(self): - """测试获取所有连接数""" - # 添加连接 - services.push_connection(self.merchant1_id, Queue()) - services.push_connection(self.merchant1_id, Queue()) - services.push_connection(self.merchant2_id, Queue()) - - # 检查总连接数 - total = services.get_all_connections_count() - self.assertEqual(total, 3) - - def test_push_event_to_merchant(self): - """测试向特定商户推送事件""" - queue1 = Queue() - queue2 = Queue() - queue3 = Queue() - - # 添加连接 - services.push_connection(self.merchant1_id, queue1) - services.push_connection(self.merchant1_id, queue2) - services.push_connection(self.merchant2_id, queue3) - - # 向商户1推送事件 - event_data = {'type': 'test', 'message': 'test message'} - services.push_sse_event_to_merchant(self.merchant1_id, event_data) - - # 检查消息 - self.assertEqual(queue1.qsize(), 1) - self.assertEqual(queue2.qsize(), 1) - self.assertEqual(queue3.qsize(), 0) # 商户2不应该收到消息 - - # 检查消息内容 - self.assertEqual(queue1.get_nowait(), event_data) - self.assertEqual(queue2.get_nowait(), event_data) - - def test_push_event_to_nonexistent_merchant(self): - """测试向不存在的商户推送事件""" - # 向不存在的商户推送事件 - event_data = {'type': 'test', 'message': 'test message'} - services.push_sse_event_to_merchant(999, event_data) - - # 不应该抛出异常,也不会有连接受到影响 - self.assertEqual(len(services._connections), 0) - - def test_push_simple_message_to_merchant(self): - """测试向特定商户推送简单消息""" - queue1 = Queue() - - # 添加连接 - services.push_connection(self.merchant1_id, queue1) - - # 推送简单消息 - services.push_simple_message_with_object_id_to_merchant( - self.merchant1_id, - 'order_paid', - '订单已支付', - 12345 - ) - - # 检查消息 - self.assertEqual(queue1.qsize(), 1) - - # 检查消息内容 - message = queue1.get_nowait() - self.assertEqual(message['mode'], 'simple_message') - self.assertEqual(message['type'], 'order_paid') - self.assertEqual(message['message'], '订单已支付') - self.assertEqual(message['object_id'], 12345) - self.assertEqual(message['merchant_id'], self.merchant1_id) \ No newline at end of file