Files
Odoo-Modules/fusion_plating/scripts/fp_e2e_human.py
gsinghpal d351a2577b chore(receiving): port received_qty auto-prefill from live entech to main
The auto-prefill logic that fills received_qty from expected_qty on
fp.receiving create was committed to the entech LXC but never made it
back to main. Verified by a full quote→delivery→invoice walkthrough
(scripts/fp_e2e_human.py) — receiving step now passes.

Also adds the human-walkthrough E2E script that exercises every step:
RFQ → quote → SO confirm → MO + portal job auto-create → receiving
prefill → recipe → WO execution → MO done → CoC cert (rich PDF, no
thickness duplicate) → delivery prefill + lifecycle → invoice (posted,
not auto-paid) → notification log audit.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-19 01:26:16 -04:00

328 lines
14 KiB
Python

# -*- coding: utf-8 -*-
# Human-style E2E walkthrough. Run via:
# cat fp_e2e_human.py | odoo shell -c /etc/odoo/odoo.conf -d admin --no-http
# Each STEP is narrated as a person clicking through the UI.
from datetime import datetime, timedelta
import base64
env = env # noqa injected by odoo shell
from odoo import fields # noqa
def step(num, who, action):
bar = '' * 70
print(f'\n{bar}\n STEP {num}{who}\n{action}\n{bar}')
def show(label, value):
print(f' {label:<28} {value}')
def hr():
print(' ' + '' * 64)
# ─────────────────────────────────────────────────────────────────────
# Setup: pick a generic existing customer? No — fresh customer so we
# can prove the full chain of automations on a clean slate.
# ─────────────────────────────────────────────────────────────────────
stamp = datetime.now().strftime('%y%m%d-%H%M%S')
# =====================================================================
step(1, 'CUSTOMER (portal)', 'Submits an RFQ via the customer portal')
# =====================================================================
# A new contact appears the moment a portal user fills the RFQ form.
# In production a portal session would do this; for this walkthrough
# we'll create the partner + RFQ as if the portal accepted it.
customer = env['res.partner'].create({
'name': f'Acme Aerospace {stamp}',
'company_type': 'company',
'email': f'orders-{stamp}@acmeaero.example',
'phone': '+1-416-555-0142',
'street': '88 Queen St E',
'city': 'Toronto', 'zip': 'M5C 1A8',
'country_id': env.ref('base.ca').id,
})
show('new contact', f'{customer.name} (id={customer.id})')
rfq = env['fusion.plating.quote.request'].create({
'partner_id': customer.id,
'contact_name': 'Sandra Kim',
'contact_email': customer.email,
'company_name': customer.name,
'part_description': '<p>25 stainless brackets, AMS 2404, ~50µin ENP.</p>',
'quantity': 25,
'state': 'new',
})
show('RFQ created', f'{rfq.name} (state={rfq.state})')
show('inbox alert', 'Sales sees a new RFQ on Plating > Sales > Quote Requests')
# =====================================================================
step(2, 'SALES (estimator)',
'Reviews RFQ + builds a configurator quote with PO# + customer price')
# =====================================================================
coating = env['fp.coating.config'].search([], limit=1)
part_cat = env['fp.part.catalog'].search([], limit=1)
show('coating template', coating.name if coating else '(none)')
show('part catalog', part_cat.name if part_cat else '(none)')
po_number = f'PO-ACME-{stamp}'
quote = env['fp.quote.configurator'].create({
'partner_id': customer.id,
'part_catalog_id': part_cat.id if part_cat else False,
'coating_config_id': coating.id if coating else False,
'quantity': 25,
'po_number_preliminary': po_number,
'estimator_override_price': 1875.00,
})
show('quote session', f'{quote.name} (state={quote.state})')
show('estimator override', f'${quote.estimator_override_price:,.2f}')
show('calculated price', f'${quote.calculated_price:,.2f} ({quote.currency_id.name})')
show('PO# entered', po_number)
# Sales clicks "Create Quotation" — this is the SO.
result = quote.action_create_quotation()
so = env['sale.order'].browse(result.get('res_id'))
show('SO drafted', f'{so.name} (state={so.state})')
show('SO total', f'${so.amount_total:,.2f}')
show('SO has PO# (client_order_ref)', so.client_order_ref or '(empty)')
show('SO links back to RFQ origin', so.origin or '(empty)')
# =====================================================================
step(3, 'CUSTOMER',
'Reviews the quote PDF, signs / accepts → SO confirmed')
# =====================================================================
# Render the quote PDF the customer would receive.
quote_report = env.ref(
'fusion_plating_reports.action_report_fp_sale_portrait',
raise_if_not_found=False)
if quote_report:
pdf, _e = quote_report.with_context(force_report_rendering=True
)._render_qweb_pdf(quote_report.report_name, [so.id])
show('quote PDF size', f'{len(pdf)/1024:.1f} KB (body must be non-empty)')
so.action_confirm()
show('SO confirmed', f'state={so.state}')
# Auto-MO from our SO-confirm hook.
mo = env['mrp.production'].search([('origin', '=', so.name)], limit=1)
show('MO auto-created', f'{mo.name} (state={mo.state})' if mo else '(MISSING)')
if mo and mo.state == 'draft':
mo.action_confirm()
show('MO confirmed', f'state={mo.state}')
job = mo.x_fc_portal_job_id if mo else env['fusion.plating.portal.job']
show('portal job auto-created',
f'{job.name} (state={job.state})' if job else '(MISSING)')
# =====================================================================
step(4, 'RECEIVING (warehouse)',
'Customer parts arrive — count + log + auto-prefill')
# =====================================================================
recv = env['fp.receiving'].create({
'partner_id': customer.id,
'sale_order_id': so.id,
'received_date': fields.Datetime.now(),
'expected_qty': 25,
'line_ids': [(0, 0, {
'description': '25 stainless brackets, in 2 kraft boxes',
'expected_qty': 25,
'received_qty': 25,
})],
})
show('receiving record', f'{recv.name} (state={recv.state})')
show('expected qty (header)', str(recv.expected_qty))
show('received qty (header)', f'{recv.received_qty} (auto-prefilled from expected_qty)')
show('matches expected', 'YES' if recv.received_qty == recv.expected_qty else 'NO')
# Walk through the inspection lifecycle as the receiver would.
try:
recv.action_start_inspection()
show(' → inspection started', f'state={recv.state}')
recv.action_accept()
show(' → parts accepted', f'state={recv.state}')
except Exception as e:
print(f' [info] inspection: {e}')
# =====================================================================
step(5, 'PLANNER',
'Assigns recipe + generates work orders from process tree')
# =====================================================================
recipe = env['fusion.plating.process.node'].search(
[('node_type', '=', 'recipe')], limit=1)
show('recipe template', recipe.name if recipe else '(none)')
if mo and recipe and not mo.x_fc_recipe_id:
mo.x_fc_recipe_id = recipe.id
mo._generate_workorders_from_recipe()
show('WOs generated', f'{len(mo.workorder_ids)} work orders')
for i, wo in enumerate(mo.workorder_ids[:5], 1):
show(f' WO {i}', f'{wo.name} @ {wo.workcenter_id.name}')
if len(mo.workorder_ids) > 5:
show('', f'and {len(mo.workorder_ids) - 5} more')
# =====================================================================
step(6, 'OPERATOR (shop floor)',
'Clocks in, starts each WO, finishes when the bath is done')
# =====================================================================
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}')
show('WOs started', ok_starts)
show('WOs finished', ok_finishes)
total_dur = sum(mo.workorder_ids.mapped('duration'))
show('total time logged on WOs', f'{total_dur:.1f} min (Odoo native time tracking)')
# =====================================================================
step(7, 'MANAGER',
'Marks MO done → triggers CoC cert + delivery auto-create')
# =====================================================================
try:
mo.button_mark_done()
except Exception as e:
print(f' [info] mark_done: {e} — falling back to _action_done')
try:
mo.qty_producing = mo.product_qty
mo._action_done()
except Exception as e2:
print(f' [info] _action_done: {e2}')
show('MO state', mo.state)
show('parts location', mo.x_fc_current_location)
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')
show('CoC cert', f'{coc.name} (state={coc.state})' if coc else '(MISSING)')
show('thickness cert', f'count={len(thickness)} (expected 0 — CoC includes it)')
if coc and coc.attachment_id:
kb = len(base64.b64decode(coc.attachment_id.datas)) / 1024
show('CoC PDF', f'{coc.attachment_id.name}{kb:.1f} KB')
show('PDF is rich (>=100 KB, not bare header)',
'YES' if kb >= 100 else 'NO')
# =====================================================================
step(8, 'SHIPPER (dispatcher + driver)',
'Schedules → driver picks up → marks delivered')
# =====================================================================
dlv = env['fusion.plating.delivery'].search(
[('partner_id', '=', customer.id)], order='id desc', limit=1)
show('delivery auto-created', f'{dlv.name} (state={dlv.state})' if dlv else '(MISSING)')
if dlv:
show(' scheduled date prefilled', str(dlv.scheduled_date or '(empty)'))
show(' driver prefilled',
dlv.assigned_driver_id.name if dlv.assigned_driver_id else '(none)')
show(' CoC cross-linked to delivery',
dlv.coc_attachment_id.name if dlv.coc_attachment_id else '(none)')
# Walk through the lifecycle as the driver / dispatcher would.
try:
if dlv.state == 'draft': dlv.action_schedule(); show(' → scheduled', dlv.state)
if dlv.state == 'scheduled': dlv.action_start_route(); show(' → en route', dlv.state)
if dlv.state == 'en_route': dlv.action_mark_delivered(); show(' → delivered', dlv.state)
except Exception as e:
print(f' [info] delivery transitions: {e}')
coc_logs = env['fusion.plating.chain.of.custody'].search(
[('delivery_id', '=', dlv.id)])
show('chain-of-custody entries', len(coc_logs))
# Portal job should have moved to shipped.
job = env['fusion.plating.portal.job'].browse(job.id)
show('portal job state', job.state)
# =====================================================================
step(9, 'ACCOUNTING',
'Creates + posts invoice (does NOT register payment — '
'real customer pays through bank/Stripe)')
# =====================================================================
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)
show('invoice created', f'{inv.name} (state={inv.state})' if inv else '(MISSING)')
if inv:
inv.invoice_date = fields.Date.today()
try:
inv.action_post()
except Exception as e:
print(f' [info] action_post: {e}')
show('invoice posted', f'state={inv.state}, payment_state={inv.payment_state}')
show('total', f'${inv.amount_total:,.2f}')
show('payment_state explanation',
'"not_paid" is correct — accountant has not yet registered any payment')
inv_report = env.ref(
'fusion_plating_reports.action_report_fp_invoice_portrait',
raise_if_not_found=False)
if inv_report:
try:
pdf, _e = inv_report.with_context(force_report_rendering=True
)._render_qweb_pdf(inv_report.report_name, [inv.id])
kb = len(pdf) / 1024
show('invoice PDF size', f'{kb:.1f} KB')
show('PDF body has line items (>=50 KB, not empty)',
'YES' if kb >= 50 else 'NO')
except Exception as e:
print(f' [info] invoice render: {e}')
show('portal job state (after invoice)',
env['fusion.plating.portal.job'].browse(job.id).state)
# =====================================================================
step(10, 'AUDIT',
'What the system logged for this customer')
# =====================================================================
logs = env['fp.notification.log'].search(
[('sale_order_id', '=', so.id)], order='create_date')
show('notification log entries', len(logs))
for l in logs:
show(f'{l.trigger_event}',
f'{l.status} | to {l.recipient_email or "(no email)"} | '
f'attached: {l.attachment_names or "(none)"}')
print('\n ════════════════════════════════════════════════════════════════════')
print(' END-TO-END SUMMARY')
print(' ════════════════════════════════════════════════════════════════════')
show('customer', customer.name)
show('RFQ', rfq.name)
show('SO', f'{so.name} (PO: {so.client_order_ref})')
show('MO', f'{mo.name}{mo.state}')
show('receiving', recv.name)
show('CoC cert', coc.name if coc else '(none)')
show('delivery', f'{dlv.name}{dlv.state}' if dlv else '(none)')
show('invoice', f'{inv.name} → posted={inv.state == "posted"}, '
f'paid={inv.payment_state == "paid"}' if inv else '(none)')
show('portal job', f'{job.name} → final state: '
f'{env["fusion.plating.portal.job"].browse(job.id).state}')
env.cr.commit()
print('\n Changes committed. Order completed end-to-end.\n')