Spaces:
Running on Zero
Running on Zero
| 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 | |