//! 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) { 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 { 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); } // ---- coverage / inspection ------------------------------------------------- async fn get_json(h: &common::Harness, cookie: &str, uri: &str) -> serde_json::Value { let resp = h .app .clone() .oneshot(common::get_with_cookie(uri, cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK, "GET {uri} failed"); common::body_json(resp).await } /// Seed a manga (1 chapter, 2 pages), analyze the first, return ids. async fn seed_partial(pool: &PgPool) -> (uuid::Uuid, uuid::Uuid, uuid::Uuid, uuid::Uuid) { use mangalord::domain::page_analysis::{OcrResult, SafetyFlag, VisionAnalysis}; let (manga_id, chapter_id, ids) = seed_manga_chapter_pages(pool, 2).await; repo::page_analysis::persist_analysis( pool, ids[0], &VisionAnalysis { ocr_results: vec![OcrResult { text: "Hello".into(), kind: "speech".into(), }], tagging_results: vec!["action".into(), "city".into()], scene_description: "A rainy street.".into(), safety_flag: SafetyFlag { is_nsfw: true, content_type: vec!["gore".into()], }, }, "test-model", ) .await .unwrap(); (manga_id, chapter_id, ids[0], ids[1]) } #[sqlx::test(migrations = "./migrations")] async fn coverage_mangas_reports_analyzed_over_total(pool: PgPool) { // Coverage works with the worker disabled — use the plain harness. let h = common::harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let (manga_id, _ch, _p1, _p2) = seed_partial(&pool).await; let body = get_json(&h, &cookie, "/api/v1/admin/analysis/mangas?search=M").await; let item = &body["items"][0]; assert_eq!(item["manga_id"].as_str().unwrap(), manga_id.to_string()); assert_eq!(item["total_pages"], 2); assert_eq!(item["analyzed_pages"], 1); } #[sqlx::test(migrations = "./migrations")] async fn coverage_chapters_lists_per_chapter_counts(pool: PgPool) { let h = common::harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let (manga_id, chapter_id, _p1, _p2) = seed_partial(&pool).await; let body = get_json( &h, &cookie, &format!("/api/v1/admin/analysis/mangas/{manga_id}/chapters"), ) .await; let item = &body["items"][0]; assert_eq!(item["chapter_id"].as_str().unwrap(), chapter_id.to_string()); assert_eq!(item["total_pages"], 2); assert_eq!(item["analyzed_pages"], 1); } #[sqlx::test(migrations = "./migrations")] async fn chapter_pages_reports_per_page_status(pool: PgPool) { let h = common::harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let (_m, chapter_id, p1, p2) = seed_partial(&pool).await; let body = get_json( &h, &cookie, &format!("/api/v1/admin/analysis/chapters/{chapter_id}/pages"), ) .await; let items = body["items"].as_array().unwrap(); assert_eq!(items.len(), 2); let status_for = |id: uuid::Uuid| { items .iter() .find(|i| i["page_id"].as_str().unwrap() == id.to_string()) .unwrap()["status"] .as_str() .unwrap() .to_string() }; assert_eq!(status_for(p1), "done"); assert_eq!(status_for(p2), "none"); } #[sqlx::test(migrations = "./migrations")] async fn page_detail_returns_full_result(pool: PgPool) { let h = common::harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let (_m, _ch, p1, p2) = seed_partial(&pool).await; let d = get_json(&h, &cookie, &format!("/api/v1/admin/analysis/pages/{p1}")).await; assert_eq!(d["status"], "done"); assert_eq!(d["is_nsfw"], true); assert_eq!(d["scene_description"], "A rainy street."); assert_eq!(d["model"], "test-model"); assert_eq!(d["ocr"][0]["kind"], "speech"); assert_eq!(d["ocr"][0]["text"], "Hello"); let tags: Vec<&str> = d["tags"].as_array().unwrap().iter().map(|t| t.as_str().unwrap()).collect(); assert!(tags.contains(&"action") && tags.contains(&"city")); assert_eq!(d["content_warnings"][0], "gore"); // Unanalyzed page → status "none", empty result. let d2 = get_json(&h, &cookie, &format!("/api/v1/admin/analysis/pages/{p2}")).await; assert_eq!(d2["status"], "none"); assert_eq!(d2["ocr"].as_array().unwrap().len(), 0); assert_eq!(d2["tags"].as_array().unwrap().len(), 0); } #[sqlx::test(migrations = "./migrations")] async fn page_detail_unknown_page_is_404(pool: PgPool) { let h = common::harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(common::get_with_cookie( &format!("/api/v1/admin/analysis/pages/{}", uuid::Uuid::new_v4()), &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } #[sqlx::test(migrations = "./migrations")] async fn coverage_requires_admin(pool: PgPool) { let h = common::harness(pool.clone()); let (_u, cookie) = common::register_user(&h.app).await; let resp = h .app .clone() .oneshot(common::get_with_cookie( "/api/v1/admin/analysis/mangas", &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); }