""" QA Checksheet Template Engine & Data Structures Automated Validator ------------------------------------------------------------------- Validates: 1. Database schema columns (model_specs, sub_points, reference_diagram_url, sub_point_values) 2. ISO Format Control Numbers & Revisions across all 11 templates 3. Dynamic Model Tolerances Matrix (Legacy, Starline, Skyline, Gen 4, Pro AMTS) 4. Multi-Point Sub-Points Structure (CTC Format 4 Points A through J) 5. Physical blueprint SVG files existence and permissions 6. Pydantic API response schema serialization """ import sys import os import asyncio from uuid import UUID # Ensure UTF-8 console output for Windows PowerShell if sys.platform == 'win32': sys.stdout.reconfigure(encoding='utf-8') sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from sqlalchemy import select, text from app.core.database import AsyncSessionLocal from app.models.qa_models import ( QaChecksheetTemplate, QaChecksheetSection, QaCheckpointItem, QaChecksheetInstance, QaCheckpointResponse, ) from app.modules.qa.service import get_instance_detail EXPECTED_FORMAT_CODES = { 'QS_OUTER': ('AA/QA/IPS-OP/IR-006', '00'), 'QS_INNER': ('AA/QA/IPS-IP/IR-007', '00'), 'QS_SURFACER': ('AA/QA/IPS-S/IR-008', '00'), 'QS_TOP_COAT': ('AA/QA/IPS-TC/IR-009', '00'), 'QS_TRIM': ('AA/QA/IPS-TR/IR-010', '06'), 'QS_FVI': ('AA/QA/FVI/IR-011', '00'), 'QS_PDI': ('AA/QA/PDI/IR-012', '00'), 'QS_FAPS': ('AA/QA/FAPS/CS-005', '02'), 'QS_SHORTAGE': ('AA/QA/MSDO/OR-003', '00'), 'QS_VECV': ('AA/QA/VECV/FM-014', '06'), 'QS_CTC_SUBASSY': ('AA/QA/CTC/04', '05'), } async def run_qa_engine_validation(): print("=" * 70) print(" QA CHECKSHEET TEMPLATE ENGINE & DATA STRUCTURES VALIDATION") print("=" * 70) async with AsyncSessionLocal() as session: # 1. DATABASE SCHEMA & COLUMNS CHECK print("\n[TEST 1] Verifying Database Schema & Dynamic Columns...") col_query = text(""" SELECT column_name, data_type FROM information_schema.columns WHERE table_name = 'qa_checkpoint_item' AND column_name IN ('model_specs', 'sub_points', 'reference_diagram_url'); """) cols = {row[0]: row[1] for row in (await session.execute(col_query)).fetchall()} assert 'model_specs' in cols, "Missing model_specs column in qa_checkpoint_item" assert 'sub_points' in cols, "Missing sub_points column in qa_checkpoint_item" assert 'reference_diagram_url' in cols, "Missing reference_diagram_url column" print(" -> qa_checkpoint_item: model_specs (jsonb), sub_points (jsonb), reference_diagram_url (varchar) [PASSED]") resp_col_query = text(""" SELECT column_name, data_type FROM information_schema.columns WHERE table_name = 'qa_checkpoint_response' AND column_name = 'sub_point_values'; """) resp_cols = {row[0]: row[1] for row in (await session.execute(resp_col_query)).fetchall()} assert 'sub_point_values' in resp_cols, "Missing sub_point_values column in qa_checkpoint_response" print(" -> qa_checkpoint_response: sub_point_values (jsonb) [PASSED]") # 2. DOCUMENT CONTROL NUMBERS & TEMPLATES print("\n[TEST 2] Verifying ISO Document Control Numbers & Revisions...") templates = (await session.execute(select(QaChecksheetTemplate))).scalars().all() template_map = {t.template_code: t for t in templates} for code, (fmt, rev) in EXPECTED_FORMAT_CODES.items(): t = template_map.get(code) assert t is not None, f"Template code {code} missing in database!" assert t.format_no == fmt, f"Format mismatch for {code}: expected {fmt}, got {t.format_no}" assert t.revision.replace('Rev ', '') == rev, f"Revision mismatch for {code}: expected {rev}, got {t.revision}" print(f" -> {code:15} | Format: {t.format_no:22} | Rev: {t.revision:6} [OK]") # 3. DYNAMIC MODEL TOLERANCES MATRIX print("\n[TEST 3] Verifying Dynamic Model Tolerance Matrix (Front FRP Height)...") outer_t = template_map.get('QS_OUTER') items = (await session.execute( select(QaCheckpointItem) .join(QaChecksheetSection, QaChecksheetSection.section_id == QaCheckpointItem.section_id) .where(QaChecksheetSection.template_id == outer_t.template_id) )).scalars().all() frp_item = next((i for i in items if 'front frp' in i.checkpoint_desc.lower() and 'height' in i.checkpoint_desc.lower()), None) assert frp_item is not None, "Front FRP Height checkpoint not found in QS_OUTER" assert frp_item.model_specs, "model_specs is empty for Front FRP Height" required_models = ['LEGACY', 'STARLINE', 'NEW_STARLINE', 'SKYLINE', 'LEGACY_GEN4', 'PRO_AMTS', 'VE31'] for model in required_models: assert model in frp_item.model_specs, f"Model {model} missing from Front FRP specs!" spec = frp_item.model_specs[model] print(f" -> Model {model:15}: Target {spec['target']} +/- {spec['tol']} {spec['uom']} (Range: {spec['min']}-{spec['max']}) [OK]") # 4. MULTI-POINT SUB-POINTS SPECIFICATION print("\n[TEST 4] Verifying Multi-Point Clearances (Points A-J)...") door_item = next((i for i in items if i.sub_points and len(i.sub_points) >= 10), None) assert door_item is not None, "Door Aperture 10-point check not found in QS_OUTER" print(f" -> Checkpoint: #{door_item.item_no} {door_item.checkpoint_desc}") assert len(door_item.sub_points) == 10, f"Expected 10 points, got {len(door_item.sub_points)}" for sp in door_item.sub_points: print(f" Point {sp['code']}: {sp['label']:28} | Spec: {sp['spec']:10} | Tool: {sp.get('tool', '—')}") # 5. VISUAL REFERENCE BLUEPRINT DIAGRAMS print("\n[TEST 5] Verifying Physical Blueprint SVG Files on Disk...") required_diagrams = [ 'uploads/qa_diagrams/front_frp_aperture.svg', 'uploads/qa_diagrams/ctc_door_aperture.svg', 'uploads/qa_diagrams/faps_control_panel.svg', ] for rel_path in required_diagrams: full_path = os.path.abspath(rel_path) assert os.path.exists(full_path), f"Blueprint missing: {full_path}" size = os.path.getsize(full_path) assert size > 500, f"Blueprint file suspiciously small: {size} bytes" print(f" -> {rel_path:45} ({size} bytes) [EXISTS & VALID]") # 6. PYDANTIC API SERIALIZATION TEST print("\n[TEST 6] Verifying API Instance Detail Schema Serialization...") ctc_t = template_map.get('QS_CTC_SUBASSY') sample_inst = (await session.execute( select(QaChecksheetInstance) .where(QaChecksheetInstance.template_id.in_([outer_t.template_id, ctc_t.template_id])) .limit(1) )).scalars().first() if sample_inst: detail = await get_instance_detail(session, sample_inst.instance_id) print(f" -> Instance ID: {detail.instance_id}") print(f" -> Template: {detail.template_name} ({detail.format_no} {detail.revision})") print(f" -> Vehicle Specs: Model '{detail.model_number}', Body Type '{detail.body_type_name}'") print(f" -> Sections Count: {len(detail.sections)}") total_items = sum(len(sec.items) for sec in detail.sections) print(f" -> Items Serialized: {total_items} checkpoints") # Check serialization of enriched fields has_dynamic = any(it.model_specs or it.sub_points or it.reference_diagram_url for sec in detail.sections for it in sec.items) assert has_dynamic, "Enriched dynamic fields did not serialize into ItemWithResponse" print(" -> Dynamic fields (model_specs, sub_points, reference_diagram_url) serialized [PASSED]") else: print(" -> (No existing instance for QS_OUTER/QS_CTC, skipping instance API load)") print("\n" + "=" * 70) print(" ALL QA TEMPLATE ENGINE & DATA STRUCTURE TESTS PASSED (6/6)!") print("=" * 70) if __name__ == '__main__': asyncio.run(run_qa_engine_validation())