This commit is contained in:
gsinghpal
2026-04-27 00:11:18 -04:00
parent d9f58b9851
commit f08f328688
116 changed files with 9891 additions and 359 deletions

View File

@@ -0,0 +1,395 @@
# -*- coding: utf-8 -*-
# E2E persona walk — order entry from start to finish.
#
# Personas:
# Sarah — Customer Service Rep
# Mike — Receiver
# Carlos — Plating Operator
# Lisa — QC Inspector
# Tom — Shipper
# Jane — Accounting
#
# This script fills every visible-to-operator field per step, walks the
# workflow with no shortcuts, asserts the data is sane after each phase,
# and prints what's actually visible in each form view.
import logging
import traceback
from datetime import date, timedelta
_log = logging.getLogger(__name__)
def section(title):
print(f'\n{"="*72}\n{title}\n{"="*72}')
def step(persona, msg):
print(f' [{persona:>7}] {msg}')
def fail(persona, msg):
print(f' [{persona:>7}] ❌ {msg}')
def find(persona, msg):
print(f' [{persona:>7}] 🔍 GAP: {msg}')
def e2e(env):
findings = []
# ----- pick a real partner with a recipe-able product -----
section('SETUP — pick a customer + a part already in the catalog')
Partner = env['res.partner']
Part = env['fp.part.catalog']
Coating = env['fp.coating.config']
partner = Partner.search([
('customer_rank', '>', 0),
('x_fc_account_hold', '=', False),
], limit=1)
if not partner:
partner = Partner.search([('customer_rank', '>', 0)], limit=1)
part = Part.search([('x_fc_default_coating_config_id', '!=', False)], limit=1) \
or Part.search([], limit=1)
coating = part.x_fc_default_coating_config_id \
if part.x_fc_default_coating_config_id \
else Coating.search([], limit=1)
step('Sarah', f'Customer: {partner.display_name} (id={partner.id})')
step('Sarah', f'Part: {part.part_number or part.name} rev {part.revision or "?"} (id={part.id})')
step('Sarah', f'Coating: {coating.display_name if coating else "NONE"} (id={coating.id if coating else 0})')
if not coating:
findings.append('No fp.coating.config found in DB — cannot create realistic SO')
return findings
# ----- Sarah builds a sale order -----
section('PHASE 1 — Sarah (CSR) creates the sale order')
SO = env['sale.order']
SOL = env['sale.order.line']
so_vals = {
'partner_id': partner.id,
'x_fc_po_number': f'PO-E2E-{date.today():%y%m%d}',
'x_fc_customer_job_number': 'CUSTJOB-001',
'x_fc_contact_phone': '+1-555-0100',
'x_fc_ship_via': 'Customer pickup',
'x_fc_planned_start_date': date.today() + timedelta(days=2),
'x_fc_internal_deadline': date.today() + timedelta(days=10),
'commitment_date': date.today() + timedelta(days=14),
'x_fc_invoice_strategy': 'net_terms',
'x_fc_delivery_method': 'shipping_partner',
'x_fc_rush_order': False,
'x_fc_is_blanket_order': False,
'x_fc_internal_note': 'E2E test SO — full persona walk.',
'x_fc_external_note': 'Standard plating per spec.',
}
so = SO.create(so_vals)
step('Sarah', f'Created SO {so.name} (id={so.id})')
# add a line — fill the part / coating / treatment fields
product = env['product.product'].search([('sale_ok', '=', True)], limit=1)
if not product:
findings.append('No saleable product available for SO line')
return findings
line_vals = {
'order_id': so.id,
'product_id': product.id,
'product_uom_qty': 25,
'name': f'{part.part_number or part.name} — Plating per coating spec',
'x_fc_part_catalog_id': part.id,
'x_fc_coating_config_id': coating.id,
'x_fc_internal_description': 'Process via standard recipe; bake ASAP.',
'x_fc_job_number': 'INTJOB-001',
}
line = SOL.create(line_vals)
step('Sarah', f'Added line: {line.product_uom_qty} × {line.name[:40]}')
# confirm — does account hold block?
if partner.x_fc_account_hold:
find('Sarah', 'Customer is on account hold; SO confirm should block (or warn)')
try:
so.action_confirm()
step('Sarah', f'SO confirmed → state={so.state}')
except Exception as e:
fail('Sarah', f'SO confirm raised: {e}')
findings.append(f'SO confirm failure: {e}')
return findings
# ----- side effects: fp.job created? receiving created? -----
Job = env['fp.job']
Receiving = env['fp.receiving']
PortalJob = env['fusion.plating.portal.job']
jobs = Job.search([('sale_order_id', '=', so.id)])
receivings = Receiving.search([('sale_order_id', '=', so.id)])
portal_jobs = PortalJob.search([('x_fc_job_id', 'in', jobs.ids)])
step('Sarah', f'After confirm: {len(jobs)} fp.job, {len(receivings)} fp.receiving, {len(portal_jobs)} portal.job')
if not jobs:
find('Sarah', 'NO fp.job auto-created on SO confirm! Operator has nothing to work.')
findings.append('SO confirm did not auto-spawn fp.job')
if not receivings:
find('Sarah', 'NO fp.receiving auto-created on SO confirm! Receiver has nothing to track.')
findings.append('SO confirm did not auto-spawn fp.receiving')
if jobs and not portal_jobs:
find('Sarah', 'fp.job exists but no portal.job mirror — customer can\'t track on portal.')
findings.append('Portal job mirror missing post-confirm')
# smart-button visibility check
so._compute_smart_button_visibility()
so._compute_fp_qc_counts()
step('Sarah', f'SO smart buttons: BOM Items visible? {so.x_fc_distinct_part_count >= 2} (count={so.x_fc_distinct_part_count}); '
f'By Job Group visible? {so.x_fc_has_wo_group_tag}; '
f'NCRs visible? {so.fp_qc_ncr_count_so > 0} (count={so.fp_qc_ncr_count_so})')
# ----- Mike receives parts -----
section('PHASE 2 — Mike (Receiver) processes inbound parts')
receiving = receivings[:1]
if not receiving:
receiving = Receiving.create({
'sale_order_id': so.id,
'expected_qty': 25,
})
step('Mike', f'Manually created receiving {receiving.name} (auto-create did not fire)')
find('Mike', 'Had to manually create receiving — auto-create from SO confirm is missing')
findings.append('Auto-receiving on SO confirm not wired')
else:
step('Mike', f'Found auto-created receiving {receiving.name} (state={receiving.state})')
# operator fills carrier + box count
receiving.write({
'carrier_name': 'Purolator Ground',
'carrier_tracking': 'PUR-1Z9999E2E',
'box_count_in': 3,
'received_qty': 25,
})
step('Mike', f'Set box_count_in={receiving.box_count_in}, carrier={receiving.carrier_name}')
# walk the state machine: draft → counted → staged → closed
try:
receiving.action_mark_counted()
step('Mike', f'Marked Counted → state={receiving.state}, SO status={so.x_fc_receiving_status}')
assert receiving.state == 'counted'
assert so.x_fc_receiving_status == 'partial', f'Expected partial after Counted, got {so.x_fc_receiving_status}'
except AssertionError as e:
fail('Mike', str(e))
findings.append(f'Receiving status mismatch after Counted: {e}')
except Exception as e:
fail('Mike', f'action_mark_counted failed: {e}')
findings.append(f'action_mark_counted: {e}')
try:
receiving.action_mark_staged()
step('Mike', f'Marked Staged → state={receiving.state}, SO status={so.x_fc_receiving_status}')
assert receiving.state == 'staged'
assert so.x_fc_receiving_status == 'partial'
except Exception as e:
fail('Mike', f'action_mark_staged failed: {e}')
findings.append(f'action_mark_staged: {e}')
try:
receiving.action_close()
step('Mike', f'Closed receiving → state={receiving.state}, SO status={so.x_fc_receiving_status}')
assert receiving.state == 'closed'
assert so.x_fc_receiving_status == 'received'
except Exception as e:
fail('Mike', f'action_close failed: {e}')
findings.append(f'receiving action_close: {e}')
# racking inspection should exist
if 'fp.racking.inspection' in env:
Inspection = env['fp.racking.inspection']
racks = Inspection.search([('sale_order_id', '=', so.id)])
step('Mike', f'Racking inspections for this SO: {len(racks)}')
if not racks:
find('Mike', 'Racking inspection NOT auto-created — racking crew has nothing to walk.')
findings.append('No racking inspection auto-created post-confirm')
# ----- Carlos works the plating job -----
section('PHASE 3 — Carlos (Operator) walks the plating job')
if not jobs:
fail('Carlos', 'No job to work — SO confirm did not spawn one. Skipping phase.')
else:
job = jobs[0]
step('Carlos', f'Job {job.name}: state={job.state}, qty={job.qty}, deadline={job.date_deadline}')
step('Carlos', f'Steps: {len(job.step_ids)} — recipe={job.recipe_id.name or "(none)"}')
if not job.step_ids:
find('Carlos', f'Job has zero steps! Recipe not assigned or not generated. Recipe field: {job.recipe_id}')
findings.append('Job confirmed with zero steps')
if job.step_ids:
first_step = job.step_ids.sorted('sequence')[0]
step('Carlos', f'Starting step {first_step.sequence}: {first_step.name}')
try:
first_step.button_start()
step('Carlos', f'After start: state={first_step.state}, started_by={first_step.started_by_user_id.name if first_step.started_by_user_id else "(none)"}')
except Exception as e:
fail('Carlos', f'button_start failed: {e}')
findings.append(f'step button_start: {e}')
try:
first_step.button_finish()
step('Carlos', f'After finish: state={first_step.state}, duration_actual={first_step.duration_actual}')
except Exception as e:
fail('Carlos', f'button_finish failed: {e}')
findings.append(f'step button_finish: {e}')
# walk the rest at warp speed
for s in job.step_ids.sorted('sequence')[1:]:
try:
if s.state == 'pending':
s.button_start()
if s.state == 'in_progress':
s.button_finish()
except Exception as e:
fail('Carlos', f'step {s.name} walk: {e}')
findings.append(f'step walk {s.name}: {e}')
done_count = len(job.step_ids.filtered(lambda st: st.state == 'done'))
step('Carlos', f'Walked {done_count}/{len(job.step_ids)} steps to done')
# try to mark job done — should hit QC gate if customer requires QC
wants_qc = 'x_fc_requires_qc' in partner._fields and partner.x_fc_requires_qc
step('Carlos', f'Customer requires QC? {wants_qc}')
try:
job.button_mark_done()
step('Carlos', f'Job done → state={job.state}, finished={job.date_finished}')
except Exception as e:
if wants_qc:
step('Carlos', f'(Expected) QC gate fired: {str(e)[:120]}')
else:
fail('Carlos', f'button_mark_done unexpectedly failed: {e}')
findings.append(f'button_mark_done: {e}')
# ----- Lisa runs QC -----
section('PHASE 4 — Lisa (QC) walks the checklist (if any)')
QC = env['fusion.plating.quality.check']
qcs = QC.search([('job_id', 'in', jobs.ids)]) if jobs else QC.browse()
step('Lisa', f'QC checks for this job: {len(qcs)}')
if jobs and 'x_fc_requires_qc' in partner._fields and partner.x_fc_requires_qc and not qcs:
find('Lisa', 'Customer requires QC but no QC check auto-spawned!')
findings.append('QC gate fired but no check spawned')
for qc in qcs:
step('Lisa', f'QC {qc.name}: state={qc.state}, lines={len(qc.line_ids)}')
# try to pass it
for ln in qc.line_ids:
try:
ln.write({'result': 'pass'})
except Exception:
pass
try:
qc.action_pass()
step('Lisa', f'After action_pass: state={qc.state}')
except Exception as e:
fail('Lisa', f'action_pass failed: {e}')
findings.append(f'qc action_pass: {e}')
# retry job done if blocked
if jobs:
job = jobs[0]
if job.state != 'done':
try:
job.button_mark_done()
step('Lisa', f'Job marked done after QC pass → state={job.state}')
except Exception as e:
fail('Lisa', f'Job still blocked: {e}')
findings.append(f'Job blocked post-QC: {e}')
# ----- Tom ships -----
section('PHASE 5 — Tom (Shipper) prepares the delivery')
Delivery = env['fusion.plating.delivery']
deliveries = Delivery.search([
'|', ('job_ref', 'in', jobs.mapped('name') if jobs else []),
('x_fc_job_id', 'in', jobs.ids) if jobs else (False, False, False),
]) if jobs else Delivery.browse()
step('Tom', f'Deliveries linked to this job: {len(deliveries)}')
if jobs and jobs[0].state == 'done' and not deliveries:
find('Tom', 'Job is done but NO delivery auto-created!')
findings.append('Delivery auto-create on job done missing')
for delivery in deliveries:
method = (
getattr(delivery, 'x_fc_delivery_method', None)
or getattr(delivery, 'delivery_method', None)
or '(no method field)'
)
step('Tom', f'Delivery {delivery.name}: state={delivery.state}, method={method}')
try:
if hasattr(delivery, 'action_schedule') and delivery.state == 'draft':
delivery.action_schedule()
step('Tom', f'Scheduled → state={delivery.state}')
except Exception as e:
fail('Tom', f'schedule: {e}')
# certificates
Cert = env['fp.certificate']
certs = Cert.search([('x_fc_job_id', 'in', jobs.ids)]) if jobs else Cert.browse()
step('Tom', f'Certificates for this job: {len(certs)}')
if jobs and jobs[0].state == 'done' and not certs:
find('Tom', 'Job done but NO certificate auto-generated.')
findings.append('Certificate auto-create missing')
# ----- Jane invoices -----
section('PHASE 6 — Jane (Accounting) creates and posts invoice')
invoices_before = env['account.move'].search_count([
('invoice_origin', '=', so.name),
])
try:
if so.invoice_status == 'to invoice':
inv_action = so._create_invoices()
step('Jane', f'Invoiced — {invoices_before}{env["account.move"].search_count([("invoice_origin","=",so.name)])} moves')
else:
step('Jane', f'invoice_status={so.invoice_status} (nothing to invoice)')
except Exception as e:
fail('Jane', f'_create_invoices failed: {e}')
findings.append(f'invoice creation: {e}')
# ----- common-sense edge case sweeps -----
section('PHASE 7 — common-sense edge case sweeps')
# smart-button results: do they actually return non-empty data?
section_name = ' smart-button result probes'
print(section_name)
if jobs:
job = jobs[0]
for action in ('action_view_fp_holds', 'action_view_fp_checks',
'action_view_fp_ncrs', 'action_view_fp_capas',
'action_view_fp_rmas'):
try:
act = getattr(job, action)()
domain = act.get('domain') or []
model = act.get('res_model')
count = env[model].search_count(domain) if model else 0
step('audit', f'{action}: model={model}, domain count={count}')
except Exception as e:
fail('audit', f'{action}: {e}')
findings.append(f'{action}: {e}')
# SO smart-buttons
for action in ('action_view_fp_holds', 'action_view_fp_checks',
'action_view_fp_ncrs_so', 'action_view_fp_capas',
'action_view_fp_rmas'):
try:
act = getattr(so, action)()
domain = act.get('domain') or []
model = act.get('res_model')
count = env[model].search_count(domain) if model else 0
step('audit', f'SO {action}: model={model}, domain count={count}')
except Exception as e:
fail('audit', f'SO {action}: {e}')
findings.append(f'SO {action}: {e}')
# final summary
section('SUMMARY')
if findings:
print(f'{len(findings)} finding(s):')
for i, f in enumerate(findings, 1):
print(f' {i}. {f}')
else:
print(' ✅ No findings — workflow is clean end-to-end.')
env.cr.commit()
return findings
# entry-point: env injected by odoo-shell
try:
findings = e2e(env) # noqa
except Exception as e:
print('FATAL:', e)
traceback.print_exc()