import asyncio from decimal import Decimal from unittest.mock import patch from django.test import SimpleTestCase from flower.utils.mingdaoyun import mappings from flower.utils.mingdaoyun.client import MingDaoYunClient from flower.utils.mingdaoyun.fetch import ( fetch_customers_from_mingdaoyun, fetch_plate_orders_from_mingdaoyun, fetch_products_from_mingdaoyun, fetch_row_by_rowid_from_mingdaoyun, sync_fabric_from_mingdaoyun, ) from flower.utils.mingdaoyun.models import Customer, Fabric, Product from flower.utils.mingdaoyun.parsers import pick_customer, pick_fabric, pick_product class _FakeMDYClient: def __init__(self, response): self.response = response self.calls = [] async def post(self, endpoint, data=None, params=None, headers=None): self.calls.append( { "endpoint": endpoint, "data": data, "params": params, "headers": headers, } ) return self.response class _AsyncCM: def __init__(self, obj): self.obj = obj async def __aenter__(self): return self.obj async def __aexit__(self, exc_type, exc, tb): return False class _FakeAiohttpResponse: def __init__(self, json_body): self._json_body = json_body self.raise_for_status_called = False def raise_for_status(self): self.raise_for_status_called = True async def json(self): return self._json_body class _FakeAiohttpSession: def __init__(self): self.post_calls = [] def post(self, url, json=None): resp = _FakeAiohttpResponse({"ok": True}) self.post_calls.append({"url": url, "json": json, "response": resp}) return _AsyncCM(resp) class _FakeAiohttpClientSession: def __init__(self, *args, **kwargs): self.headers = kwargs.get("headers") self.session = _FakeAiohttpSession() async def __aenter__(self): return self.session async def __aexit__(self, exc_type, exc, tb): return False class MingDaoYunParsersTestCase(SimpleTestCase): def test_pick_product_maps_fields_and_converts_numbers(self): row = { mappings.product_type_map["uid"]: "P-001", mappings.product_type_map["rowid"]: "row-1", mappings.product_type_map["created_at"]: "2025-12-01 00:00:00", mappings.product_type_map["name"]: "产品A", mappings.product_type_map["pieces"]: "12", mappings.product_type_map["segment_size"]: "3", mappings.product_type_map["unit"]: "米", mappings.product_type_map["color"]: "红", mappings.product_type_map["detail_str"]: '{"k": 1}', } product = pick_product(row) self.assertIsInstance(product, Product) self.assertEqual(product.uid, "P-001") self.assertEqual(product.rowid, "row-1") 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, Decimal("3")) # width 与 segment_size 当前共用同一个 controlId；保持同步代码现状 self.assertEqual(product.width, "3") self.assertEqual(product.unit, "米") self.assertEqual(product.color, "红") self.assertEqual(product.detail, {"k": 1}) def test_pick_product_invalid_numbers_become_none(self): row = { mappings.product_type_map["uid"]: "P-002", mappings.product_type_map["rowid"]: "row-2", mappings.product_type_map["created_at"]: "2025-12-01 00:00:00", mappings.product_type_map["name"]: "产品B", mappings.product_type_map["pieces"]: "invalid", mappings.product_type_map["segment_size"]: "", } product = pick_product(row) self.assertIsNone(product.pieces) self.assertIsNone(product.segment_size) def test_pick_customer_maps_fields(self): row = { mappings.customer_type_map["uid"]: "C-001", mappings.customer_type_map["rowid"]: "crow-1", mappings.customer_type_map["created_at"]: "2025-12-01 00:00:00", mappings.customer_type_map["name"]: "客户甲", mappings.customer_type_map["area"]: "杭州", } customer = pick_customer(row) self.assertIsInstance(customer, Customer) self.assertEqual(customer.uid, "C-001") self.assertEqual(customer.rowid, "crow-1") self.assertEqual(customer.created_at, "2025-12-01 00:00:00") self.assertEqual(customer.name, "客户甲") self.assertEqual(customer.area, "杭州") def test_pick_fabric_maps_fields(self): row = { mappings.fabric_type_map["name"]: "120克本白四面弹", } fabric = pick_fabric(row) self.assertIsInstance(fabric, Fabric) self.assertEqual(fabric.name, "120克本白四面弹") class MingDaoYunFetchTestCase(SimpleTestCase): def test_fetch_products_calls_client_with_expected_payload_and_parses(self): response = { "data": { "rows": [ { mappings.product_type_map["uid"]: "P-001", mappings.product_type_map["rowid"]: "row-1", mappings.product_type_map["created_at"]: "2025-12-01 00:00:00", mappings.product_type_map["name"]: "产品A", mappings.product_type_map["pieces"]: "1", mappings.product_type_map["segment_size"]: "2", } ], "total": 123, } } fake_client = _FakeMDYClient(response) with patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client): products, total = asyncio.run(fetch_products_from_mingdaoyun(page=2, page_size=5)) self.assertEqual(total, 123) self.assertEqual(len(products), 1) self.assertIsInstance(products[0], Product) self.assertEqual(products[0].uid, "P-001") self.assertEqual(len(fake_client.calls), 1) call = fake_client.calls[0] self.assertEqual(call["endpoint"], "/v2/open/worksheet/getFilterRows") self.assertEqual(call["data"]["worksheetId"], mappings.MDY_WORKSHEET_ID_PRODUCT) self.assertEqual(call["data"]["pageIndex"], 2) self.assertEqual(call["data"]["pageSize"], 5) self.assertEqual(call["data"]["sortId"], "ctime") self.assertIs(call["data"]["isAsc"], True) def test_fetch_customers_calls_client_with_expected_payload_and_parses(self): response = { "data": { "rows": [ { mappings.customer_type_map["uid"]: "C-001", mappings.customer_type_map["rowid"]: "crow-1", mappings.customer_type_map["created_at"]: "2025-12-01 00:00:00", mappings.customer_type_map["name"]: "客户甲", } ], "total": 9, } } fake_client = _FakeMDYClient(response) with patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client): customers, total = asyncio.run(fetch_customers_from_mingdaoyun(page=3, page_size=7)) self.assertEqual(total, 9) self.assertEqual(len(customers), 1) self.assertIsInstance(customers[0], Customer) self.assertEqual(customers[0].uid, "C-001") call = fake_client.calls[0] self.assertEqual(call["endpoint"], "/v2/open/worksheet/getFilterRows") self.assertEqual(call["data"]["worksheetId"], mappings.MDY_WORKSHEET_ID_CUSTOMER) self.assertEqual(call["data"]["pageIndex"], 3) self.assertEqual(call["data"]["pageSize"], 7) def test_fetch_plate_orders_returns_raw_rows_and_total(self): rows = [{"rowid": "r1", "ctime": "2025-12-17 10:07:00"}] response = {"data": {"rows": rows, "total": 100}} fake_client = _FakeMDYClient(response) with patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client): got_rows, total = asyncio.run( fetch_plate_orders_from_mingdaoyun(page=4, page_size=11, sort_id="ctime", is_asc=False) ) self.assertEqual(total, 100) self.assertEqual(got_rows, rows) call = fake_client.calls[0] self.assertEqual(call["data"]["worksheetId"], mappings.MDY_WORKSHEET_ID_PLATE_ORDER) self.assertEqual(call["data"]["pageIndex"], 4) self.assertEqual(call["data"]["pageSize"], 11) self.assertEqual(call["data"]["sortId"], "ctime") self.assertIs(call["data"]["isAsc"], False) def test_fetch_products_returns_empty_when_no_data(self): fake_client = _FakeMDYClient({}) with patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client): products, total = asyncio.run(fetch_products_from_mingdaoyun(page=1, page_size=1)) self.assertEqual(products, []) self.assertEqual(total, 0) def test_fetch_row_by_rowid_builds_filters_payload_and_returns_first_row(self): rowid = "row-xyz" row = { "rowid": rowid, mappings.product_type_map["uid"]: "P-ROW", } response = {"data": {"rows": [row]}} fake_client = _FakeMDYClient(response) with patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client): got = asyncio.run( fetch_row_by_rowid_from_mingdaoyun( worksheet_id=mappings.MDY_WORKSHEET_ID_PRODUCT, rowid=rowid, ) ) self.assertEqual(got, row) self.assertEqual(len(fake_client.calls), 1) call = fake_client.calls[0] self.assertEqual(call["endpoint"], "/v2/open/worksheet/getFilterRows") self.assertEqual(call["data"]["worksheetId"], mappings.MDY_WORKSHEET_ID_PRODUCT) self.assertEqual(call["data"]["listType"], 1) self.assertEqual(call["data"]["sortId"], "ctime") self.assertIs(call["data"]["isAsc"], False) self.assertIs(call["data"]["notGetTotal"], True) self.assertEqual( call["data"]["filters"], [ { "controlId": "rowId", "dataType": 2, "filterType": 3, "value": rowid, } ], ) def test_fetch_row_by_rowid_returns_none_when_no_data(self): fake_client = _FakeMDYClient({}) with patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client): got = asyncio.run( fetch_row_by_rowid_from_mingdaoyun( worksheet_id=mappings.MDY_WORKSHEET_ID_PRODUCT, rowid="missing", ) ) self.assertIsNone(got) class MingDaoYunFabricSyncTestCase(SimpleTestCase): def test_sync_fabric_calls_create_hook_and_returns_total(self): response = { "data": { "rows": [ {mappings.fabric_type_map["name"]: "布料A"}, {mappings.fabric_type_map["name"]: "布料B"}, ], "total": 2, } } fake_client = _FakeMDYClient(response) called = {} def fake_create(page, page_size, fabrics): called["page"] = page called["page_size"] = page_size called["fabrics"] = fabrics def fake_sync_to_async(func): async def _wrapper(*args, **kwargs): return func(*args, **kwargs) return _wrapper with ( patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client), patch("flower.utils.mingdaoyun.fetch._create_fabric_quick_inputs", side_effect=fake_create), patch("flower.utils.mingdaoyun.fetch.sync_to_async", side_effect=fake_sync_to_async), ): total = asyncio.run(sync_fabric_from_mingdaoyun(page=9, page_size=10)) self.assertEqual(total, 2) self.assertEqual(called["page"], 9) self.assertEqual(called["page_size"], 10) self.assertEqual([f.name for f in called["fabrics"]], ["布料A", "布料B"]) def test_sync_fabric_returns_zero_when_no_data(self): fake_client = _FakeMDYClient({}) called = {"hit": False} def fake_create(*_args, **_kwargs): called["hit"] = True def fake_sync_to_async(func): async def _wrapper(*args, **kwargs): return func(*args, **kwargs) return _wrapper with ( patch("flower.utils.mingdaoyun.fetch.get_default_mingdaoyun_client", return_value=fake_client), patch("flower.utils.mingdaoyun.fetch._create_fabric_quick_inputs", side_effect=fake_create), patch("flower.utils.mingdaoyun.fetch.sync_to_async", side_effect=fake_sync_to_async), ): total = asyncio.run(sync_fabric_from_mingdaoyun(page=1, page_size=1)) self.assertEqual(total, 0) self.assertIs(called["hit"], False) class MingDaoYunClientTestCase(SimpleTestCase): def test_client_post_merges_auth_params_and_passes_headers(self): created = {} def fake_client_session(*args, **kwargs): cs = _FakeAiohttpClientSession(*args, **kwargs) created["cs"] = cs return cs client = MingDaoYunClient(app_key="ak", sign="sg", base_url="https://api.mingdao.com") with patch("flower.utils.mingdaoyun.client.aiohttp.ClientSession", side_effect=fake_client_session): body = asyncio.run( client.post( endpoint="/v2/open/worksheet/getFilterRows", data={"worksheetId": "dummy", "pageIndex": 1}, headers={"X-Test": "1"}, ) ) self.assertEqual(body, {"ok": True}) cs = created["cs"] self.assertIsInstance(cs.headers, dict) self.assertEqual(cs.headers.get("Content-Type"), "application/json") self.assertEqual(cs.headers.get("X-Test"), "1") self.assertEqual(len(cs.session.post_calls), 1) call = cs.session.post_calls[0] self.assertEqual(call["url"], "https://api.mingdao.com/v2/open/worksheet/getFilterRows") self.assertEqual(call["json"]["worksheetId"], "dummy") self.assertEqual(call["json"]["pageIndex"], 1) self.assertEqual(call["json"]["appKey"], "ak") self.assertEqual(call["json"]["sign"], "sg") self.assertTrue(call["response"].raise_for_status_called)