Odoo-Modules/fusion_plating/scripts/fp_e2e_full.py

# -*- coding: utf-8 -*-
# Full quote→payment E2E. Run via: odoo shell --no-http -d admin
# < scripts/fp_e2e_full.py
#
# Hits every workflow step and prints PASS/FAIL per check.

from datetime import datetime, timedelta
import base64
import sys

env = env  # noqa  injected by odoo shell


def banner(txt):
    print(f"\n{'='*72}\n  {txt}\n{'='*72}")


def check(label, ok, detail=''):
    flag = 'PASS' if ok else 'FAIL'
    print(f"  [{flag}] {label}{(' — ' + detail) if detail else ''}")
    return ok


RESULTS = []


def expect(label, ok, detail=''):
    RESULTS.append((label, ok))
    check(label, ok, detail)


# =====================================================================
banner('PHASE 1 — Quote configurator + PO → client_order_ref')
# =====================================================================

# Fresh customer per run so we can verify automations on a clean slate.
stamp = datetime.now().strftime('%y%m%d-%H%M%S')
cust_name = f'E2E Customer {stamp}'
customer = env['res.partner'].create({
    'name': cust_name,
    'company_type': 'company',
    'email': f'e2e-{stamp}@example.com',
    'street': '100 King St W',
    'city': 'Toronto', 'zip': 'M5X 1A1',
    'country_id': env.ref('base.ca').id,
})
expect('customer created', bool(customer))

# Use first available coating + part catalog. The seeder has plenty.
coating = env['fp.coating.config'].search([], limit=1)
part_cat = env['fp.part.catalog'].search([], limit=1)
expect('coating config available', bool(coating))
expect('part catalog available', bool(part_cat))

po_number = f'PO-E2E-{stamp}'
quote = env['fp.quote.configurator'].create({
    'partner_id': customer.id,
    'part_catalog_id': part_cat.id,
    'coating_config_id': coating.id,
    'quantity': 25,
    'po_number_preliminary': po_number,
})
expect('quote configurator created', bool(quote), quote.name or '')

# Trigger price calc + create the SO.
quote.action_calculate_price()
expect('price calculated', quote.unit_price > 0,
       f'unit ${quote.unit_price:.2f}')
result = quote.action_create_quotation()
so_id = result.get('res_id') if isinstance(result, dict) else False
so = env['sale.order'].browse(so_id)
expect('SO created from quote', bool(so), so.name or '')
expect('client_order_ref carries PO', so.client_order_ref == po_number,
       f'got "{so.client_order_ref}"')

# =====================================================================
banner('PHASE 2 — SO confirm → MO + portal job + WOs')
# =====================================================================

so.action_confirm()
expect('SO confirmed', so.state == 'sale')

# Auto-MO from our hook (FP-SERVICE bypasses native sale_mrp routing).
mo = env['mrp.production'].search([('origin', '=', so.name)], limit=1)
expect('MO auto-created from SO', bool(mo), mo.name or '')

# Portal job auto-created on MO confirm.
if mo.state == 'draft':
    mo.action_confirm()
job = env['fusion.plating.portal.job'].search(
    [('production_id', '=', mo.id)], limit=1,
)
expect('portal job auto-created', bool(job), job.name or '')
expect('portal job linked back to MO',
       mo.x_fc_portal_job_id.id == job.id if job else False)

# Recipe + WOs
recipe = env['fusion.plating.process.node'].search(
    [('node_type', '=', 'recipe')], limit=1)
if recipe and not mo.x_fc_recipe_id:
    mo.x_fc_recipe_id = recipe.id
    mo._generate_workorders_from_recipe()
expect('recipe assigned', bool(mo.x_fc_recipe_id),
       mo.x_fc_recipe_id.name if mo.x_fc_recipe_id else '')
expect('work orders generated', len(mo.workorder_ids) > 0,
       f'{len(mo.workorder_ids)} WOs')

# =====================================================================
banner('PHASE 3 — Receiving with auto-prefilled qty')
# =====================================================================

Receiving = env['fp.receiving']
recv = Receiving.create({
    'partner_id': customer.id,
    'sale_order_id': so.id,
    'received_date': fields.Datetime.now(),
    'line_ids': [(0, 0, {
        'description': 'E2E test parts',
        'expected_qty': 25,
    })],
})
expect('receiving record created', bool(recv), recv.name or '')
line = recv.line_ids[:1]
expect('received_qty auto-prefilled from expected_qty',
       line.received_qty == line.expected_qty,
       f'expected={line.expected_qty} received={line.received_qty}')
if recv.state == 'draft':
    try:
        recv.action_confirm()
    except Exception as e:
        print(f"  [info] receiving confirm: {e}")

# =====================================================================
banner('PHASE 4 — Execute work orders + audit timing')
# =====================================================================

# Use the first operator we can find. Auto-promotion test only needs
# one completion for verification — N is configurable per role.
operator = env['hr.employee'].search([('active', '=', True)], limit=1)
expect('operator available', bool(operator), operator.name or '')

ok_starts = ok_finishes = 0
for wo in mo.workorder_ids:
    try:
        if wo.state in ('pending', 'waiting', 'ready'):
            wo.button_start()
            ok_starts += 1
        if wo.state == 'progress':
            wo.button_finish()
            ok_finishes += 1
    except Exception as e:
        print(f"  [info] WO {wo.name}: {e}")
expect('WOs started', ok_starts > 0, f'{ok_starts} starts')
expect('WOs finished', ok_finishes > 0, f'{ok_finishes} finishes')

# Timer audit fields populated?
audited = mo.workorder_ids.filtered(
    lambda w: getattr(w, 'x_fc_started_at_utc', False)
    and getattr(w, 'x_fc_finished_at_utc', False)
)
expect('timer audit fields populated',
       len(audited) > 0, f'{len(audited)}/{len(mo.workorder_ids)}')

# Mark MO done. Triggers cert + delivery automations.
try:
    mo.button_mark_done()
except Exception as e:
    print(f"  [info] mark_done: {e}")
    # Best-effort: bypass produced-qty wizard
    try:
        mo.qty_producing = mo.product_qty
        mo._action_done()
    except Exception as e2:
        print(f"  [info] _action_done: {e2}")
expect('MO state = done', mo.state == 'done', mo.state)

# =====================================================================
banner('PHASE 5 — Certificates (CoC issued + rich PDF, no thickness dup)')
# =====================================================================

certs = env['fp.certificate'].search([('production_id', '=', mo.id)])
coc = certs.filtered(lambda c: c.certificate_type == 'coc')[:1]
thickness = certs.filtered(lambda c: c.certificate_type == 'thickness_report')
expect('CoC certificate created', bool(coc), coc.name if coc else '')
expect('thickness cert SKIPPED (CoC includes thickness)',
       len(thickness) == 0, f'thickness count={len(thickness)}')
if coc:
    expect('CoC auto-issued (state != draft)',
           coc.state != 'draft', f'state={coc.state}')
    expect('CoC has attachment', bool(coc.attachment_id),
           coc.attachment_id.name if coc.attachment_id else 'none')
    if coc.attachment_id:
        size_kb = len(base64.b64decode(coc.attachment_id.datas)) / 1024
        # Bare-header PDF is ~30KB, rich PDF is 200KB+.
        expect('CoC PDF is rich (>= 100KB, not bare header)',
               size_kb >= 100, f'{size_kb:.1f}KB')
        expect('CoC filename uses customer slug',
               cust_name.replace(' ', '_') in coc.attachment_id.name
               or 'CoC' in coc.attachment_id.name,
               coc.attachment_id.name)

# =====================================================================
banner('PHASE 6 — Delivery auto-prefilled + chain of custody')
# =====================================================================

dlv = env['fusion.plating.delivery'].search(
    [('partner_id', '=', customer.id)], order='id desc', limit=1)
expect('delivery auto-created', bool(dlv), dlv.name or '')
if dlv:
    expect('scheduled_date prefilled', bool(dlv.scheduled_date),
           str(dlv.scheduled_date or 'none'))
    # CoC cross-link from MO done hook.
    expect('CoC attached to delivery',
           bool(dlv.coc_attachment_id),
           dlv.coc_attachment_id.name if dlv.coc_attachment_id else 'none')
    # Walk through delivery states.
    try:
        if dlv.state == 'draft':
            dlv.action_schedule()
        if dlv.state == 'scheduled':
            dlv.action_start_delivery()
        if dlv.state == 'en_route':
            dlv.action_mark_delivered()
    except Exception as e:
        print(f"  [info] delivery transitions: {e}")
    expect('delivery delivered', dlv.state == 'delivered', dlv.state)
    coc_logs = env['fusion.plating.chain.of.custody'].search(
        [('delivery_id', '=', dlv.id)])
    expect('chain of custody logged', len(coc_logs) > 0,
           f'{len(coc_logs)} entries')

# Portal job should now show shipped.
job = env['fusion.plating.portal.job'].browse(job.id)
expect('portal job state advanced to shipped/complete',
       job.state in ('shipped', 'complete'), job.state)

# =====================================================================
banner('PHASE 7 — Invoice creation + post + body has lines')
# =====================================================================

# Let Odoo's standard invoice flow handle it.
try:
    inv_act = so._create_invoices()
    inv = inv_act if hasattr(inv_act, '_name') else env['account.move'].browse(
        inv_act.get('res_id') if isinstance(inv_act, dict) else inv_act
    )
except Exception as e:
    print(f"  [info] _create_invoices: {e}")
    inv = env['account.move'].search(
        [('invoice_origin', '=', so.name)], limit=1)
expect('invoice created', bool(inv), inv.name or '')

if inv:
    inv.invoice_date = fields.Date.today()
    try:
        inv.action_post()
    except Exception as e:
        print(f"  [info] action_post: {e}")
    expect('invoice posted', inv.state == 'posted', inv.state)
    # Render the new invoice PDF and confirm body has product lines
    # (Odoo 19 display_type='product' fix).
    report = env.ref('fusion_plating_reports.action_report_fp_invoice_portrait',
                     raise_if_not_found=False)
    if report:
        try:
            pdf, _ext = report.with_context(force_report_rendering=True
                )._render_qweb_pdf(report.report_name, [inv.id])
            kb = len(pdf) / 1024
            expect('invoice PDF body is non-empty (>= 50KB)',
                   kb >= 50, f'{kb:.1f}KB')
        except Exception as e:
            print(f"  [info] invoice render: {e}")
            expect('invoice PDF rendered', False, str(e))

    # Portal job should now be complete.
    expect('portal job state = complete',
           job.state == 'complete', job.state)

# =====================================================================
banner('PHASE 8 — Notification log + email attachments')
# =====================================================================

logs = env['fp.notification.log'].search(
    [('sale_order_id', '=', so.id)])
events = logs.mapped('trigger_event')
expect('notification log entries written',
       len(logs) > 0, f'{len(logs)} entries: {events}')
expect('SO confirmed notification fired',
       'so_confirmed' in events,
       'present' if 'so_confirmed' in events else 'missing')
# Shipping email should carry the CoC attachment we generated.
shipping_logs = logs.filtered(lambda l: l.trigger_event == 'shipped')
if shipping_logs:
    sl = shipping_logs[:1]
    expect('shipping email logged',
           sl.status == 'sent', sl.status)
    if sl.attachment_names:
        expect('shipping email has CoC attachment',
               'CoC' in (sl.attachment_names or '')
               or '.pdf' in (sl.attachment_names or '').lower(),
               sl.attachment_names)

# =====================================================================
banner('SUMMARY')
# =====================================================================

passed = sum(1 for _, ok in RESULTS if ok)
failed = sum(1 for _, ok in RESULTS if not ok)
print(f'\n  {passed} PASS / {failed} FAIL out of {len(RESULTS)} checks')
print(f'  customer: {customer.name}')
print(f'  SO:       {so.name}')
print(f'  MO:       {mo.name}')
print(f'  job:      {job.name}')
print(f'  delivery: {dlv.name if dlv else "(none)"}')
print(f'  invoice:  {inv.name if inv else "(none)"}')
print(f'  CoC cert: {coc.name if coc else "(none)"}')
if failed:
    print('\n  FAILURES:')
    for label, ok in RESULTS:
        if not ok:
            print(f'    - {label}')

env.cr.commit()