feat(fp.job): post-shop transition helpers (Tasks 2+3)

_fp_check_advance_post_shop: in_progress + all steps terminal →
  awaiting_cert (cert required) or awaiting_ship. Auto-spawns cert
  + delivery and fires notifications. Idempotent. Does NOT raise —
  gate failures bubble up via fp.job.step.button_finish (Task 4).
_fp_check_advance_after_cert_issue: awaiting_cert → awaiting_ship
  when every required cert is state=issued.
_fp_check_regress_after_cert_void: awaiting_ship → awaiting_cert
  when a previously-issued cert is voided. Re-notifies QM.

hasattr guards on _fp_schedule_cert_activity + _fp_resolve_cert_activities
keep this safe during incremental rollout — those land in Task 20.

Test scaffolding added covering helper existence + idempotency.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
2026-05-25 09:21:22 -04:00
parent c2b693c97e
commit 5173554281
3 changed files with 203 additions and 0 deletions

View File

@@ -2144,6 +2144,108 @@ class FpJob(models.Model):
vals['coc_attachment_id'] = issued_cert.attachment_id.id
return vals
# ==================================================================
# Post-shop auto-advance helpers (spec 2026-05-25)
# ------------------------------------------------------------------
# When the last open recipe step finishes, the job auto-advances to
# awaiting_cert (if any cert is required) or awaiting_ship (if not).
# Cert issue auto-advances awaiting_cert → awaiting_ship. Cert void
# regresses awaiting_ship → awaiting_cert. All helpers are
# idempotent — safe to call from any hook.
# ==================================================================
def _fp_check_advance_post_shop(self):
"""Auto-advance in_progress jobs whose recipe steps are all
terminal. Called from fp.job.step.button_finish post-super().
Does NOT raise — gate failures (bake/qty/QC) are surfaced by
fp.job.step.button_finish BEFORE this is called (per spec D12).
At this point the step IS finished and the transition is safe.
Idempotent: re-running on a job already past in_progress is a
no-op.
"""
for job in self:
if job.state != 'in_progress':
continue
if not job.step_ids:
continue
if any(s.state not in ('done', 'skipped', 'cancelled')
for s in job.step_ids):
continue
required = job._resolve_required_cert_types() or set()
new_state = 'awaiting_cert' if required else 'awaiting_ship'
job.state = new_state
# Side effects that used to run in button_mark_done — still
# need to fire here so cert + delivery records exist.
if new_state == 'awaiting_cert':
job._fp_create_certificates()
job._fp_fire_notification('cert_awaiting_issuance')
# Forward reference — _fp_schedule_cert_activity is
# defined in Task 20. hasattr guard keeps this safe
# during incremental rollout.
if hasattr(job, '_fp_schedule_cert_activity'):
job._fp_schedule_cert_activity()
else:
job._fp_create_delivery()
job._fp_fire_notification('job_complete')
def _fp_check_advance_after_cert_issue(self):
"""Called from fp.certificate.action_issue. If every required
cert for this job is now `issued`, advance awaiting_cert →
awaiting_ship. Idempotent — safe to call repeatedly.
"""
for job in self:
if job.state != 'awaiting_cert':
continue
if 'fp.certificate' not in self.env:
continue
required = job._resolve_required_cert_types() or set()
if not required:
# Edge case: required set went empty after creation
# (e.g. partner flag toggled). Treat as "ready to ship".
job.state = 'awaiting_ship'
job._fp_create_delivery()
if hasattr(job, '_fp_resolve_cert_activities'):
job._fp_resolve_cert_activities()
continue
Cert = self.env['fp.certificate'].sudo()
outstanding = Cert.search_count([
('x_fc_job_id', '=', job.id),
('certificate_type', 'in', list(required)),
('state', '!=', 'issued'),
])
if outstanding == 0:
job.state = 'awaiting_ship'
job._fp_create_delivery()
if hasattr(job, '_fp_resolve_cert_activities'):
job._fp_resolve_cert_activities()
def _fp_check_regress_after_cert_void(self):
"""Called from fp.certificate.write when state=voided. If a
previously-issued cert is no longer issued, slide the job back
to awaiting_cert so it reappears in Final Inspection and the
QM is re-notified.
"""
for job in self:
if job.state != 'awaiting_ship':
continue
if 'fp.certificate' not in self.env:
continue
required = job._resolve_required_cert_types() or set()
if not required:
continue
Cert = self.env['fp.certificate'].sudo()
outstanding = Cert.search_count([
('x_fc_job_id', '=', job.id),
('certificate_type', 'in', list(required)),
('state', '!=', 'issued'),
])
if outstanding > 0:
job.state = 'awaiting_cert'
job._fp_fire_notification('cert_voided_re_notify')
if hasattr(job, '_fp_schedule_cert_activity'):
job._fp_schedule_cert_activity()
def _fp_create_certificates(self):
"""Auto-create one draft fp.certificate per type returned by
_resolve_required_cert_types. Idempotent per type — re-running

View File

@@ -7,3 +7,4 @@ from . import test_blocker_compute
from . import test_late_risk_ratio
from . import test_active_step_id
from . import test_autopause_cron
from . import test_post_shop_states

View File

@@ -0,0 +1,100 @@
# -*- coding: utf-8 -*-
"""Post-shop state transitions (awaiting_cert + awaiting_ship).
Spec: docs/superpowers/specs/2026-05-25-post-shop-cert-shipping-job-states-design.md
Plan: docs/superpowers/plans/2026-05-25-post-shop-cert-shipping-job-states-plan.md
"""
from odoo.tests.common import TransactionCase
class TestPostShopAdvance(TransactionCase):
def setUp(self):
super().setUp()
self.partner = self.env['res.partner'].create({'name': 'Cust'})
self.product = self.env['product.product'].create({'name': 'Widget'})
def _make_job(self, state='in_progress', **kw):
vals = {
'partner_id': self.partner.id,
'product_id': self.product.id,
'qty': 1.0,
'state': state,
}
vals.update(kw)
return self.env['fp.job'].create(vals)
# ===== Task 2 — _fp_check_advance_post_shop helper ==================
def test_advance_helper_exists(self):
job = self._make_job()
self.assertTrue(hasattr(job, '_fp_check_advance_post_shop'))
def test_advance_noop_when_state_not_in_progress(self):
# confirmed jobs should not be auto-advanced
job = self._make_job(state='confirmed')
job._fp_check_advance_post_shop()
self.assertEqual(job.state, 'confirmed')
def test_advance_noop_when_no_steps(self):
# job with zero steps stays put — nothing to evaluate
job = self._make_job(state='in_progress')
self.assertFalse(job.step_ids)
job._fp_check_advance_post_shop()
self.assertEqual(job.state, 'in_progress')
# ===== Task 3 — cert-issue + cert-void helpers =====================
def test_advance_after_cert_issue_helper_exists(self):
job = self._make_job()
self.assertTrue(hasattr(job, '_fp_check_advance_after_cert_issue'))
def test_regress_after_cert_void_helper_exists(self):
job = self._make_job()
self.assertTrue(hasattr(job, '_fp_check_regress_after_cert_void'))
def test_advance_after_cert_issue_idempotent_when_state_wrong(self):
# Calling on a draft job is a no-op.
job = self._make_job(state='draft')
job._fp_check_advance_after_cert_issue()
self.assertEqual(job.state, 'draft')
# ===== Task 4 — button_finish gates + auto-advance =================
def test_button_finish_on_last_step_triggers_advance(self):
"""Finishing the only step of an in_progress job flips state
to awaiting_ship (no cert required for this partner)."""
if 'fp.job.step' not in self.env:
self.skipTest('fp.job.step not available')
job = self._make_job(state='in_progress')
step = self.env['fp.job.step'].create({
'job_id': job.id,
'name': 'Final Inspection',
'state': 'in_progress',
'sequence': 10,
})
step.button_finish()
self.assertEqual(job.state, 'awaiting_ship')
# ===== Task 5 — button_mark_shipped ================================
def test_button_mark_shipped_requires_awaiting_ship(self):
from odoo.exceptions import UserError
job = self._make_job(state='in_progress')
with self.assertRaises(UserError):
job.button_mark_shipped()
def test_button_mark_shipped_from_awaiting_ship_lands_done(self):
job = self._make_job(state='awaiting_ship')
job.button_mark_shipped()
self.assertEqual(job.state, 'done')
self.assertTrue(job.date_finished)
# ===== Task 20 — activity helpers ==================================
def test_schedule_cert_activity_helper_exists(self):
job = self._make_job()
self.assertTrue(hasattr(job, '_fp_schedule_cert_activity'))
def test_resolve_cert_activities_helper_exists(self):
job = self._make_job()
self.assertTrue(hasattr(job, '_fp_resolve_cert_activities'))