8c76a16366
Replace em-dashes and en-dashes with hyphens across 789 shipped source files (py/xml/js/scss) so the delivered module reads as human-written; em-dashes had become a recognizable AI-generated tell. Internal .md dev notes are excluded. The WO-sticker mojibake strippers keep their dash search targets (now written — / –). No logic changes: comments and display strings only; validated with py_compile + lxml parse. Rewrite the 7 customer notification emails to be intake-neutral (ship-in / drop-off / pickup) and repair-aware, and fix the Shipped email documents line (packing slip vs bill of lading; certificate only when issued). Subjects use a hyphen separator. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
464 lines
17 KiB
Python
464 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Workflow enforcement audit.
|
|
|
|
For each workflow transition (SO confirm, MO confirm, WO start, WO finish,
|
|
delivery delivered, invoice post, NCR close, CAPA close, discharge close,
|
|
cert issue), tries to perform it with MISSING data and reports whether
|
|
the system blocks (PASS) or lets it through silently (GAP).
|
|
|
|
Each test is wrapped in a SAVEPOINT so the DB is unchanged.
|
|
"""
|
|
from datetime import datetime
|
|
env = env # noqa
|
|
from odoo import fields # noqa
|
|
|
|
|
|
RESULTS = []
|
|
|
|
|
|
def gate(label, fn, expect_keywords):
|
|
"""Run fn() in a savepoint. PASS = raises UserError mentioning one
|
|
of expect_keywords. GAP = succeeds silently. ERR = unexpected error."""
|
|
sp = f'gate_{abs(hash(label))}'
|
|
env.cr.execute(f'SAVEPOINT {sp}')
|
|
fired = False
|
|
msg = ''
|
|
err = ''
|
|
try:
|
|
fn()
|
|
except Exception as e:
|
|
msg = str(e).splitlines()[0][:140]
|
|
low = str(e).lower()
|
|
fired = any(k.lower() in low for k in expect_keywords)
|
|
if not fired and 'NotNullViolation' not in str(type(e).__name__):
|
|
err = str(e).splitlines()[0][:140]
|
|
finally:
|
|
env.cr.execute(f'ROLLBACK TO SAVEPOINT {sp}')
|
|
if err and not fired:
|
|
verdict = 'ERR '
|
|
elif fired:
|
|
verdict = 'PASS'
|
|
else:
|
|
verdict = 'GAP '
|
|
RESULTS.append((verdict, label, msg))
|
|
sym = {'PASS': '✓', 'GAP ': '✗', 'ERR ': '?'}[verdict]
|
|
print(f' {sym} {verdict} [{label:<60}] {msg[:80]}')
|
|
|
|
|
|
def section(title):
|
|
print(f'\n{"="*78}\n {title}\n{"="*78}')
|
|
|
|
|
|
# Setup: pick existing fixtures
|
|
customer = env['res.partner'].search([('is_company', '=', True)], limit=1)
|
|
fac = env['fusion.plating.facility'].search([('active', '=', True)], limit=1)
|
|
coating = env['fp.coating.config'].search([], limit=1)
|
|
mo_done = env['mrp.production'].search([('state', '=', 'done')], order='id desc', limit=1)
|
|
mo_progress = env['mrp.production'].search([('state', 'in', ('confirmed', 'progress'))], limit=1)
|
|
|
|
# =====================================================================
|
|
section('1. Sale Order - quote → confirmation')
|
|
# =====================================================================
|
|
|
|
def t_so_no_partner():
|
|
env['sale.order'].sudo().create({})
|
|
|
|
gate('SO create without partner', t_so_no_partner,
|
|
['partner', 'required', 'NotNull'])
|
|
|
|
|
|
def t_so_confirm_account_hold():
|
|
p = customer.copy({'name': f'AcctHold {datetime.now().timestamp()}', 'x_fc_account_hold': True})
|
|
so = env['sale.order'].sudo().create({
|
|
'partner_id': p.id,
|
|
'order_line': [(0, 0, {'name': 'svc', 'price_unit': 100,
|
|
'product_id': env['product.product'].search([('default_code', '=', 'FP-SERVICE')], limit=1).id or False})],
|
|
})
|
|
so.with_user(env.ref('base.user_demo', raise_if_not_found=False) or env.user).action_confirm()
|
|
|
|
|
|
gate('SO confirm blocked when customer on Account Hold', t_so_confirm_account_hold,
|
|
['Account Hold', 'account hold'])
|
|
|
|
|
|
# =====================================================================
|
|
section('2. Manufacturing Order - confirm + done')
|
|
# =====================================================================
|
|
|
|
def t_mo_confirm_no_facility():
|
|
saved = env.company.x_fc_default_facility_id
|
|
env.company.sudo().x_fc_default_facility_id = False
|
|
facs = env['fusion.plating.facility'].search([('active', '=', True)])
|
|
facs.sudo().write({'active': False})
|
|
try:
|
|
product = env['product.product'].search([], limit=1)
|
|
m = env['mrp.production'].sudo().create({
|
|
'product_id': product.id, 'product_qty': 1,
|
|
'company_id': env.company.id,
|
|
})
|
|
m.action_confirm()
|
|
finally:
|
|
facs.sudo().write({'active': True})
|
|
env.company.sudo().x_fc_default_facility_id = saved
|
|
|
|
|
|
gate('MO confirm blocked when no facility resolvable', t_mo_confirm_no_facility,
|
|
['facility'])
|
|
|
|
|
|
# =====================================================================
|
|
section('3. Work Order - start (per kind) + finish')
|
|
# =====================================================================
|
|
|
|
if mo_progress:
|
|
test_wos = mo_progress.workorder_ids[:5]
|
|
if test_wos:
|
|
wo = test_wos[0]
|
|
|
|
def t_wo_start_no_operator():
|
|
saved = wo.x_fc_assigned_user_id.id
|
|
wo.sudo().x_fc_assigned_user_id = False
|
|
try:
|
|
wo.sudo().button_start()
|
|
finally:
|
|
wo.sudo().x_fc_assigned_user_id = saved
|
|
|
|
gate('WO start blocked without assigned operator', t_wo_start_no_operator,
|
|
['Assigned Operator', 'operator'])
|
|
|
|
# Find a wet WO + bake WO + rack WO + mask WO from any MO
|
|
def find_wo(kind, state=None):
|
|
for mo in env['mrp.production'].search([], order='id desc', limit=20):
|
|
for w in mo.workorder_ids:
|
|
if hasattr(w, '_fp_classify_kind') and w._fp_classify_kind() == kind:
|
|
if state is None or w.state == state:
|
|
return w
|
|
return None
|
|
|
|
|
|
wet_wo = find_wo('wet')
|
|
bake_wo = find_wo('bake')
|
|
rack_wo = find_wo('rack')
|
|
mask_wo = find_wo('mask')
|
|
|
|
if wet_wo:
|
|
def t_wet_no_bath():
|
|
saved_b, saved_t = wet_wo.x_fc_bath_id.id, wet_wo.x_fc_tank_id.id
|
|
wet_wo.sudo().write({'x_fc_bath_id': False, 'x_fc_tank_id': False})
|
|
try:
|
|
wet_wo.sudo().button_start()
|
|
finally:
|
|
wet_wo.sudo().write({'x_fc_bath_id': saved_b, 'x_fc_tank_id': saved_t})
|
|
gate('WO[wet] start blocked without bath+tank', t_wet_no_bath,
|
|
['Bath', 'Tank'])
|
|
|
|
if bake_wo:
|
|
def t_bake_no_oven():
|
|
saved = bake_wo.x_fc_oven_id.id
|
|
bake_wo.sudo().x_fc_oven_id = False
|
|
try:
|
|
bake_wo.sudo().button_start()
|
|
finally:
|
|
bake_wo.sudo().x_fc_oven_id = saved
|
|
gate('WO[bake] start blocked without oven', t_bake_no_oven, ['Oven'])
|
|
|
|
def t_bake_finish_no_actuals():
|
|
# Already started? Need to be in 'progress' state to finish
|
|
if bake_wo.state == 'progress':
|
|
saved_t, saved_d = bake_wo.x_fc_bake_temp, bake_wo.x_fc_bake_duration_hours
|
|
bake_wo.sudo().write({'x_fc_bake_temp': 0, 'x_fc_bake_duration_hours': 0})
|
|
try:
|
|
bake_wo.sudo().button_finish()
|
|
finally:
|
|
bake_wo.sudo().write({'x_fc_bake_temp': saved_t,
|
|
'x_fc_bake_duration_hours': saved_d})
|
|
else:
|
|
raise Exception('bake WO not in progress, cannot test finish')
|
|
gate('WO[bake] finish blocked without temp+duration+chart_recorder',
|
|
t_bake_finish_no_actuals,
|
|
['Bake Temp', 'Bake Duration', 'Chart Recorder', 'progress'])
|
|
|
|
if rack_wo:
|
|
def t_rack_no_rack():
|
|
saved = rack_wo.x_fc_rack_id.id
|
|
rack_wo.sudo().x_fc_rack_id = False
|
|
try:
|
|
rack_wo.sudo().button_start()
|
|
finally:
|
|
rack_wo.sudo().x_fc_rack_id = saved
|
|
gate('WO[rack] start blocked without rack/fixture', t_rack_no_rack, ['Rack', 'Fixture'])
|
|
|
|
if mask_wo:
|
|
def t_mask_no_material():
|
|
saved = mask_wo.x_fc_masking_material
|
|
mask_wo.sudo().x_fc_masking_material = False
|
|
try:
|
|
mask_wo.sudo().button_start()
|
|
finally:
|
|
mask_wo.sudo().x_fc_masking_material = saved
|
|
gate('WO[mask] start blocked without masking material', t_mask_no_material,
|
|
['Masking Material'])
|
|
|
|
|
|
# =====================================================================
|
|
section('4. Receiving - accept/discrepancy with damage')
|
|
# =====================================================================
|
|
|
|
if customer:
|
|
so_for_recv = env['sale.order'].search([('partner_id', '=', customer.id)], limit=1)
|
|
if so_for_recv:
|
|
def t_recv_accept_with_unresolved_damage():
|
|
r = env['fp.receiving'].sudo().create({
|
|
'sale_order_id': so_for_recv.id,
|
|
'expected_qty': 5, 'received_qty': 5,
|
|
})
|
|
env['fp.receiving.damage'].sudo().create({
|
|
'receiving_id': r.id,
|
|
'description': 'Test damage',
|
|
'resolved': False,
|
|
})
|
|
r.action_start_inspection()
|
|
r.action_accept() # should fail - unresolved damage
|
|
gate('Receiving accept blocked when unresolved damage exists',
|
|
t_recv_accept_with_unresolved_damage,
|
|
['unresolved damage'])
|
|
|
|
|
|
# =====================================================================
|
|
section('5. Certificate - action_issue')
|
|
# =====================================================================
|
|
|
|
if mo_done:
|
|
def t_cert_issue_no_spec():
|
|
c = env['fp.certificate'].sudo().create({
|
|
'partner_id': customer.id,
|
|
'production_id': mo_done.id,
|
|
'certificate_type': 'coc',
|
|
'spec_reference': False,
|
|
})
|
|
c.action_issue()
|
|
gate('Cert issue blocked without spec_reference', t_cert_issue_no_spec,
|
|
['Spec', 'spec_reference'])
|
|
|
|
|
|
# =====================================================================
|
|
section('6. Delivery - schedule → en_route → delivered')
|
|
# =====================================================================
|
|
|
|
if customer:
|
|
def t_dlv_delivered_no_pod():
|
|
d = env['fusion.plating.delivery'].sudo().create({
|
|
'partner_id': customer.id, 'state': 'en_route',
|
|
'company_id': env.company.id,
|
|
})
|
|
d.action_mark_delivered()
|
|
gate('Delivery mark_delivered blocked without POD', t_dlv_delivered_no_pod,
|
|
['POD', 'Proof of Delivery'])
|
|
|
|
|
|
# =====================================================================
|
|
section('7. Invoice - post')
|
|
# =====================================================================
|
|
|
|
if customer:
|
|
def t_inv_post_no_terms():
|
|
saved = customer.property_payment_term_id
|
|
customer.sudo().property_payment_term_id = False
|
|
try:
|
|
i = env['account.move'].sudo().create({
|
|
'move_type': 'out_invoice',
|
|
'partner_id': customer.id,
|
|
'invoice_date': fields.Date.today(),
|
|
'invoice_line_ids': [(0, 0, {'name': 'x', 'quantity': 1, 'price_unit': 1})],
|
|
})
|
|
i.invoice_payment_term_id = False
|
|
i.action_post()
|
|
finally:
|
|
customer.sudo().property_payment_term_id = saved
|
|
gate('Invoice post blocked without payment terms', t_inv_post_no_terms,
|
|
['payment term'])
|
|
|
|
def t_inv_post_account_hold():
|
|
p = customer.copy({'name': f'Hold-{datetime.now().timestamp()}',
|
|
'x_fc_account_hold': True})
|
|
i = env['account.move'].sudo().create({
|
|
'move_type': 'out_invoice',
|
|
'partner_id': p.id,
|
|
'invoice_date': fields.Date.today(),
|
|
'invoice_payment_term_id': env.ref('account.account_payment_term_30days', raise_if_not_found=False).id if env.ref('account.account_payment_term_30days', raise_if_not_found=False) else False,
|
|
'invoice_line_ids': [(0, 0, {'name': 'x', 'quantity': 1, 'price_unit': 1})],
|
|
})
|
|
i.with_user(env.ref('base.user_admin').id).action_post()
|
|
gate('Invoice post blocked when customer on Account Hold', t_inv_post_account_hold,
|
|
['Account Hold', 'account hold'])
|
|
|
|
|
|
# =====================================================================
|
|
section('8. QMS - NCR / CAPA / Discharge sample close')
|
|
# =====================================================================
|
|
|
|
def t_ncr_close_missing():
|
|
n = env['fusion.plating.ncr'].sudo().create({
|
|
'facility_id': fac.id,
|
|
'description': '', 'containment': '', 'root_cause': '',
|
|
'disposition': False,
|
|
})
|
|
n.action_close()
|
|
gate('NCR close blocked without RC/containment/disposition', t_ncr_close_missing,
|
|
['Root Cause', 'Containment', 'Disposition'])
|
|
|
|
|
|
def t_capa_close_missing():
|
|
c = env['fusion.plating.capa'].sudo().create({
|
|
'description': '', 'root_cause_analysis': '', 'action_plan': '',
|
|
})
|
|
c.action_close()
|
|
gate('CAPA close blocked without RCA/plan/verification', t_capa_close_missing,
|
|
['Root Cause Analysis', 'Action Plan', 'Verification'])
|
|
|
|
|
|
def t_discharge_close_missing():
|
|
s = env['fusion.plating.discharge.sample'].sudo().create({
|
|
'facility_id': fac.id,
|
|
})
|
|
s.action_close()
|
|
gate('Discharge sample close blocked without lab evidence', t_discharge_close_missing,
|
|
['Lab Report', 'Results Received', 'parameter', 'Lab certificate'])
|
|
|
|
|
|
# =====================================================================
|
|
section('9. SUSPECTED GAPS - these probably AREN\'T enforced today')
|
|
# =====================================================================
|
|
|
|
# Each of these MIGHT slip through silently. If they do, we'll see GAP.
|
|
|
|
# 9a. SO confirm without customer PO#
|
|
def t_so_confirm_no_po():
|
|
p = customer.copy({'name': f'NoPO-{datetime.now().timestamp()}'})
|
|
so = env['sale.order'].sudo().create({
|
|
'partner_id': p.id,
|
|
'order_line': [(0, 0, {'name': 'svc', 'price_unit': 100,
|
|
'product_id': env['product.product'].search([('default_code', '=', 'FP-SERVICE')], limit=1).id})],
|
|
})
|
|
# client_order_ref intentionally empty
|
|
so.action_confirm()
|
|
gate('SO confirm blocked without customer PO# (client_order_ref)',
|
|
t_so_confirm_no_po,
|
|
['PO', 'client_order_ref', 'customer reference'])
|
|
|
|
|
|
# 9b. Receiving accept with qty mismatch
|
|
if customer:
|
|
def t_recv_accept_qty_mismatch():
|
|
so = env['sale.order'].search([('partner_id', '=', customer.id)], limit=1)
|
|
if not so:
|
|
raise Exception('no SO available')
|
|
r = env['fp.receiving'].sudo().create({
|
|
'sale_order_id': so.id,
|
|
'expected_qty': 10, 'received_qty': 7, # short!
|
|
})
|
|
r.action_start_inspection()
|
|
r.action_accept() # should this be allowed with qty mismatch?
|
|
gate('Receiving accept blocked when qty mismatch (expected != received)',
|
|
t_recv_accept_qty_mismatch,
|
|
['mismatch', 'short', 'discrepancy', 'qty', 'quantity'])
|
|
|
|
|
|
# 9c. MO done without all WOs done - Odoo enforces this natively, so should PASS
|
|
# Skip - Odoo handles it.
|
|
|
|
# 9d. Cert issue without thickness readings (only blocks when partner
|
|
# is flagged aerospace via x_fc_strict_thickness_required)
|
|
if mo_done:
|
|
def t_cert_issue_no_readings_aero():
|
|
# Flag the customer as aerospace for the test, then unflag
|
|
saved = customer.x_fc_strict_thickness_required
|
|
customer.sudo().x_fc_strict_thickness_required = True
|
|
try:
|
|
c = env['fp.certificate'].sudo().create({
|
|
'partner_id': customer.id,
|
|
'production_id': mo_done.id,
|
|
'certificate_type': 'coc',
|
|
'spec_reference': 'AMS 2404',
|
|
})
|
|
c.action_issue()
|
|
finally:
|
|
customer.sudo().x_fc_strict_thickness_required = saved
|
|
gate('Cert issue blocked without thickness readings (aerospace customer)',
|
|
t_cert_issue_no_readings_aero,
|
|
['thickness', 'reading', 'Nadcap'])
|
|
|
|
|
|
# 9e. Delivery start_route without driver
|
|
if customer:
|
|
def t_dlv_start_no_driver():
|
|
d = env['fusion.plating.delivery'].sudo().create({
|
|
'partner_id': customer.id, 'state': 'scheduled',
|
|
'company_id': env.company.id,
|
|
})
|
|
# No driver, no vehicle
|
|
d.action_start_route()
|
|
gate('Delivery start_route blocked without driver',
|
|
t_dlv_start_no_driver,
|
|
['driver', 'vehicle'])
|
|
|
|
|
|
# 9f. WO finish for inspection WO without thickness readings logged
|
|
inspect_wo = find_wo('inspect')
|
|
if inspect_wo and inspect_wo.state == 'progress':
|
|
def t_inspect_finish_no_readings():
|
|
# Wipe all readings linked to this MO
|
|
readings = env['fp.thickness.reading'].sudo().search([
|
|
('production_id', '=', inspect_wo.production_id.id),
|
|
])
|
|
readings.unlink()
|
|
inspect_wo.sudo().button_finish()
|
|
gate('WO[inspect] finish blocked without any thickness readings',
|
|
t_inspect_finish_no_readings,
|
|
['thickness', 'reading'])
|
|
|
|
|
|
# 9g. Bath log create without any readings
|
|
bath = env['fusion.plating.bath'].search([], limit=1)
|
|
if bath:
|
|
def t_bath_log_no_lines():
|
|
env['fusion.plating.bath.log'].sudo().create({
|
|
'bath_id': bath.id,
|
|
}) # no line_ids - should this be allowed?
|
|
gate('Bath log create blocked without any parameter readings',
|
|
t_bath_log_no_lines,
|
|
['line', 'reading', 'parameter'])
|
|
|
|
|
|
# 9h. Quality hold without inspector / reason
|
|
def t_hold_no_data():
|
|
env['fusion.plating.quality.hold'].sudo().create({
|
|
# All optional except partner_id?
|
|
})
|
|
gate('Quality hold create requires partner+reason+description',
|
|
t_hold_no_data,
|
|
['partner', 'reason', 'description', 'NotNull'])
|
|
|
|
|
|
# =====================================================================
|
|
section('SUMMARY')
|
|
# =====================================================================
|
|
|
|
passed = sum(1 for v, _, _ in RESULTS if v == 'PASS')
|
|
gaps = sum(1 for v, _, _ in RESULTS if v == 'GAP ')
|
|
errs = sum(1 for v, _, _ in RESULTS if v == 'ERR ')
|
|
total = len(RESULTS)
|
|
print(f'\n {passed} ENFORCED / {gaps} GAPS / {errs} ERR (out of {total} checks)')
|
|
|
|
if gaps:
|
|
print('\n Gates that DON\'T fire today (potential enforcement gaps):')
|
|
for v, label, msg in RESULTS:
|
|
if v == 'GAP ':
|
|
print(f' ✗ {label}')
|
|
|
|
if errs:
|
|
print('\n Gates that errored unexpectedly (test setup issue or new bug):')
|
|
for v, label, msg in RESULTS:
|
|
if v == 'ERR ':
|
|
print(f' ? {label}: {msg}')
|