full_encoach_platform/smoke_provider_fallback.py

"""Smoke test for the AI provider fallback chain (Phase 24.1).

Exercises the new code paths added in this session:

  1. provider_router.classify_provider_error  — error taxonomy
  2. provider_router.resolve_chain            — fallback ordering
  3. free_image.render_placeholder            — offline Pillow card
  4. free_tts.synthesize_silent               — offline silent MP3 stub
  5. MediaService.generate_image('pillow')   — full fallback round-trip
  6. MediaService.synthesize_audio('silent') — silent fallback persists
  7. AI settings ir.config_parameter wiring   — get/set via params

Run via:
    odoo-bin shell -c odoo.conf -d encoach_v2 < smoke_provider_fallback.py

External APIs are NEVER called by this script — the whole point is to
prove the platform stays up when paid keys are missing or exhausted.
"""

import json
import sys


def hr(title):
    print(f"\n{'='*72}\n{title}\n{'='*72}")


def fail(msg):
    print(f"  FAIL  {msg}")
    sys.exit(1)


def ok(msg):
    print(f"  PASS  {msg}")


# ------------------------------------------------------------------ Step 1
hr("1. provider_router.classify_provider_error")
from odoo.addons.encoach_ai.services.provider_router import (
    classify_provider_error,
    resolve_chain,
    CAPABILITIES,
    get_active_provider,
)


class _OpenAIInsufficientQuota(Exception):
    pass


cases = [
    (Exception('insufficient_quota: You exceeded your current quota'), 'quota'),
    (Exception('Error code: 429 - {"error": {"code": "rate_limit_exceeded"}}'), 'quota'),
    (Exception('HTTP 402 Payment Required'), 'quota'),
    (Exception('Invalid API key provided'), 'auth'),
    (Exception('AccessDenied: 403'), 'auth'),
    (Exception('Connection timeout after 30s'), 'network'),
    (Exception('Something completely unexpected'), 'other'),
]
for exc, expected in cases:
    got = classify_provider_error(exc)
    if got != expected:
        fail(f"expected {expected!r} got {got!r} for {exc}")
ok(f"all {len(cases)} error-classification cases match")


# ------------------------------------------------------------------ Step 2
hr("2. provider_router.resolve_chain")
print(f"  capabilities: {sorted(CAPABILITIES.keys())}")
chain = resolve_chain(env, 'image')
print(f"  image chain (auto): {chain}")
if 'pillow' not in chain:
    fail("pillow free fallback missing from image chain")
ok("image chain contains free fallback")

chain_audio = resolve_chain(env, 'audio')
print(f"  audio chain (auto): {chain_audio}")
if 'silent' not in chain_audio:
    fail("silent fallback missing from audio chain")
ok("audio chain ends in silent stub")

chain_forced = resolve_chain(env, 'image', requested='pillow')
print(f"  image chain when requested=pillow: {chain_forced}")
if chain_forced[0] != 'pillow':
    fail("explicit request was not pinned to front of chain")
ok("explicit provider goes to front of chain")


# ------------------------------------------------------------------ Step 3
hr("3. free_image.render_placeholder")
try:
    from odoo.addons.encoach_ai.services.free_image import render_placeholder
    png = render_placeholder(
        'Week 1 — Daily Routines',
        subtitle='CEFR A2 · Reading',
        size='512x512',
        seed=42,
    )
    if not png or len(png) < 1000:
        fail(f"rendered PNG too small: {len(png)} bytes")
    if not png.startswith(b'\x89PNG'):
        fail("rendered bytes are not a valid PNG header")
    ok(f"placeholder rendered ({len(png):,} bytes, valid PNG)")
except RuntimeError as exc:
    print(f"  SKIP  Pillow not installed: {exc}")


# ------------------------------------------------------------------ Step 4
hr("4. free_tts.synthesize_silent")
from odoo.addons.encoach_ai.services.free_tts import synthesize_silent
res = synthesize_silent(duration_seconds=2)
if not res.get('audio') or len(res['audio']) < 100:
    fail(f"silent stub too small: {len(res.get('audio') or b'')}")
if res.get('content_type') not in ('audio/mpeg', 'audio/wav'):
    fail(f"expected audio/mpeg|wav, got {res.get('content_type')}")
ok(f"silent stub: {len(res['audio'])} bytes, voice={res['voice']}, "
   f"content_type={res['content_type']}")


# ------------------------------------------------------------------ Step 5
hr("5. MediaService.generate_image (forced pillow provider)")
Plan = env['encoach.course.plan'].sudo()
plan = Plan.search([], order='id desc', limit=1)
if not plan:
    fail("No course plan in DB; seed one first")
material = plan.material_ids.filtered(
    lambda m: m.material_type == 'reading_text'
)[:1]
if not material:
    fail("Plan has no reading_text material")
print(f"  using plan #{plan.id}, material #{material.id}: {material.title!r}")

from odoo.addons.encoach_ai_course.services.media_service import MediaService
svc = MediaService(env)
image_media = svc.generate_image(material, provider='pillow')
print(f"  image media #{image_media.id} status={image_media.status} "
      f"provider={image_media.provider} size={image_media.size_bytes}")
if image_media.status != 'ready':
    fail(f"pillow image generation failed: {image_media.error}")
if image_media.provider != 'pillow':
    fail(f"expected provider=pillow, got {image_media.provider}")
ok(f"pillow fallback produced ready image of {image_media.size_bytes:,} bytes")
env.cr.commit()


# ------------------------------------------------------------------ Step 6
hr("6. MediaService.synthesize_audio (forced silent provider)")
mat_listening = plan.material_ids.filtered(
    lambda m: m.material_type == 'listening_script'
)[:1]
if not mat_listening:
    print("  SKIP  no listening_script material on plan")
else:
    audio_media = svc.synthesize_audio(mat_listening, provider='silent')
    print(f"  audio media #{audio_media.id} status={audio_media.status} "
          f"provider={audio_media.provider}")
    if audio_media.status != 'ready':
        fail(f"silent audio generation failed: {audio_media.error}")
    if audio_media.provider != 'silent':
        fail(f"expected provider=silent, got {audio_media.provider}")
    ok(f"silent fallback produced ready audio of {audio_media.size_bytes:,} bytes")
    env.cr.commit()


# ------------------------------------------------------------------ Step 7
hr("7. ir.config_parameter wiring (provider switch)")
Param = env['ir.config_parameter'].sudo()
original = Param.get_param('encoach.ai.image_provider', '')
print(f"  current encoach.ai.image_provider = {original!r}")
Param.set_param('encoach.ai.image_provider', 'pillow')
got_after = get_active_provider(env, 'image')
print(f"  after set: {got_after!r}")
if got_after != 'pillow':
    fail(f"provider switch did not persist: got {got_after!r}")
# Restore so we don't leave the DB in test state
Param.set_param('encoach.ai.image_provider', original or 'auto')
ok("provider switch read back correctly (no caching)")


hr("DONE — All 7 fallback checks passed")
print("Free providers verified end-to-end. Paid providers will continue")
print("to work whenever credentials are configured; quota errors silently")
print("degrade to the same free chain we just exercised.")