zkxw/backend/tests/test_production_cascade_complete.py

"""生产完成任务/工单状态级联。"""
import unittest
from datetime import datetime
from unittest.mock import ANY, MagicMock, patch

from pydantic import ValidationError

from app.crud import production_task as production_task_svc
from app.crud import production_work_order as production_work_order_svc
from app.schemas.production_work_order import ProductionWorkOrderComplete


class TestProductionWorkOrderCompleteSchema(unittest.TestCase):
    def test_requires_conclusion(self):
        with self.assertRaises(ValidationError):
            ProductionWorkOrderComplete()

    def test_accepts_cg_and_remark(self):
        body = ProductionWorkOrderComplete(conclusion="cg", remark="ok")
        self.assertEqual(body.conclusion, "cg")
        self.assertEqual(body.remark, "ok")


class TestForceCompleteTaskSteps(unittest.TestCase):
    def test_marks_incomplete_steps_ywc(self):
        db = MagicMock()
        step_dzx = MagicMock(step_status="dzx", completed_by=None, completed_time=None)
        step_jxz = MagicMock(step_status="jxz", completed_by=None, completed_time=None)
        q = MagicMock()
        db.query.return_value = q
        q.filter.return_value = q
        q.all.return_value = [step_dzx, step_jxz]
        now = datetime(2026, 8, 31, 12, 0, 0)

        production_task_svc._force_complete_task_steps(
            db, "task-1", "t1", "user-1", now
        )

        self.assertEqual(step_dzx.step_status, "ywc")
        self.assertEqual(step_dzx.completed_by, "user-1")
        self.assertEqual(step_dzx.completed_time, now)
        self.assertEqual(step_jxz.step_status, "ywc")
        db.add.assert_any_call(step_dzx)
        db.commit.assert_not_called()


class TestUpdateTaskStatusCascadeSteps(unittest.TestCase):
    @patch("app.crud.production_task._consume_task_input_cell_products")
    @patch("app.crud.production_work_order.sync_production_work_order_status_from_tasks")
    @patch("app.crud.production_task._force_complete_task_steps")
    def test_ywc_no_longer_blocks_on_incomplete_steps(
        self, mock_force, mock_sync, mock_consume
    ):
        db = MagicMock()
        task = MagicMock(
            id="task-1",
            tenant_no="t1",
            del_flag=0,
            task_status="jxz",
            work_order_id="wo-1",
            started_time=None,
            completed_time=None,
            conclusion=None,
            remark=None,
        )
        q = MagicMock()
        db.query.return_value = q
        q.filter.return_value = q
        q.first.return_value = task

        result = production_task_svc.update_task_status(
            db, "task-1", "t1", "user-1", "ywc", conclusion="cg"
        )

        mock_force.assert_called_once()
        self.assertEqual(task.task_status, "ywc")
        self.assertEqual(task.conclusion, "cg")
        mock_consume.assert_called_once()
        mock_sync.assert_called_once()
        db.commit.assert_called_once()
        self.assertIs(result, task)


class TestCompleteProductionWorkOrderCascade(unittest.TestCase):
    @patch("app.crud.pilot_outbound_status_sync.sync_schedule_from_production_work_order")
    @patch("app.crud.production_task._consume_task_input_cell_products")
    @patch("app.crud.production_task._force_complete_task_steps")
    @patch("app.crud.production_work_order.get_production_work_order")
    def test_cascades_incomplete_tasks_and_steps(
        self, mock_get_wo, mock_force, mock_consume, mock_sync_sched
    ):
        db = MagicMock()
        wo = MagicMock(id="wo-1", work_order_status="jxz", remark=None)
        mock_get_wo.return_value = wo

        task_a = MagicMock(
            id="ta",
            task_status="dzx",
            completed_time=None,
            conclusion=None,
            remark=None,
        )
        q = MagicMock()
        db.query.return_value = q
        q.filter.return_value = q
        q.all.return_value = [task_a]

        body = ProductionWorkOrderComplete(conclusion="sb", remark="batch fail")
        result = production_work_order_svc.complete_production_work_order(
            db, "wo-1", "t1", "user-1", body
        )

        mock_force.assert_called_once_with(db, "ta", "t1", "user-1", ANY)
        self.assertEqual(task_a.task_status, "ywc")
        self.assertEqual(task_a.conclusion, "sb")
        self.assertEqual(task_a.remark, "batch fail")
        mock_consume.assert_called_once()
        self.assertEqual(wo.work_order_status, "ywc")
        self.assertEqual(wo.remark, "batch fail")
        mock_sync_sched.assert_called_once()
        db.commit.assert_called_once()
        self.assertIs(result, wo)

    @patch("app.crud.production_work_order.get_production_work_order")
    def test_rejects_invalid_conclusion(self, mock_get_wo):
        db = MagicMock()
        mock_get_wo.return_value = MagicMock(work_order_status="jxz")
        bad = MagicMock(conclusion="xx", remark=None)
        with self.assertRaises(ValueError):
            production_work_order_svc.complete_production_work_order(
                db, "wo-1", "t1", "user-1", bad
            )

    @patch("app.crud.production_work_order.get_production_work_order")
    def test_rejects_already_completed(self, mock_get_wo):
        db = MagicMock()
        mock_get_wo.return_value = MagicMock(work_order_status="ywc")
        with self.assertRaises(ValueError) as ctx:
            production_work_order_svc.complete_production_work_order(
                db,
                "wo-1",
                "t1",
                "user-1",
                ProductionWorkOrderComplete(conclusion="cg"),
            )
        self.assertIn("已完成", str(ctx.exception))


if __name__ == "__main__":
    unittest.main()