# -*- 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}')