Generalizes the admin re-enqueue beyond global backfill:
- repo::page_analysis::enqueue_pages(scope, only_unanalyzed) with
ReenqueueScope::{All,Manga,Chapter}. Fixes include-analyzed semantics:
only_unanalyzed=false now enqueues jobs with force=true so the worker
actually re-analyzes already-done pages instead of skipping them.
- POST /v1/admin/analysis/reenqueue accepts optional manga_id / chapter_id
(mutually exclusive, 404 on unknown target) alongside only_unanalyzed.
Tests: manga/chapter scope counts, include-analyzed sets force=true on the
done page, mutual-exclusivity 422, unknown-target 404. Existing global
backfill tests still green (13 total).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
412 lines
13 KiB
Rust
412 lines
13 KiB
Rust
//! Integration tests for the AI-analysis enqueue wiring: chapter upload
|
|
//! enqueues `analyze_page` jobs, and the admin backfill / force-reanalyze
|
|
//! endpoints. All gated on `analysis_enabled` (the `harness_with_analysis`
|
|
//! variant turns it on; the default harness leaves it off → 503).
|
|
|
|
mod common;
|
|
|
|
use axum::http::StatusCode;
|
|
use axum::Router;
|
|
use serde_json::json;
|
|
use sqlx::PgPool;
|
|
use tower::ServiceExt;
|
|
|
|
use mangalord::repo;
|
|
|
|
async fn seed_admin(pool: &PgPool, app: &Router) -> String {
|
|
let (username, cookie) = common::register_user(app).await;
|
|
let u = repo::user::find_by_username(pool, &username)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
repo::user::set_is_admin_unchecked(pool, u.id, true)
|
|
.await
|
|
.unwrap();
|
|
cookie
|
|
}
|
|
|
|
/// Seed a manga → chapter with `n` pages, returning (manga_id, chapter_id,
|
|
/// page_ids).
|
|
async fn seed_manga_chapter_pages(
|
|
pool: &PgPool,
|
|
n: i32,
|
|
) -> (uuid::Uuid, uuid::Uuid, Vec<uuid::Uuid>) {
|
|
let manga_id: uuid::Uuid =
|
|
sqlx::query_scalar("INSERT INTO mangas (title) VALUES ('M') RETURNING id")
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap();
|
|
let chapter_id: uuid::Uuid = sqlx::query_scalar(
|
|
"INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id",
|
|
)
|
|
.bind(manga_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap();
|
|
let mut ids = Vec::new();
|
|
for i in 1..=n {
|
|
let id: uuid::Uuid = sqlx::query_scalar(
|
|
"INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \
|
|
VALUES ($1, $2, $3, 'image/png') RETURNING id",
|
|
)
|
|
.bind(chapter_id)
|
|
.bind(i)
|
|
.bind(format!("k/{manga_id}/{i}.png"))
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap();
|
|
ids.push(id);
|
|
}
|
|
(manga_id, chapter_id, ids)
|
|
}
|
|
|
|
/// Convenience wrapper for tests that only need the page ids.
|
|
async fn seed_pages(pool: &PgPool, n: i32) -> Vec<uuid::Uuid> {
|
|
seed_manga_chapter_pages(pool, n).await.2
|
|
}
|
|
|
|
async fn analyze_job_count(pool: &PgPool) -> i64 {
|
|
sqlx::query_scalar(
|
|
"SELECT COUNT(*) FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'",
|
|
)
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn chapter_upload_enqueues_one_analysis_job_per_page(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let (_user, cookie) = common::register_user(&h.app).await;
|
|
let manga_id = common::seed_manga_via_api(&h.app, &cookie, "Uploaded").await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_multipart_with_cookie(
|
|
&format!("/api/v1/mangas/{manga_id}/chapters"),
|
|
common::MultipartBuilder::new()
|
|
.add_json("metadata", json!({ "number": 1, "title": "Ch 1" }))
|
|
.add_file("page", "p1.png", "image/png", &common::fake_png_bytes())
|
|
.add_file("page", "p2.png", "image/png", &common::fake_png_bytes()),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::CREATED);
|
|
|
|
assert_eq!(analyze_job_count(&pool).await, 2);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn chapter_upload_does_not_enqueue_when_analysis_disabled(pool: PgPool) {
|
|
// Default harness has analysis_enabled = false.
|
|
let h = common::harness(pool.clone());
|
|
let (_user, cookie) = common::register_user(&h.app).await;
|
|
let manga_id = common::seed_manga_via_api(&h.app, &cookie, "Uploaded").await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_multipart_with_cookie(
|
|
&format!("/api/v1/mangas/{manga_id}/chapters"),
|
|
common::MultipartBuilder::new()
|
|
.add_json("metadata", json!({ "number": 1 }))
|
|
.add_file("page", "p1.png", "image/png", &common::fake_png_bytes()),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::CREATED);
|
|
assert_eq!(analyze_job_count(&pool).await, 0);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_is_forbidden_for_non_admin(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let (_user, cookie) = common::register_user(&h.app).await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({}),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::FORBIDDEN);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_returns_503_when_analysis_disabled(pool: PgPool) {
|
|
let h = common::harness(pool.clone()); // analysis disabled
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({}),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_backfills_existing_pages(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
seed_pages(&pool, 3).await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({ "only_unanalyzed": true }),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::OK);
|
|
let body = common::body_json(resp).await;
|
|
assert_eq!(body["enqueued"], 3);
|
|
assert_eq!(analyze_job_count(&pool).await, 3);
|
|
|
|
// Idempotent: a second call enqueues nothing (pending jobs already exist).
|
|
let resp2 = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({ "only_unanalyzed": true }),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
let body2 = common::body_json(resp2).await;
|
|
assert_eq!(body2["enqueued"], 0);
|
|
assert_eq!(analyze_job_count(&pool).await, 3);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_scoped_to_manga(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
let (manga_a, _ch_a, _pa) = seed_manga_chapter_pages(&pool, 2).await;
|
|
seed_manga_chapter_pages(&pool, 3).await; // a different manga
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({ "manga_id": manga_a }),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::OK);
|
|
let body = common::body_json(resp).await;
|
|
assert_eq!(body["enqueued"], 2, "only manga_a's pages enqueued");
|
|
assert_eq!(analyze_job_count(&pool).await, 2);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_scoped_to_chapter(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
let (_m, chapter_id, _ids) = seed_manga_chapter_pages(&pool, 4).await;
|
|
seed_manga_chapter_pages(&pool, 5).await; // unrelated chapter
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({ "chapter_id": chapter_id }),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
let body = common::body_json(resp).await;
|
|
assert_eq!(body["enqueued"], 4);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_including_analyzed_sets_force_and_covers_done_pages(pool: PgPool) {
|
|
use mangalord::domain::page_analysis::{SafetyFlag, VisionAnalysis};
|
|
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
let ids = seed_pages(&pool, 2).await;
|
|
// First page already analyzed.
|
|
repo::page_analysis::persist_analysis(
|
|
&pool,
|
|
ids[0],
|
|
&VisionAnalysis {
|
|
ocr_results: vec![],
|
|
tagging_results: vec![],
|
|
scene_description: String::new(),
|
|
safety_flag: SafetyFlag::default(),
|
|
},
|
|
"m",
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
// only_unanalyzed=false → include the done page, with force=true.
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({ "only_unanalyzed": false }),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
let body = common::body_json(resp).await;
|
|
assert_eq!(body["enqueued"], 2, "both pages enqueued (incl. the done one)");
|
|
|
|
// The job for the already-done page must carry force=true so the worker
|
|
// re-analyzes instead of skipping it.
|
|
let force: bool = sqlx::query_scalar(
|
|
"SELECT (payload->>'force')::boolean FROM crawler_jobs \
|
|
WHERE payload->>'page_id' = $1",
|
|
)
|
|
.bind(ids[0].to_string())
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert!(force, "include-analyzed jobs must force re-analysis");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_manga_and_chapter_are_mutually_exclusive(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({ "manga_id": uuid::Uuid::new_v4(), "chapter_id": uuid::Uuid::new_v4() }),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY);
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_unknown_scope_target_is_404(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
|
|
for body in [
|
|
json!({ "manga_id": uuid::Uuid::new_v4() }),
|
|
json!({ "chapter_id": uuid::Uuid::new_v4() }),
|
|
] {
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
body,
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
|
|
}
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn reenqueue_only_unanalyzed_skips_done_pages(pool: PgPool) {
|
|
use mangalord::domain::page_analysis::{SafetyFlag, VisionAnalysis};
|
|
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
let ids = seed_pages(&pool, 2).await;
|
|
|
|
// Mark the first page as already analyzed.
|
|
repo::page_analysis::persist_analysis(
|
|
&pool,
|
|
ids[0],
|
|
&VisionAnalysis {
|
|
ocr_results: vec![],
|
|
tagging_results: vec![],
|
|
scene_description: String::new(),
|
|
safety_flag: SafetyFlag::default(),
|
|
},
|
|
"m",
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
"/api/v1/admin/analysis/reenqueue",
|
|
json!({ "only_unanalyzed": true }),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
let body = common::body_json(resp).await;
|
|
assert_eq!(body["enqueued"], 1, "only the un-analyzed page is enqueued");
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn force_analyze_page_enqueues_with_force_flag(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
let ids = seed_pages(&pool, 1).await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
&format!("/api/v1/admin/pages/{}/analyze", ids[0]),
|
|
json!({}),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::OK);
|
|
|
|
let payload: serde_json::Value = sqlx::query_scalar(
|
|
"SELECT payload FROM crawler_jobs WHERE payload->>'kind' = 'analyze_page'",
|
|
)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(payload["force"].as_bool(), Some(true));
|
|
}
|
|
|
|
#[sqlx::test(migrations = "./migrations")]
|
|
async fn force_analyze_unknown_page_is_404(pool: PgPool) {
|
|
let h = common::harness_with_analysis(pool.clone());
|
|
let cookie = seed_admin(&pool, &h.app).await;
|
|
|
|
let resp = h
|
|
.app
|
|
.clone()
|
|
.oneshot(common::post_json_with_cookie(
|
|
&format!("/api/v1/admin/pages/{}/analyze", uuid::Uuid::new_v4()),
|
|
json!({}),
|
|
&cookie,
|
|
))
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
|
|
}
|