Odoo-Modules/fusion_plating/scripts/fp_enforcement_audit.py

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