from odoo import fields from odoo.exceptions import UserError from odoo.tests.common import TransactionCase, tagged @tagged("standard", "at_install", "directive_logic") class TestMrpWorkorderScenario(TransactionCase): """Сценарный тест директив на примере mrp.workorder (FR-10) + stock.picking путь A (FR-3.3) + крон-симулятор (MEDIUM-2/3, NIT-8). Запускается после установки directive_common (tests/ находится автоматически), поэтому directive.mixin уже примешан в mrp.workorder/stock.picking и методы on_callback доступны — в отличие от тестов directive_layer, которые при at_install бегут до установки companion-модуля. """ def setUp(self): super().setUp() Stage = self.env["directive.stage"] self.work_type = self.env["directive.work.type"].create( {"name": "CNC", "code": "cnc"} ) # Стадии: ищем сид-стадии directive_layer по коду, создаём недостающие. def _stage(code, name): st = Stage.search([("code", "=", code)], limit=1) if not st: st = Stage.create({"name": name, "code": code}) return st self.stage_draft = _stage("draft", "Draft") self.stage_confirmed = _stage("confirmed", "Confirmed") self.stage_planned = _stage("planned", "Planned") self.stage_in_progress = _stage("in_progress", "In Progress") self.stage_paused = _stage("paused", "Paused") self.stage_done = _stage("done", "Done") # Шаблон сценарной директивы (путь B через handler_key=action_iot_signal). self.tmpl_cnc = self.env["directive.template"].create({ "name": "CNC резка", "code": "tmpl_cnc_rezka", "work_type_id": self.work_type.id, "planned_hours": 1.0, "handler_key": "action_iot_signal", }) # Шаблон для крон-директивы — БЕЗ handler_key, уникальный code (Finding 3), # чтобы action_done в кроне был чистым переходом стадии. self.tmpl_agent = self.env["directive.template"].create({ "name": "Agent task", "code": "tmpl_agent", "work_type_id": self.work_type.id, "planned_hours": 1.0, }) self.group = self.env["directive.template.group"].create({ "name": "CNC Group", "template_line_ids": [(0, 0, {"template_id": self.tmpl_cnc.id})], }) # Сценарная политика: on_write state->progress, домен по имени операции. self.policy = self.env["directive.policy"].create({ "name": "MRP резка on_write", "origin_model": "mrp.workorder", "event": "on_write", "trigger_field": "state", "trigger_value_char": "progress", "domain_filter": "[('name', 'ilike', 'резка')]", "mode": "create_once", "use_record_template_group": True, }) # MRP-фикстуры: продукт + BoM с двумя операциями (одна с «резка», одна без). self.product = self.env["product.product"].create({ "name": "CNC Part", "type": "consu", "is_storable": True, }) self.workcenter = self.env["mrp.workcenter"].create({"name": "CNC Workcenter"}) self.bom = self.env["mrp.bom"].create({ "product_tmpl_id": self.product.product_tmpl_id.id, "product_qty": 1.0, "type": "normal", }) self.op_rezka = self.env["mrp.routing.workcenter"].create({ "name": "Операция резка", "workcenter_id": self.workcenter.id, "bom_id": self.bom.id, }) self.op_other = self.env["mrp.routing.workcenter"].create({ "name": "Сверловка", "workcenter_id": self.workcenter.id, "bom_id": self.bom.id, }) self.mo = self.env["mrp.production"].create({ "product_id": self.product.id, "product_qty": 1.0, "bom_id": self.bom.id, }) self.mo.action_confirm() self.wo_rezka = self.mo.workorder_ids.filtered( lambda w: "резка" in (w.name or "").lower() ) self.wo_other = self.mo.workorder_ids.filtered( lambda w: "резка" not in (w.name or "").lower() ) self.assertTrue(self.wo_rezka, "Операция «резка» не создалась на MO") self.assertTrue(self.wo_other, "Контрольная операция (без «резка») не создалась") # Группа шаблонов на обеих операциях (use_record_template_group=True). (self.wo_rezka | self.wo_other).write( {"directive_template_group_id": self.group.id} ) def _directives_for(self, workorder): return self.env["directive.directive"].search([ ("origin_id.res_model", "=", "mrp.workorder"), ("origin_id.res_id", "=", workorder.id), ("policy_id", "=", self.policy.id), ]) # ------------------------------------------------------------------ FR-10.1 def test_01_creation_on_button_start(self): """FR-10.1: button_start() (ready->progress) создаёт ровно одну директиву для операции «резка»; контрольная операция — ни одной (негатив домена).""" self.assertEqual(self.wo_rezka.state, "ready") self.wo_rezka.button_start() self.assertEqual(self.wo_rezka.state, "progress") directives = self._directives_for(self.wo_rezka) self.assertEqual(len(directives), 1, "Директива не создалась через on_write") self.assertEqual(directives.name, self.tmpl_cnc.name) self.assertEqual(directives.planned_hours, self.tmpl_cnc.planned_hours) # Негатив домена: операция без «резка» не порождает директив. self.wo_other.button_start() self.assertEqual(self.wo_other.state, "progress") self.assertFalse(self._directives_for(self.wo_other), "Директива создалась вопреки domain_filter") # create_once: повторный вход в progress не плодит дубль. self.wo_rezka.write({"state": "ready"}) self.wo_rezka.write({"state": "progress"}) self.assertEqual(len(self._directives_for(self.wo_rezka)), 1, "Сломался create_once") # ------------------------------------------------------------------ FR-10.2 def test_02_stage_movement_happy_path(self): """FR-10.2: корректная последовательность стадий директивы.""" self.wo_rezka.button_start() directive = self._directives_for(self.wo_rezka) self.assertEqual(len(directive), 1) self.assertEqual(directive.stage_code, "draft") directive._set_stage("confirmed") self.assertEqual(directive.stage_code, "confirmed") directive.action_assign_to_me() self.assertTrue(directive.executor_user_id) self.assertEqual(directive.stage_code, "confirmed") directive.action_start() self.assertEqual(directive.stage_code, "in_progress") directive.action_pause() self.assertEqual(directive.stage_code, "paused") # Возобновление через сеттер, НЕ через action_start (гард action_start). directive._set_stage("in_progress") self.assertEqual(directive.stage_code, "in_progress") directive.action_done() self.assertEqual(directive.stage_code, "done") def test_03_invalid_transition_raises_usererror(self): """FR-10.2 / FEAT-001: недопустимый переход даёт UserError (не TypeError).""" self.wo_rezka.button_start() directive = self._directives_for(self.wo_rezka) self.assertEqual(directive.stage_code, "draft") with self.assertRaises(UserError): directive._set_stage("done") # draft->done не разрешён def test_04_action_guards(self): """FR-10.2: action_assign_to_me требует confirmed/planned; action_start требует назначенного исполнителя.""" self.wo_rezka.button_start() directive = self._directives_for(self.wo_rezka) # Из draft нельзя назначить на себя. with self.assertRaises(UserError): directive.action_assign_to_me() # confirmed без исполнителя — action_start падает. directive._set_stage("confirmed") with self.assertRaises(UserError): directive.action_start() # --------------------------------------------------------- FR-10.2 cron def test_05_cron_simulate_system_agents(self): """Крон двигает planned->in_progress, затем in_progress->done — только для выделенной software-директивы (MEDIUM-2/3, NIT-8). Инвариант: сценарная директива из button_start остаётся executor_type='human' (дефолт) и крон её игнорирует — отдельная software-директива с tmpl_agent (без handler_key) используется для крона. """ self.wo_rezka.button_start() origin = self.wo_rezka.directive_origin_id self.assertTrue(origin) cron_directive = self.env["directive.directive"].create({ "name": "Agent directive", "deadline_at": fields.Datetime.now(), "planned_hours": 1.0, "work_type_id": self.work_type.id, "executor_type": "software", "origin_id": origin.id, "template_id": self.tmpl_agent.id, }) # Finding 2: директива создаётся в draft -> переводим в planned ПЕРВЫМ шагом. cron_directive._set_stage("planned") self.assertEqual(cron_directive.stage_code, "planned") self.env["directive.directive"]._cron_simulate_system_agents() self.assertEqual(cron_directive.stage_code, "in_progress") self.env["directive.directive"]._cron_simulate_system_agents() self.assertEqual(cron_directive.stage_code, "done") # ------------------------------------------------------------------ FR-10.3 def test_06_buttons_and_fields(self): """FR-10.3: поля/методы на модели + кнопка только на форме mrp.production.""" wo = self.wo_rezka self.assertIn("directive_execution_status", wo._fields) self.assertTrue(hasattr(wo, "action_choose_directive_group")) self.assertTrue(hasattr(wo, "action_open_directives")) view = self.env.ref( "directive_common.mrp_production_form_directives_inherit" ) arch = self.env["mrp.production"].get_view(view_id=view.id)["arch"] self.assertIn("action_choose_directive_group", arch, "Кнопка выбора группы не найдена на форме mrp.production") # ------------------------------------------------------------------ FR-10.4 def test_07_callback_path_b(self): """FR-10.4: action_done -> action_iot_signal пишет телеметрию в MO (путь B, без monkey-patch).""" self.wo_rezka.button_start() directive = self._directives_for(self.wo_rezka) # draft -> confirmed -> in_progress (прямого draft->in_progress нет). directive._set_stage("confirmed") directive._set_stage("in_progress") before = len(self.mo.message_ids) directive.action_done() self.assertEqual(directive.stage_code, "done") self.mo.invalidate_recordset(["message_ids"]) self.assertGreater(len(self.mo.message_ids), before, "action_iot_signal не записал сообщение в MO") # ------------------------------------------------------------------ FR-3.3 def test_08_stock_picking_callback_path_a(self): """FR-3.3 путь A: stock.picking.button_validate -> helper создаёт директиву.""" picking_type = self.env["stock.picking.type"].search( [("code", "=", "internal")], limit=1 ) or self.env["stock.picking.type"].search([], limit=1) policy_picking = self.env["directive.policy"].create({ "name": "Picking on_callback", "origin_model": "stock.picking", "event": "on_callback", "trigger_field": "button_validate", "mode": "create_once", "use_record_template_group": True, }) src = picking_type.default_location_src_id or self.env["stock.location"].search( [("usage", "=", "internal")], limit=1 ) dest = picking_type.default_location_dest_id or self.env["stock.location"].search( [("usage", "=", "internal")], limit=1 ) picking = self.env["stock.picking"].create({ "picking_type_id": picking_type.id, "location_id": src.id, "location_dest_id": dest.id, "directive_template_group_id": self.group.id, "move_ids": [(0, 0, { "product_id": self.product.id, "product_uom_qty": 1.0, "location_id": src.id, "location_dest_id": dest.id, })], }) picking.action_confirm() for move in picking.move_ids: move.quantity = move.product_uom_qty picking.move_ids.picked = True picking.button_validate() def _picking_directives(): return self.env["directive.directive"].search([ ("origin_id.res_model", "=", "stock.picking"), ("origin_id.res_id", "=", picking.id), ("policy_id", "=", policy_picking.id), ]) self.assertEqual(len(_picking_directives()), 1, "Директива не создалась через button_validate (путь A)") # create_once: повторный проход хелпера не плодит дубль. picking._directive_run_callback_policies("button_validate") self.assertEqual(len(_picking_directives()), 1, "Сломался create_once для stock.picking")