Blame for tests/test_generation_errors.py

import sqlite3
import time
from types import SimpleNamespace
from fastapi import HTTPException
from backend.generation_errors import generation_failure
from backend.store import Store
from tests.test_api_publication import setup,client,fixture_generation

def test_quota_is_actionable_without_raw_html_or_credentials():
    e=SimpleNamespace(title='ZeroGPU quota exceeded',message='You have exceeded your free ZeroGPU quota (120s requested vs. 87.5s left). Try again in 0:23:05. <a href="https://unsafe.invalid/hf_secret">token</a>')
    now=time.time();f=generation_failure(e,'inference')
    assert f.code=='gpu_quota' and '120 seconds' in f.message and '87.5 seconds' in f.message
    assert now+1384<f.retry_at<now+1387
    assert '24 minutes' in f.message and all(x not in f.message for x in ('<','unsafe','hf_secret'))

def test_other_errors_do_not_leak_and_session_has_distinct_recovery():
    for phase in ('provenance','validation_preview','normalization'):
        f=generation_failure(RuntimeError('hf_secret /private/user/file.png'),'inference' if phase=='normalization' else phase)
        assert f.code=='generation_failed' and 'hf_secret' not in f.message
    assert generation_failure(HTTPException(401,'private detail'),'session').code=='session_expired'
    for title in ('ZeroGPU queue timeout','ZeroGPU client error','ZeroGPU pending credits exceeded'):
        assert generation_failure(SimpleNamespace(title=title),'inference').code=='gpu_unavailable'

def test_failure_survives_polling_and_stays_owned(setup):
    app,_=setup;owner=client(app,'alice');other=client(app,'bob');identifier=fixture_generation(app)
    f=generation_failure(SimpleNamespace(title='ZeroGPU quota exceeded',message='You have exceeded your ZeroGPU runs limit.'),'inference')
    app.state.generations.fail(identifier,f)
    result=owner.get('/api/generations/'+identifier).json()
    assert result['error_code']=='gpu_quota' and result['error']==f.message and result['status']=='failed'
    assert other.get('/api/generations/'+identifier).status_code==404

def test_existing_private_journal_migration_is_repeatable(tmp_path):
    state=tmp_path/'state';state.mkdir()
    with sqlite3.connect(state/'journal.sqlite') as db:
        db.execute('CREATE TABLE generations (id TEXT PRIMARY KEY,owner TEXT,created REAL,error TEXT)')
        db.execute("INSERT INTO generations VALUES ('saved','owner',0,'old error')")
    Store(state);store=Store(state)
    row=store.one('SELECT * FROM generations WHERE id=?',('saved',))
    assert row['error']=='old error' and row['error_code'] is None and row['retry_at'] is None