feat(jobs+certs): milestone-cascade Phase 1 + session patch catch-up

Implements the milestone-cascade design (Phase 1) and catches the fusion_plating_jobs / fusion_plating_certificates source up to entech. Milestone cascade (this PR's core): - fp.job: new computes all_steps_terminal, next_milestone_action, next_milestone_label; dispatcher action_advance_next_milestone with 3 helpers (_action_open_draft_certs, _action_open_draft_delivery, _action_mark_active_delivery_delivered); _resolve_required_cert_types resolver; _fp_create_certificates rewritten to honour part.certificate_requirement + partner flags + loop over resolved cert types - fp.job.workflow.state: new trigger_on_delivery_state Boolean; _fp_is_passed_for_job extended with delivery-state branch; Shipped state seed reroutes from default_kind=ship to the new trigger - View: hide Finish & Next when all_steps_terminal; add 4 mutually- exclusive milestone buttons (Mark Job Done / Issue Certs / Schedule Delivery / Mark Shipped) bound to one dispatcher - Cert gate (fusion_plating_certificates/models/fp_delivery.py): action_mark_delivered hard-blocks on draft certs; manager bypass via fp_skip_cert_gate=True context key - 24 unit tests in test_fp_job_milestone_cascade.py covering computes, resolver, dispatcher, cert gate - Spec: docs/superpowers/specs/2026-05-12-job-milestone-cascade-design.md - Plan: docs/superpowers/plans/2026-05-12-job-milestone-cascade.md Other entech changes caught up in this sync (from earlier session patches not previously committed): - fp.job version bump series 18.x → 19.0 - res_users_views.xml addition (signature widget in user prefs) - racking inspection smart button removal - various view/manifest touch-ups Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-11 22:40:25 -04:00
parent 1c1f517847
commit 913311653f
10 changed files with 755 additions and 106 deletions
--- a/fusion_plating/fusion_plating_jobs/tests/init.py
+++ b/fusion_plating/fusion_plating_jobs/tests/init.py
@@ -1,2 +1,3 @@
 # -*- coding: utf-8 -*-
 from . import test_fp_job_extensions
+from . import test_fp_job_milestone_cascade
--- a/fusion_plating/fusion_plating_jobs/tests/test_fp_job_milestone_cascade.py
+++ b/fusion_plating/fusion_plating_jobs/tests/test_fp_job_milestone_cascade.py
@@ -0,0 +1,333 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Nexa Systems Inc.
+# License OPL-1 (Odoo Proprietary License v1.0)
+"""Milestone cascade Phase 1 tests.
+
+Covers:
+  - all_steps_terminal (Task 2)
+  - _resolve_required_cert_types (Task 3)
+  - _fp_create_certificates (Task 4)
+  - next_milestone_action (Task 5)
+  - action_advance_next_milestone dispatcher (Task 6)
+  - action_mark_delivered cert gate (Task 8)
+
+See docs/superpowers/plans/2026-05-12-job-milestone-cascade.md.
+"""
+from odoo.tests.common import TransactionCase
+
+
+class TestMilestoneCascade(TransactionCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.partner = cls.env['res.partner'].create({'name': 'CustA'})
+        cls.product = cls.env['product.product'].create({
+            'name': 'Widget',
+        })
+
+    def _make_job(self, **kw):
+        vals = {
+            'partner_id': self.partner.id,
+            'product_id': self.product.id,
+            'qty': 1.0,
+        }
+        vals.update(kw)
+        return self.env['fp.job'].create(vals)
+
+    def _make_step(self, job, name='Step', state='pending'):
+        return self.env['fp.job.step'].create({
+            'job_id': job.id,
+            'name': name,
+            'state': state,
+        })
+
+    # ---------------- Task 2: all_steps_terminal ----------------------
+
+    def test_all_steps_terminal_false_when_no_steps(self):
+        job = self._make_job()
+        self.assertFalse(job.all_steps_terminal)
+
+    def test_all_steps_terminal_false_when_any_step_pending(self):
+        job = self._make_job()
+        self._make_step(job, state='done')
+        self._make_step(job, state='pending')
+        job.invalidate_recordset(['all_steps_terminal'])
+        self.assertFalse(job.all_steps_terminal)
+
+    def test_all_steps_terminal_true_when_all_done(self):
+        job = self._make_job()
+        self._make_step(job, state='done')
+        self._make_step(job, state='done')
+        job.invalidate_recordset(['all_steps_terminal'])
+        self.assertTrue(job.all_steps_terminal)
+
+    def test_all_steps_terminal_true_with_skipped_and_cancelled(self):
+        job = self._make_job()
+        self._make_step(job, state='done')
+        self._make_step(job, state='skipped')
+        self._make_step(job, state='cancelled')
+        job.invalidate_recordset(['all_steps_terminal'])
+        self.assertTrue(job.all_steps_terminal)
+
+    # ---------------- Task 3: _resolve_required_cert_types -----------
+
+    def _make_part(self, certificate_requirement='inherit'):
+        return self.env['fp.part.catalog'].create({
+            'name': 'PartA',
+            'part_number': 'PN-001-%s' % certificate_requirement,
+            'partner_id': self.partner.id,
+            'certificate_requirement': certificate_requirement,
+        })
+
+    def test_resolve_certs_none_returns_empty(self):
+        part = self._make_part(certificate_requirement='none')
+        job = self._make_job(part_catalog_id=part.id)
+        self.assertEqual(job._resolve_required_cert_types(), set())
+
+    def test_resolve_certs_coc_only(self):
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        self.assertEqual(job._resolve_required_cert_types(), {'coc'})
+
+    def test_resolve_certs_coc_plus_thickness(self):
+        part = self._make_part(certificate_requirement='coc_thickness')
+        job = self._make_job(part_catalog_id=part.id)
+        self.assertEqual(
+            job._resolve_required_cert_types(),
+            {'coc', 'thickness_report'},
+        )
+
+    def test_resolve_certs_inherit_falls_back_to_partner(self):
+        part = self._make_part(certificate_requirement='inherit')
+        self.partner.x_fc_send_coc = True
+        self.partner.x_fc_send_thickness_report = True
+        job = self._make_job(part_catalog_id=part.id)
+        self.assertEqual(
+            job._resolve_required_cert_types(),
+            {'coc', 'thickness_report'},
+        )
+
+    def test_resolve_certs_inherit_partner_says_no(self):
+        part = self._make_part(certificate_requirement='inherit')
+        self.partner.x_fc_send_coc = False
+        self.partner.x_fc_send_thickness_report = False
+        job = self._make_job(part_catalog_id=part.id)
+        self.assertEqual(job._resolve_required_cert_types(), set())
+
+    def test_resolve_certs_no_part_no_partner_flags(self):
+        self.partner.x_fc_send_coc = False
+        self.partner.x_fc_send_thickness_report = False
+        job = self._make_job()
+        self.assertEqual(job._resolve_required_cert_types(), set())
+
+    # ---------------- Task 4: _fp_create_certificates -----------------
+
+    def test_create_certs_skips_when_no_required(self):
+        part = self._make_part(certificate_requirement='none')
+        job = self._make_job(part_catalog_id=part.id)
+        job._fp_create_certificates()
+        certs = self.env['fp.certificate'].search([
+            ('x_fc_job_id', '=', job.id),
+        ])
+        self.assertFalse(certs)
+
+    def test_create_certs_coc_only(self):
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        job._fp_create_certificates()
+        certs = self.env['fp.certificate'].search([
+            ('x_fc_job_id', '=', job.id),
+        ])
+        self.assertEqual(len(certs), 1)
+        self.assertEqual(certs.certificate_type, 'coc')
+        self.assertEqual(certs.state, 'draft')
+
+    def test_create_certs_coc_plus_thickness(self):
+        part = self._make_part(certificate_requirement='coc_thickness')
+        job = self._make_job(part_catalog_id=part.id)
+        job._fp_create_certificates()
+        certs = self.env['fp.certificate'].search([
+            ('x_fc_job_id', '=', job.id),
+        ])
+        self.assertEqual(len(certs), 2)
+        self.assertEqual(
+            set(certs.mapped('certificate_type')),
+            {'coc', 'thickness_report'},
+        )
+
+    def test_create_certs_idempotent(self):
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        job._fp_create_certificates()
+        job._fp_create_certificates()  # second call must be no-op
+        certs = self.env['fp.certificate'].search([
+            ('x_fc_job_id', '=', job.id),
+        ])
+        self.assertEqual(len(certs), 1)
+
+    # ---------------- Task 5: next_milestone_action -------------------
+
+    def test_next_milestone_false_while_steps_running(self):
+        job = self._make_job()
+        self._make_step(job, state='pending')
+        job.invalidate_recordset(['all_steps_terminal'])
+        self.assertFalse(job.next_milestone_action)
+
+    def test_next_milestone_mark_done_when_state_not_done(self):
+        job = self._make_job()
+        self._make_step(job, state='done')
+        job.invalidate_recordset(['all_steps_terminal'])
+        # default state is draft after create
+        self.assertNotEqual(job.state, 'done')
+        self.assertEqual(job.next_milestone_action, 'mark_done')
+        self.assertEqual(job.next_milestone_label, 'Mark Job Done')
+
+    def test_next_milestone_issue_certs_when_draft_cert_exists(self):
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        self._make_step(job, state='done')
+        job.state = 'done'
+        job._fp_create_certificates()  # creates draft CoC
+        job.invalidate_recordset([
+            'all_steps_terminal', 'next_milestone_action',
+        ])
+        self.assertEqual(job.next_milestone_action, 'issue_certs')
+
+    def test_next_milestone_schedule_delivery_when_no_certs(self):
+        part = self._make_part(certificate_requirement='none')
+        job = self._make_job(part_catalog_id=part.id)
+        self._make_step(job, state='done')
+        job.state = 'done'
+        job.invalidate_recordset([
+            'all_steps_terminal', 'next_milestone_action',
+        ])
+        self.assertEqual(job.next_milestone_action, 'schedule_delivery')
+
+    def test_next_milestone_closed_when_delivered(self):
+        part = self._make_part(certificate_requirement='none')
+        job = self._make_job(part_catalog_id=part.id)
+        self._make_step(job, state='done')
+        job.state = 'done'
+        delivery = self.env['fusion.plating.delivery'].create({
+            'partner_id': self.partner.id,
+            'job_ref': job.name,
+            'state': 'delivered',
+        })
+        job.delivery_id = delivery.id
+        job.invalidate_recordset([
+            'all_steps_terminal', 'next_milestone_action',
+        ])
+        self.assertEqual(job.next_milestone_action, 'closed')
+
+    # ---------------- Task 6: dispatcher ------------------------------
+
+    def test_dispatcher_raises_when_no_action(self):
+        from odoo.exceptions import UserError
+        job = self._make_job()
+        self._make_step(job, state='pending')  # not terminal
+        job.invalidate_recordset(['all_steps_terminal'])
+        with self.assertRaises(UserError):
+            job.action_advance_next_milestone()
+
+    def test_open_draft_certs_returns_filtered_action(self):
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        self._make_step(job, state='done')
+        job.state = 'done'
+        job._fp_create_certificates()
+        action = job._action_open_draft_certs()
+        self.assertEqual(action['res_model'], 'fp.certificate')
+        self.assertIn(('state', '=', 'draft'), action['domain'])
+        self.assertIn(('x_fc_job_id', '=', job.id), action['domain'])
+
+    def test_open_draft_delivery_returns_form_when_draft(self):
+        job = self._make_job()
+        delivery = self.env['fusion.plating.delivery'].create({
+            'partner_id': self.partner.id,
+            'job_ref': job.name,
+            'state': 'draft',
+        })
+        job.delivery_id = delivery.id
+        action = job._action_open_draft_delivery()
+        self.assertEqual(action['res_model'], 'fusion.plating.delivery')
+        self.assertEqual(action.get('res_id'), delivery.id)
+        self.assertEqual(action['view_mode'], 'form')
+
+    def test_open_draft_delivery_falls_back_to_list(self):
+        # Delivery not draft → returns list view filtered to this job.
+        job = self._make_job()
+        self.env['fusion.plating.delivery'].create({
+            'partner_id': self.partner.id,
+            'job_ref': job.name,
+            'state': 'delivered',
+        })
+        action = job._action_open_draft_delivery()
+        self.assertEqual(action['view_mode'], 'list,form')
+        self.assertIn(('job_ref', '=', job.name), action['domain'])
+
+    def test_mark_active_raises_without_active_delivery(self):
+        from odoo.exceptions import UserError
+        job = self._make_job()
+        with self.assertRaises(UserError):
+            job._action_mark_active_delivery_delivered()
+
+    # ---------------- Task 8: cert gate on action_mark_delivered ------
+
+    def test_mark_delivered_blocks_on_draft_certs(self):
+        from odoo.exceptions import UserError
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        job.state = 'done'
+        job._fp_create_certificates()  # creates one draft CoC
+        delivery = self.env['fusion.plating.delivery'].create({
+            'partner_id': self.partner.id,
+            'job_ref': job.name,
+            'state': 'scheduled',
+        })
+        with self.assertRaises(UserError):
+            delivery.action_mark_delivered()
+
+    def test_mark_delivered_bypass_skips_cert_gate(self):
+        """With fp_skip_cert_gate=True the gate doesn't raise. Downstream
+        super() chain (notifications, invoicing) may still raise for
+        their own reasons — out of scope for this test."""
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        job.state = 'done'
+        job._fp_create_certificates()
+        delivery = self.env['fusion.plating.delivery'].create({
+            'partner_id': self.partner.id,
+            'job_ref': job.name,
+            'state': 'scheduled',
+        })
+        try:
+            delivery.with_context(
+                fp_skip_cert_gate=True,
+            ).action_mark_delivered()
+        except Exception as e:
+            # Cert-gate message must NOT appear. Anything else is fine.
+            self.assertNotIn('draft certificate', str(e))
+
+    def test_mark_delivered_passes_when_cert_issued(self):
+        """Issuing the cert clears the gate. Downstream chain errors
+        are accepted (delivery PDF render etc. — see test above)."""
+        part = self._make_part(certificate_requirement='coc')
+        job = self._make_job(part_catalog_id=part.id)
+        job.state = 'done'
+        job._fp_create_certificates()
+        cert = self.env['fp.certificate'].search([
+            ('x_fc_job_id', '=', job.id),
+        ])
+        cert.spec_reference = 'AMS 2404'
+        cert.action_issue()
+        self.assertEqual(cert.state, 'issued')
+        delivery = self.env['fusion.plating.delivery'].create({
+            'partner_id': self.partner.id,
+            'job_ref': job.name,
+            'state': 'scheduled',
+        })
+        try:
+            delivery.action_mark_delivered()
+        except Exception as e:
+            self.assertNotIn('draft certificate', str(e))