feat(jobs): add lifecycle hooks — portal/QC/delivery/invoice (Tasks 2.6-2.9)

- Task 2.6: fp.job.action_confirm auto-creates fusion.plating.portal.job
  with x_fc_job_id back-reference. Idempotent (skip if already linked).
- Task 2.7: fp.job.action_confirm checks customer.x_fc_requires_qc
  and best-effort creates a fusion.plating.quality.check (the model
  lives in bridge_mrp; runtime-detected to avoid dep cycle).
- Task 2.8: fp.job.button_mark_done sets state='done', date_finished,
  auto-creates draft fusion.plating.delivery and best-effort triggers
  fp.certificate generation.
- Task 2.9: account.move.action_post links invoice -> fp.job via SO
  origin lookup, updates portal_job state to complete and stamps
  invoice_ref.

5 new tests cover: portal job creation + idempotency, mark_done state
+ delivery, cancel-then-mark-done blocked.

Best-effort patterns (try/except + runtime model detection) used for
QC + cert because their target models are in dependent modules
that this module doesn't depend on by design.

qc_check_id field on fp.job still deferred — adding it here would
require depending on bridge_mrp.

Manifest 19.0.1.4.0 -> 19.0.1.5.0.

Part of: native job model migration (spec 2026-04-25)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
2026-04-24 23:27:38 -04:00
parent 294cea0e50
commit dd88afdf53
6 changed files with 289 additions and 1 deletions

View File

@@ -3,7 +3,7 @@
# License OPL-1 (Odoo Proprietary License v1.0)
{
'name': 'Fusion Plating — Native Jobs',
'version': '19.0.1.4.0',
'version': '19.0.1.5.0',
'category': 'Manufacturing/Plating',
'summary': 'Native plating job model — replaces mrp.production / mrp.workorder bridge.',
'description': """

View File

@@ -7,5 +7,7 @@
from . import fp_job
from . import fp_job_node_override
from . import fp_portal_job
from . import account_move
from . import res_config_settings
from . import sale_order

View File

@@ -0,0 +1,47 @@
# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
#
# When an invoice is posted, find the linked fp.job (via origin) and
# update the portal job state to 'complete' + stamp invoice_ref.
import logging
from odoo import models
_logger = logging.getLogger(__name__)
class AccountMove(models.Model):
_inherit = 'account.move'
def action_post(self):
result = super().action_post()
for invoice in self.filtered(
lambda m: m.move_type in ('out_invoice', 'out_refund')
):
invoice._fp_link_to_job()
return result
def _fp_link_to_job(self):
self.ensure_one()
if not self.invoice_origin:
return
Job = self.env['fp.job'].sudo()
# Walk SO -> fp.job
SO = self.env['sale.order'].sudo()
so = SO.search([('name', '=', self.invoice_origin)], limit=1)
if not so:
return
job = Job.search([('sale_order_id', '=', so.id)], limit=1)
if not job or not job.portal_job_id:
return
portal = job.portal_job_id
if 'state' in portal._fields:
portal.state = 'complete'
if 'invoice_ref' in portal._fields:
portal.invoice_ref = self.name
_logger.info(
'Invoice %s linked to fp.job %s portal %s',
self.name, job.name, portal.name,
)

View File

@@ -15,6 +15,7 @@ import logging
from markupsafe import Markup
from odoo import fields, models
from odoo.exceptions import UserError
_logger = logging.getLogger(__name__)
@@ -240,3 +241,170 @@ class FpJob(models.Model):
),
)
return True
# ------------------------------------------------------------------
# Lifecycle hooks (Tasks 2.6, 2.7, 2.8)
#
# On confirm: create the portal-job mirror record and (when the
# customer requires QC) a fusion.plating.quality.check.
# On done: create a draft fusion.plating.delivery and best-effort
# trigger fp.certificate auto-generation.
#
# The QC and certificate models live in modules this module does NOT
# depend on by design (bridge_mrp). We runtime-detect those models so
# the hooks degrade gracefully when those modules are absent.
# ------------------------------------------------------------------
def action_confirm(self):
result = super().action_confirm()
for job in self:
job._fp_create_portal_job()
job._fp_create_qc_check_if_needed()
return result
def _fp_create_portal_job(self):
"""Create the fusion.plating.portal.job mirror record."""
self.ensure_one()
if self.portal_job_id:
return # already exists — idempotent
Portal = self.env['fusion.plating.portal.job'].sudo()
portal = Portal.create({
'name': self.name,
'partner_id': self.partner_id.id,
'state': 'in_progress',
'x_fc_job_id': self.id,
})
self.portal_job_id = portal.id
def _fp_create_qc_check_if_needed(self):
"""If customer has x_fc_requires_qc=True, create a QC check.
The fusion.plating.quality.check model lives in
fusion_plating_bridge_mrp; we runtime-detect it to avoid a
depends-on-bridge_mrp cycle. If the model isn't registered, log
a warning and skip — bridge_mrp can be installed later without
breaking this flow.
"""
self.ensure_one()
partner = self.partner_id
wants_qc = (
'x_fc_requires_qc' in partner._fields
and partner.x_fc_requires_qc
)
if not wants_qc:
return
if 'fusion.plating.quality.check' not in self.env:
_logger.warning(
"Job %s: customer wants QC but fusion.plating.quality.check "
"model not registered (bridge_mrp deferral).", self.name,
)
return
QC = self.env['fusion.plating.quality.check'].sudo()
# Try to create with the most likely required fields. If the
# model has a different schema than expected, this may need
# adjustment when bridge_mrp's QC model lands here.
try:
qc_vals = {
'partner_id': partner.id,
'state': 'pending',
}
# Try the new field name first; fallback to mrp-bound.
if 'job_id' in QC._fields:
qc_vals['job_id'] = self.id
elif 'production_id' in QC._fields:
# bridge_mrp's QC binds to production. We can't fill that
# from here — leave it null and let a manual link happen.
pass
QC.create(qc_vals)
except Exception as e:
_logger.warning(
"Job %s: failed to create QC check: %s", self.name, e,
)
# ------------------------------------------------------------------
# button_mark_done — Task 2.8
# ------------------------------------------------------------------
def button_mark_done(self):
"""Transition the job to 'done' and trigger downstream side effects.
- Sets state='done', date_finished=now
- Auto-creates a draft fusion.plating.delivery
- Triggers certificate auto-generation (best-effort)
"""
for job in self:
if job.state == 'done':
continue
if job.state == 'cancelled':
raise UserError(
"Job %s is cancelled — cannot mark done." % job.name
)
job.state = 'done'
job.date_finished = fields.Datetime.now()
job._fp_create_delivery()
job._fp_create_certificates()
return True
def _fp_create_delivery(self):
"""Create a draft fusion.plating.delivery linked to this job."""
self.ensure_one()
if self.delivery_id:
return
Delivery = self.env['fusion.plating.delivery'].sudo()
# Verify the model has a job link field. The current delivery
# model uses `job_ref` (Char) as a soft reference; some forks
# may add `x_fc_job_id` (Many2one).
if 'x_fc_job_id' in Delivery._fields:
ref_field = 'x_fc_job_id'
ref_value = self.id
elif 'job_ref' in Delivery._fields:
ref_field = 'job_ref'
ref_value = self.name
else:
_logger.warning(
"Job %s: fusion.plating.delivery has no job link field; "
"delivery created without job back-reference.", self.name,
)
ref_field = None
ref_value = None
try:
vals = {
'partner_id': self.partner_id.id,
}
if ref_field:
vals[ref_field] = ref_value
delivery = Delivery.create(vals)
self.delivery_id = delivery.id
except Exception as e:
_logger.warning(
"Job %s: failed to auto-create delivery: %s", self.name, e,
)
def _fp_create_certificates(self):
"""Trigger cert auto-create on job done.
Best-effort: if fp.certificate has the right fields, create a
draft CoC. Otherwise log + skip.
"""
self.ensure_one()
if 'fp.certificate' not in self.env:
return
Cert = self.env['fp.certificate'].sudo()
try:
vals = {
'partner_id': self.partner_id.id,
}
if 'certificate_type' in Cert._fields:
vals['certificate_type'] = 'coc'
if 'state' in Cert._fields:
vals['state'] = 'draft'
# Add job link if Cert has the field
if 'x_fc_job_id' in Cert._fields:
vals['x_fc_job_id'] = self.id
elif 'job_id' in Cert._fields:
vals['job_id'] = self.id
elif 'sale_order_id' in Cert._fields and self.sale_order_id:
vals['sale_order_id'] = self.sale_order_id.id
Cert.create(vals)
except Exception as e:
_logger.warning(
"Job %s: failed to auto-create cert: %s", self.name, e,
)

View File

@@ -0,0 +1,21 @@
# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
#
# Add a back-reference from fusion.plating.portal.job to the native
# fp.job. Coexists with any future x_fc_production_id (legacy
# mrp.production link) added by bridge_mrp.
from odoo import fields, models
class FusionPlatingPortalJob(models.Model):
_inherit = 'fusion.plating.portal.job'
x_fc_job_id = fields.Many2one(
'fp.job',
string='Plating Job',
index=True,
help='Native fp.job link. Coexists with x_fc_production_id (legacy '
'mrp.production link).',
)

View File

@@ -332,3 +332,53 @@ class TestSoConfirmHook(TransactionCase):
self.assertEqual(count_after_first, count_after_second)
else:
self.skipTest('x_fc_part_catalog_id field not present')
class TestJobLifecycleHooks(TransactionCase):
def setUp(self):
super().setUp()
self.partner = self.env['res.partner'].create({'name': 'C'})
self.product = self.env['product.product'].create({'name': 'P'})
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 test_confirm_creates_portal_job(self):
job = self._make_job()
job.action_confirm()
self.assertTrue(job.portal_job_id)
self.assertEqual(job.portal_job_id.partner_id, self.partner)
def test_confirm_idempotent_portal_job(self):
job = self._make_job()
job.action_confirm()
portal_id = job.portal_job_id.id
# Second call (e.g. via a re-trigger) shouldn't create a duplicate
job._fp_create_portal_job()
self.assertEqual(job.portal_job_id.id, portal_id)
def test_button_mark_done_sets_state(self):
job = self._make_job()
job.action_confirm()
job.button_mark_done()
self.assertEqual(job.state, 'done')
self.assertTrue(job.date_finished)
def test_button_mark_done_creates_delivery(self):
job = self._make_job()
job.action_confirm()
job.button_mark_done()
self.assertTrue(job.delivery_id)
def test_button_mark_done_blocks_when_cancelled(self):
from odoo.exceptions import UserError
job = self._make_job()
job.action_cancel()
with self.assertRaises(UserError):
job.button_mark_done()