feat: change segment_size to decimal in Product model

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)
+    segment_decimal = _ensure_decimal(product_data.segment_size)
-    if segment_int is not None:
+    if segment_decimal is not None:
-        defaults['segment_size'] = segment_int
+        defaults['segment_size'] = segment_decimal
 
     product_obj, created = basic_models.Product.objects.get_or_create(
         merchant=merchant,

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, "米")

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='一段尺寸'),
+        ),
+    ]

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,

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(
+            # ThreadPoolExecutor 会复用线程；Django 的 DB connection 是线程局部的，
-                sales_order_id=self.sales_order.id,
+            # 若不显式关闭，可能导致测试 DB 在 teardown 时仍被占用，无法 DROP。
-                target_status=business_models.SalesOrderStatusEnum.APPROVED,
+            close_old_connections()
-                reviewed_by=self.user,
+            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)]

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

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)
 

sse/tests.py

@@ -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)