import json from datetime import datetime, timezone as dt_timezone from io import StringIO from django.core.management import call_command from django.test import TestCase from django.utils import timezone from basic_info import models as basic_models from printing import models as printing_models class BackfillPrintingOrderKdRiQiCommandTest(TestCase): def setUp(self): self.merchant = basic_models.Merchant.objects.create( name='测试印染厂', type=basic_models.MerchantTypeEnum.FACTORY, ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户', ) def _create_order(self, *, external_order_id, external_raw, kd_riqi=None, merchant=None): return printing_models.PrintingOrder.objects.create( merchant=merchant or self.merchant, customer=self.customer, fabric='测试面料', width='160cm', external_order_id=external_order_id, external_raw=external_raw, kd_riqi=kd_riqi, ) def test_dry_run_reports_without_writing(self): order = self._create_order( external_order_id='KD001', external_raw={'first_record': {'KdRiQi': '2026-01-26T20:00:52Z'}}, ) output = StringIO() call_command('backfill_printing_order_kd_riqi', '--dry-run', stdout=output) order.refresh_from_db() payload = json.loads(output.getvalue()) self.assertIsNone(order.kd_riqi) self.assertEqual(payload['would_update'], 1) self.assertEqual(payload['updated'], 0) def test_backfills_missing_kd_riqi_from_external_raw(self): order = self._create_order( external_order_id='KD001', external_raw={'first_record': {'KdRiQi': '2026-01-26T20:00:52Z'}}, ) output = StringIO() call_command('backfill_printing_order_kd_riqi', stdout=output) order.refresh_from_db() payload = json.loads(output.getvalue()) self.assertEqual(timezone.localtime(order.kd_riqi).isoformat(), '2026-01-26T20:00:52+08:00') self.assertEqual(payload['updated'], 1) def test_skips_missing_and_invalid_values(self): missing = self._create_order( external_order_id='KD001', external_raw={'first_record': {}}, ) invalid = self._create_order( external_order_id='KD002', external_raw={'first_record': {'KdRiQi': 'invalid'}}, ) output = StringIO() call_command('backfill_printing_order_kd_riqi', stdout=output) missing.refresh_from_db() invalid.refresh_from_db() payload = json.loads(output.getvalue()) self.assertIsNone(missing.kd_riqi) self.assertIsNone(invalid.kd_riqi) self.assertEqual(payload['skipped_missing'], 1) self.assertEqual(payload['skipped_invalid'], 1) def test_filters_by_external_order_id(self): target = self._create_order( external_order_id='KD001', external_raw={'first_record': {'KdRiQi': '2026-01-26T20:00:52Z'}}, ) other = self._create_order( external_order_id='KD002', external_raw={'first_record': {'KdRiQi': '2026-01-27T20:00:52Z'}}, ) call_command('backfill_printing_order_kd_riqi', '--external-order-id', 'KD001', stdout=StringIO()) target.refresh_from_db() other.refresh_from_db() self.assertIsNotNone(target.kd_riqi) self.assertIsNone(other.kd_riqi) def test_overwrite_can_update_kd_riqi_and_outgoing_date(self): order = self._create_order( external_order_id='KD001', external_raw={'first_record': {'KdRiQi': '2026-01-26T20:00:52Z'}}, kd_riqi=datetime(2026, 1, 26, 20, 0, 52, tzinfo=dt_timezone.utc), ) order.outgoing_date = datetime(2026, 1, 26, 20, 0, 52, tzinfo=dt_timezone.utc) order.save(update_fields=['outgoing_date']) output = StringIO() call_command( 'backfill_printing_order_kd_riqi', '--overwrite', '--update-outgoing-date', stdout=output, ) order.refresh_from_db() payload = json.loads(output.getvalue()) self.assertEqual(timezone.localtime(order.kd_riqi).isoformat(), '2026-01-26T20:00:52+08:00') self.assertEqual(timezone.localtime(order.outgoing_date).isoformat(), '2026-01-26T20:00:52+08:00') self.assertEqual(payload['updated'], 1) self.assertTrue(payload['overwrite']) self.assertTrue(payload['update_outgoing_date'])