Odoo-Modules/fusion_plating/fusion_plating/scripts/bt_e2e_anodize.py

# -*- coding: utf-8 -*-
"""End-to-end battle test — full Anodize job for ABC Manufacturing.

Phases:
  A. Auto-infer default_kind on existing Anodize recipe steps + seed prompts
  B. Create SO line + confirm (trigger job creation)
  C. Walk every step recording values exercising every input type
  D. Generate CoC report; verify all values rendered

Reports PASS/FAIL/findings as it goes.
"""

import json
import logging
from odoo import fields
_logger = logging.getLogger('bt_e2e_anodize')

findings = []  # list of (severity, message)
def find(severity, msg):
    findings.append((severity, msg))
    print('[%s] %s' % (severity, msg))

Node = env['fusion.plating.process.node']
NodeInput = env['fusion.plating.process.node.input']
Template = env['fp.step.template']

# ============================================================
# Phase A — auto-seed prompts on the Anodize recipe
# ============================================================
print('\n========== PHASE A: seed prompts on Anodize recipe ==========\n')

KIND_KEYWORDS = [
    ('cleaning', ['solvent clean', 'alkaline clean', 'soak clean']),
    ('etch', ['etch', 'deoxidize', 'desmut', 'activation', 'acid dip']),
    ('rinse', ['rinse']),
    ('plate', ['anodize', 'plat', 'e-nickel', 'enp', 'chrome']),
    ('bake', ['bake', 'seal', 'hot water seal']),
    ('racking', ['racking', 'rack ', 'rack)']),
    ('derack', ['unrack', 'de-rack', 'derack']),
    ('mask', ['masking', 'apply mask']),
    ('demask', ['de-mask', 'demask', 'unmask']),
    ('dry', ['dry', 'drying']),
    ('inspect', ['inspect']),
    ('final_inspect', ['final inspect', 'final inspection']),
    ('wbf_test', ['water break', 'wbf']),
    ('ship', ['ship', 'shipping']),
    ('packaging', ['packag', 'pack ']),
    ('contract_review', ['contract review', 'qa-005']),
    ('receiving', ['receiving', 'incoming']),
    ('electroclean', ['electroclean', 'electro clean']),
    ('strike', ['strike', 'wood', 'activation strike']),
    ('hardness_test', ['hardness']),
    ('adhesion_test', ['adhesion']),
    ('salt_spray', ['salt spray', 'corrosion test']),
    ('replenishment', ['replenish', 'top-up']),
]

def infer_kind(name):
    n = (name or '').lower()
    for kind, kws in KIND_KEYWORDS:
        if any(kw in n for kw in kws):
            return kind
    return None

def collect_step_nodes(root, out):
    if root.node_type in ('step', 'operation'):
        out.append(root)
    for c in root.child_ids:
        collect_step_nodes(c, out)

anodize = Node.browse(136)
all_steps = []
collect_step_nodes(anodize, all_steps)
find('INFO', 'Anodize recipe has %d step/operation nodes' % len(all_steps))

# 1. Set default_kind by inference on steps that lack one
patched_kinds = 0
for s in all_steps:
    if not s.default_kind:
        guess = infer_kind(s.name)
        if guess:
            s.default_kind = guess
            patched_kinds += 1
        else:
            # Gating-style "Ready For X" steps stay without a kind
            if 'ready' in (s.name or '').lower() or 'gating' in (s.name or '').lower():
                s.default_kind = 'gating'
                patched_kinds += 1
find('INFO', 'Patched default_kind on %d steps' % patched_kinds)

# 2. For each step with a kind, seed prompts onto the recipe node
DEFAULTS = Template.DEFAULT_INPUTS_BY_KIND
seeded_count = 0
for s in all_steps:
    if not s.default_kind or s.default_kind == 'gating':
        continue
    existing_names = set(s.input_ids.mapped('name'))
    specs = DEFAULTS.get(s.default_kind, [])
    for spec in specs:
        if spec['name'] in existing_names:
            continue
        NodeInput.create({
            'node_id': s.id,
            'name': spec['name'],
            'input_type': spec.get('input_type', 'text'),
            'kind': 'step_input',
            'collect': True,
            'sequence': spec.get('sequence', 10),
            'required': spec.get('required', False),
            'hint': spec.get('hint', ''),
            'selection_options': spec.get('selection_options', ''),
            'target_unit': spec.get('target_unit', False),
        })
        seeded_count += 1
find('INFO', 'Seeded %d prompts onto Anodize recipe steps' % seeded_count)
env.cr.commit()

# Verify
for s in all_steps[:5]:
    print('  %s (kind=%s) inputs=%d' % (
        s.name, s.default_kind or '-', len(s.input_ids)
    ))

# ============================================================
# Phase B — create SO line + confirm
# ============================================================
print('\n========== PHASE B: create SO + confirm ==========\n')

abc = env['res.partner'].browse(943)
part = env['fp.part.catalog'].browse(148)  # ABC part 4321
print('Customer:', abc.name)
print('Part:', part.part_number, 'rev', part.revision)
print('Recipe:', anodize.name, 'id=', anodize.id)

# Find a product to use on the SO line. Prefer existing service product.
prod = env['product.product'].search([], limit=1)

so = env['sale.order'].create({
    'partner_id': abc.id,
    'x_fc_planned_start_date': fields.Date.today(),
    'x_fc_po_number': 'BT-PO-E2E-001',
    'client_order_ref': 'BT-PO-E2E-001',
})
sol = env['sale.order.line'].create({
    'order_id': so.id,
    'product_id': prod.id,
    'product_uom_qty': 5,
    'price_unit': 100.0,
    'name': 'BT-E2E Anodize Test Line',
    'x_fc_part_catalog_id': part.id,
    'x_fc_process_variant_id': anodize.id,
})
find('INFO', 'Created SO %s line %d' % (so.name, sol.id))

# Confirm
try:
    so.action_confirm()
    find('PASS', 'SO %s confirmed (state=%s)' % (so.name, so.state))
except Exception as e:
    find('FAIL', 'SO confirm raised: %s' % str(e)[:200])
    raise SystemExit
env.cr.commit()

# Find resulting fp.job
job = env['fp.job'].search([('origin', '=', so.name)], limit=1)
if not job:
    find('FAIL', 'No fp.job created from SO %s' % so.name)
    raise SystemExit
find('PASS', 'fp.job %d created' % job.id)

job_steps = env['fp.job.step'].search([('job_id', '=', job.id)], order='sequence')
find('INFO', 'job has %d steps' % len(job_steps))
for js in job_steps[:5]:
    rn = js.recipe_node_id
    print('  step %s -- recipe_node=%d inputs=%d kind=%s' % (
        js.name, rn.id if rn else 0,
        len(rn.input_ids) if rn else 0,
        rn.default_kind if rn else '-',
    ))

# ============================================================
# Phase C — record measurements covering every input type
# ============================================================
print('\n========== PHASE C: record measurements ==========\n')

# Pick steps that exercise different input types
steps_to_walk = [
    js for js in job_steps
    if js.recipe_node_id and any(
        i.kind == 'step_input' for i in js.recipe_node_id.input_ids
    )
]
find('INFO', 'walking %d steps with prompts' % len(steps_to_walk))

# Helper: simulate the wizard + commit for a single step
SAMPLE_VALUES = {
    'text': 'TEST-VAL',
    'number': 42.5,
    'boolean': True,
    'selection': None,  # filled in per-prompt from selection_options
    'date': fields.Datetime.now(),
    'signature': 'JD',
    'time_hms': 1800,  # seconds = 30 min
    'time_seconds': 600,
    'temperature': 185.5,
    'thickness': 0.0012,
    'pass_fail': True,
    'photo': None,  # binary attachment created on demand
    'multi_point_thickness': [0.0010, 0.0011, 0.0012, 0.0011, 0.0013],
    'bath_chemistry_panel': {'ph': 4.8, 'concentration': 18.5,
                              'temperature': 75.0, 'bath_id': 'TANK-A1'},
    'ph': 4.8,
}

Move = env['fp.job.step.move']
Value = env['fp.job.step.move.input.value']
Att = env['ir.attachment']

types_exercised = set()
total_values = 0
errors = []

for js in steps_to_walk[:20]:
    rn = js.recipe_node_id
    prompts = rn.input_ids.filtered(lambda i: i.kind == 'step_input' and i.collect)
    if not prompts:
        continue
    try:
        move = Move.create({
            'job_id': job.id,
            'from_step_id': js.id,
            'to_step_id': js.id,
            'transfer_type': 'step',
            'qty_moved': job.qty or 1,
            'moved_by_user_id': env.user.id,
        })
    except Exception as e:
        errors.append('Move create on step %s: %s' % (js.name, str(e)[:120]))
        continue

    for p in prompts:
        types_exercised.add(p.input_type)
        vals = {
            'move_id': move.id,
            'node_input_id': p.id,
        }
        try:
            t = p.input_type
            if t == 'photo':
                # Create a tiny attachment
                att = Att.create({
                    'name': 'bt_photo.png',
                    'datas': b'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII=',
                    'res_model': 'fp.job.step.move',
                    'res_id': move.id,
                })
                vals['value_attachment_id'] = att.id
            elif t == 'multi_point_thickness':
                pts = SAMPLE_VALUES['multi_point_thickness']
                avg = sum(pts) / len(pts)
                vals['value_text'] = json.dumps({'readings': pts, 'avg': avg})
                vals['value_number'] = avg
            elif t == 'bath_chemistry_panel':
                vals['value_text'] = json.dumps(SAMPLE_VALUES['bath_chemistry_panel'])
            elif t == 'ph':
                vals['value_number'] = SAMPLE_VALUES['ph']
            elif t == 'pass_fail' or t == 'boolean':
                vals['value_boolean'] = True
            elif t == 'date':
                vals['value_date'] = fields.Datetime.now()
            elif t == 'selection':
                opts = (p.selection_options or '').split(',')
                vals['value_text'] = opts[0].strip() if opts else 'option_a'
            elif t in ('number', 'temperature', 'thickness', 'time_seconds'):
                vals['value_number'] = SAMPLE_VALUES.get(t, 1.0)
            elif t == 'time_hms':
                vals['value_number'] = SAMPLE_VALUES['time_hms']
            elif t == 'signature':
                vals['value_text'] = SAMPLE_VALUES['signature']
            else:
                vals['value_text'] = SAMPLE_VALUES.get(t, 'TEST')
            Value.create(vals)
            total_values += 1
        except Exception as e:
            errors.append('Value create on prompt %s (type %s): %s' % (
                p.name, p.input_type, str(e)[:120]
            ))

env.cr.commit()
find('INFO', 'Recorded %d values across %d steps' % (total_values, len(steps_to_walk)))
find('INFO', 'Exercised input types: %s' % sorted(types_exercised))
if errors:
    for e in errors[:10]:
        find('FAIL', e)
else:
    find('PASS', 'No value-creation errors')

# ============================================================
# Phase D — verify CoC chronological can render
# ============================================================
print('\n========== PHASE D: render CoC report ==========\n')

# Try rendering the chronological CoC for this job.
try:
    # The report is on fp.job (or sale.order). Let me just call the QWeb
    # render directly to catch template errors.
    moves = Move.search([('job_id', '=', job.id)], order='move_datetime')
    find('INFO', '%d moves on job %d' % (len(moves), job.id))
    # Index captured values by node_input_id
    for mv in moves[:5]:
        cvs = mv.transition_input_value_ids
        print('  move id=%d step=%s captured=%d' % (
            mv.id, mv.to_step_id.name, len(cvs)
        ))

    # Try to render the report XML
    Report = env['ir.actions.report']
    coc_report = env.ref(
        'fusion_plating_reports.action_report_coc_chronological',
        raise_if_not_found=False,
    )
    if coc_report:
        try:
            html, _ = coc_report._render_qweb_html([job.id])
            find('PASS', 'CoC report rendered, len=%d' % len(html or ''))
            # Spot-check it includes a recorded value
            html_str = (html or b'').decode('utf-8', errors='ignore') if isinstance(html, bytes) else str(html)
            if 'TEST-VAL' in html_str or '42.5' in html_str or '0.0011' in html_str:
                find('PASS', 'CoC report contains at least one recorded value')
            else:
                find('FAIL', 'CoC report rendered but no recorded value found in body')
        except Exception as e:
            find('FAIL', 'CoC render exception: %s' % str(e)[:200])
    else:
        find('FAIL', 'CoC chronological report action not found')
except Exception as e:
    find('FAIL', 'CoC verification step crashed: %s' % str(e)[:200])

# ============================================================
# Summary
# ============================================================
print('\n========== SUMMARY ==========')
passes = sum(1 for f in findings if f[0] == 'PASS')
fails = sum(1 for f in findings if f[0] == 'FAIL')
infos = sum(1 for f in findings if f[0] == 'INFO')
print('PASS=%d  FAIL=%d  INFO=%d' % (passes, fails, infos))
print('SO:', so.name, 'JOB:', job.id)