//! Integration tests for the admin crawler observability/control API. //! //! The default test harness wires `AppState.crawler = None` (no daemon), //! so the *control* endpoints return 503 and the *read* endpoints that //! work off the DB (status shell, dead-jobs list/requeue) still function. //! This is exactly the production "daemon disabled" posture. mod common; use std::time::Duration; use axum::http::StatusCode; use axum::Router; use http_body_util::BodyExt; use serde_json::json; use sqlx::PgPool; use tower::ServiceExt; use uuid::Uuid; use common::{ body_json, get, get_with_cookie, harness, harness_with_admin_origins, post_json_with_cookie, post_json_with_cookie_origin, register_user, }; async fn seed_admin(pool: &PgPool, app: &Router) -> String { let (username, cookie) = register_user(app).await; let u = mangalord::repo::user::find_by_username(pool, &username) .await .unwrap() .unwrap(); mangalord::repo::user::set_is_admin_unchecked(pool, u.id, true) .await .unwrap(); cookie } async fn seed_dead_job(pool: &PgPool, title: &str) -> Uuid { let manga_id = Uuid::new_v4(); let chapter_id = Uuid::new_v4(); sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)") .bind(manga_id) .bind(title) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)") .bind(chapter_id) .bind(manga_id) .execute(pool) .await .unwrap(); let job_id = Uuid::new_v4(); sqlx::query( "INSERT INTO crawler_jobs (id, payload, state, attempts, last_error) \ VALUES ($1, $2, 'dead', 5, 'boom')", ) .bind(job_id) .bind(json!({ "kind": "sync_chapter_content", "source_id": "target", "chapter_id": chapter_id, "source_chapter_key": "k", })) .execute(pool) .await .unwrap(); job_id } /// Seed a chapter-content job in a given state ('pending'/'running'). async fn seed_job(pool: &PgPool, title: &str, state: &str) { let manga_id = Uuid::new_v4(); let chapter_id = Uuid::new_v4(); sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)") .bind(manga_id) .bind(title) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 1)") .bind(chapter_id) .bind(manga_id) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO crawler_jobs (id, payload, state) VALUES ($1, $2, $3)") .bind(Uuid::new_v4()) .bind(json!({ "kind": "sync_chapter_content", "source_id": "target", "chapter_id": chapter_id, "source_chapter_key": "k", })) .bind(state) .execute(pool) .await .unwrap(); } /// Seed a manga with no cover + a live source row (queued for cover fetch). async fn seed_missing_cover(pool: &PgPool, title: &str) { let manga_id = Uuid::new_v4(); sqlx::query("INSERT INTO mangas (id, title, cover_image_path) VALUES ($1, $2, NULL)") .bind(manga_id) .bind(title) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO sources (id, name, base_url) VALUES ('target','T','http://x') ON CONFLICT DO NOTHING") .execute(pool) .await .unwrap(); sqlx::query( "INSERT INTO manga_sources (source_id, source_manga_key, manga_id, source_url) \ VALUES ('target', $1, $2, 'http://x/m')", ) .bind(format!("k-{manga_id}")) .bind(manga_id) .execute(pool) .await .unwrap(); } #[sqlx::test(migrations = "./migrations")] async fn active_jobs_and_covers_lists_over_http(pool: PgPool) { seed_job(&pool, "Naruto", "pending").await; seed_job(&pool, "Bleach", "running").await; seed_missing_cover(&pool, "One Piece").await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; // Queued/active chapters. let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/active-jobs", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 2); // Queued covers. let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/covers", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 1); assert_eq!(body["items"][0]["manga_title"], "One Piece"); // Both are admin-gated. let (_u, plain) = register_user(&h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/active-jobs", &plain)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } #[sqlx::test(migrations = "./migrations")] async fn get_status_requires_admin(pool: PgPool) { let h = harness(pool); // Unauthenticated → 401. let resp = h.app.clone().oneshot(get("/api/v1/admin/crawler")).await.unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); // Authenticated non-admin → 403. let (_u, cookie) = register_user(&h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } #[sqlx::test(migrations = "./migrations")] async fn get_status_reports_disabled_daemon_with_queue_counts(pool: PgPool) { seed_dead_job(&pool, "Naruto").await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; assert_eq!(body["daemon"], "disabled"); assert_eq!(body["queue"]["dead"], 1); assert_eq!(body["browser"], "down"); } #[sqlx::test(migrations = "./migrations")] async fn control_endpoints_return_503_when_daemon_disabled(pool: PgPool) { let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; for uri in [ "/api/v1/admin/crawler/run", "/api/v1/admin/crawler/reconcile", "/api/v1/admin/crawler/browser/restart", "/api/v1/admin/crawler/session/clear-expired", ] { let resp = h .app .clone() .oneshot(post_json_with_cookie(uri, json!({}), &cookie)) .await .unwrap(); assert_eq!( resp.status(), StatusCode::SERVICE_UNAVAILABLE, "{uri} should be 503 when daemon disabled" ); } } #[sqlx::test(migrations = "./migrations")] async fn status_stream_requires_admin(pool: PgPool) { let h = harness(pool); // Unauthenticated → 401. let resp = h .app .clone() .oneshot(get("/api/v1/admin/crawler/stream")) .await .unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); // Non-admin → 403. let (_u, cookie) = register_user(&h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/stream", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } #[sqlx::test(migrations = "./migrations")] async fn status_stream_emits_initial_event(pool: PgPool) { let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/stream", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let ct = resp .headers() .get(axum::http::header::CONTENT_TYPE) .and_then(|v| v.to_str().ok()) .unwrap_or_default() .to_string(); assert!(ct.starts_with("text/event-stream"), "content-type was {ct:?}"); // Accumulate frames (the immediate snapshot may arrive split across // frames) until the status payload appears, with an overall timeout so // the never-ending stream can't hang the test. let mut body = resp.into_body(); let mut acc = String::new(); let deadline = tokio::time::timeout(Duration::from_secs(5), async { loop { let Some(frame) = body.frame().await else { break }; if let Ok(data) = frame.expect("frame ok").into_data() { acc.push_str(&String::from_utf8_lossy(&data)); if acc.contains("\"daemon\"") { break; } } } }) .await; assert!(deadline.is_ok(), "did not receive status within 5s; got: {acc:?}"); assert!(acc.contains("\"daemon\""), "missing status payload: {acc}"); assert!(acc.contains("status"), "missing SSE event name: {acc}"); } #[sqlx::test(migrations = "./migrations")] async fn mutating_endpoints_reject_non_admin(pool: PgPool) { let h = harness(pool); // A logged-in non-admin must be forbidden from a mutating endpoint. let (_u, cookie) = register_user(&h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie( "/api/v1/admin/crawler/dead-jobs/requeue", json!({ "scope": "all" }), &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } // --------------------------------------------------------------------------- // CSRF Origin/Referer allowlist (T2) // --------------------------------------------------------------------------- #[sqlx::test(migrations = "./migrations")] async fn csrf_rejects_mutation_with_cross_origin_header(pool: PgPool) { let h = harness_with_admin_origins( pool.clone(), vec!["http://localhost:3000".to_string()], ); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie_origin( "/api/v1/admin/crawler/session/clear-expired", json!({}), &cookie, Some("https://evil.example.com"), None, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } #[sqlx::test(migrations = "./migrations")] async fn csrf_allows_mutation_with_allowed_origin(pool: PgPool) { let h = harness_with_admin_origins( pool.clone(), vec!["http://localhost:3000".to_string()], ); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie_origin( "/api/v1/admin/crawler/session/clear-expired", json!({}), &cookie, Some("http://localhost:3000"), None, )) .await .unwrap(); // Daemon disabled in the harness → 503 from the handler; the CSRF // gate must have let the request through before that response. assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); } #[sqlx::test(migrations = "./migrations")] async fn csrf_rejects_cookie_auth_without_origin_or_referer(pool: PgPool) { // Cookie-auth + neither Origin nor Referer was previously waved through // ("curl bypass"). It's now refused: an extension or no-referrer-policy // page can produce exactly this shape and would otherwise be a CSRF // vector. Real curl/script callers should authenticate with a bearer // token instead — see `csrf_allows_bearer_token_without_origin`. let h = harness_with_admin_origins( pool.clone(), vec!["http://localhost:3000".to_string()], ); let cookie = seed_admin(&pool, &h.app).await; // `post_json_with_cookie_origin(..., None, None)` builds the exact "no // Origin, no Referer" cookie-auth shape this test pins. let resp = h .app .clone() .oneshot(post_json_with_cookie_origin( "/api/v1/admin/crawler/session/clear-expired", json!({}), &cookie, None, None, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } #[sqlx::test(migrations = "./migrations")] async fn csrf_falls_back_to_referer_when_origin_missing(pool: PgPool) { let h = harness_with_admin_origins( pool.clone(), vec!["http://localhost:3000".to_string()], ); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie_origin( "/api/v1/admin/crawler/session/clear-expired", json!({}), &cookie, None, Some("http://localhost:3000/admin/crawler"), )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); } #[sqlx::test(migrations = "./migrations")] async fn csrf_skipped_on_safe_methods(pool: PgPool) { let h = harness_with_admin_origins( pool.clone(), vec!["http://localhost:3000".to_string()], ); let cookie = seed_admin(&pool, &h.app).await; // GET from a hostile origin is fine — browsers can't mutate via GET. let resp = h .app .clone() .oneshot(common::get_with_cookie_origin( "/api/v1/admin/crawler", &cookie, Some("https://evil.example.com"), )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); } #[sqlx::test(migrations = "./migrations")] async fn csrf_rejects_cookie_auth_when_allowlist_empty(pool: PgPool) { // Previously an empty allowlist silently skipped the CSRF check, so an // operator who forgot to set ADMIN_ALLOWED_ORIGINS shipped an unguarded // admin surface. We now fail-closed for cookie-auth admin mutations // when the allowlist is empty — operators must either set the env var // (browser deploy) or authenticate with `Authorization: Bearer …`. let h = harness_with_admin_origins(pool.clone(), vec![]); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie_origin( "/api/v1/admin/crawler/session/clear-expired", json!({}), &cookie, Some("https://evil.example.com"), None, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } // --------------------------------------------------------------------------- // Cache-Control: no-store on admin responses (S3) // --------------------------------------------------------------------------- #[sqlx::test(migrations = "./migrations")] async fn admin_responses_have_no_store(pool: PgPool) { let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let cc = resp .headers() .get(axum::http::header::CACHE_CONTROL) .and_then(|v| v.to_str().ok()) .unwrap_or_default() .to_string(); assert!( cc.contains("no-store"), "admin response missing Cache-Control: no-store (got {cc:?})" ); } #[sqlx::test(migrations = "./migrations")] async fn non_admin_responses_unaffected_by_no_store(pool: PgPool) { let h = harness(pool); let resp = h .app .clone() .oneshot(get("/api/v1/health")) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let cc = resp .headers() .get(axum::http::header::CACHE_CONTROL) .and_then(|v| v.to_str().ok()); assert!( cc.is_none() || !cc.unwrap_or_default().contains("no-store"), "non-admin response should not get no-store (got {cc:?})" ); } // --------------------------------------------------------------------------- // scope=all confirm:true guard (S1) // --------------------------------------------------------------------------- #[sqlx::test(migrations = "./migrations")] async fn requeue_all_without_confirm_rejected(pool: PgPool) { seed_dead_job(&pool, "X").await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie( "/api/v1/admin/crawler/dead-jobs/requeue", json!({ "scope": "all" }), &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); // The dead row must NOT have been touched. let state: String = sqlx::query_scalar("SELECT state FROM crawler_jobs LIMIT 1") .fetch_one(&pool) .await .unwrap(); assert_eq!(state, "dead"); } #[sqlx::test(migrations = "./migrations")] async fn requeue_all_with_confirm_flips_dead_pile(pool: PgPool) { seed_dead_job(&pool, "X").await; seed_dead_job(&pool, "Y").await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie( "/api/v1/admin/crawler/dead-jobs/requeue", json!({ "scope": "all", "confirm": true }), &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let pending: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM crawler_jobs WHERE state = 'pending'", ) .fetch_one(&pool) .await .unwrap(); assert_eq!(pending, 2); } // --------------------------------------------------------------------------- // admin_audit target_id + PHPSESSID fingerprint (S2) // --------------------------------------------------------------------------- #[sqlx::test(migrations = "./migrations")] async fn requeue_chapter_audit_records_chapter_target_id(pool: PgPool) { let job_id = seed_dead_job(&pool, "Bleach").await; let chapter_id: Uuid = sqlx::query_scalar( "SELECT (payload->>'chapter_id')::uuid FROM crawler_jobs WHERE id = $1", ) .bind(job_id) .fetch_one(&pool) .await .unwrap(); let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(post_json_with_cookie( "/api/v1/admin/crawler/dead-jobs/requeue", json!({ "scope": "chapter", "chapter_id": chapter_id }), &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let (target_kind, target_id): (String, Option) = sqlx::query_as( "SELECT target_kind, target_id FROM admin_audit \ WHERE action = 'crawler_dead_jobs_requeue' \ ORDER BY at DESC LIMIT 1", ) .fetch_one(&pool) .await .unwrap(); assert_eq!(target_kind, "chapter"); assert_eq!(target_id, Some(chapter_id)); } #[sqlx::test(migrations = "./migrations")] async fn requeue_all_audit_omits_target_id_but_logs_count(pool: PgPool) { seed_dead_job(&pool, "X").await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let _ = h .app .clone() .oneshot(post_json_with_cookie( "/api/v1/admin/crawler/dead-jobs/requeue", json!({ "scope": "all", "confirm": true }), &cookie, )) .await .unwrap(); let (target_kind, target_id, payload): (String, Option, serde_json::Value) = sqlx::query_as( "SELECT target_kind, target_id, payload FROM admin_audit \ WHERE action = 'crawler_dead_jobs_requeue' \ ORDER BY at DESC LIMIT 1", ) .fetch_one(&pool) .await .unwrap(); assert_eq!(target_kind, "crawler"); assert_eq!(target_id, None); assert_eq!(payload["scope"], "all"); assert_eq!(payload["requeued"], 1); } // --------------------------------------------------------------------------- // Job history: unified, searchable, filterable list over all states/kinds. // --------------------------------------------------------------------------- /// Seed a chapter-content job in a given state and return its (manga, chapter). async fn seed_history_chapter_job( pool: &PgPool, title: &str, number: i32, state: &str, ) -> (Uuid, Uuid) { let manga_id = Uuid::new_v4(); let chapter_id = Uuid::new_v4(); sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)") .bind(manga_id) .bind(title) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, $3)") .bind(chapter_id) .bind(manga_id) .bind(number) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO crawler_jobs (id, payload, state) VALUES ($1, $2, $3)") .bind(Uuid::new_v4()) .bind(json!({ "kind": "sync_chapter_content", "source_id": "target", "chapter_id": chapter_id, "source_chapter_key": "k", })) .bind(state) .execute(pool) .await .unwrap(); (manga_id, chapter_id) } /// Seed a `done` analyze_page job pointing at a real page so the history /// row resolves manga/chapter/page-number through the page breadcrumb. async fn seed_history_analyze_job(pool: &PgPool, title: &str, page_number: i32) { let manga_id = Uuid::new_v4(); let chapter_id = Uuid::new_v4(); let page_id = Uuid::new_v4(); sqlx::query("INSERT INTO mangas (id, title) VALUES ($1, $2)") .bind(manga_id) .bind(title) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO chapters (id, manga_id, number) VALUES ($1, $2, 3)") .bind(chapter_id) .bind(manga_id) .execute(pool) .await .unwrap(); sqlx::query( "INSERT INTO pages (id, chapter_id, page_number, storage_key, content_type) \ VALUES ($1, $2, $3, 'k', 'image/png')", ) .bind(page_id) .bind(chapter_id) .bind(page_number) .execute(pool) .await .unwrap(); sqlx::query("INSERT INTO crawler_jobs (id, payload, state) VALUES ($1, $2, 'done')") .bind(Uuid::new_v4()) .bind(json!({ "kind": "analyze_page", "page_id": page_id, "force": false })) .execute(pool) .await .unwrap(); } #[sqlx::test(migrations = "./migrations")] async fn job_history_lists_filters_and_paginates(pool: PgPool) { seed_history_chapter_job(&pool, "Naruto", 1, "done").await; seed_history_chapter_job(&pool, "Bleach", 2, "running").await; seed_dead_job(&pool, "Vinland").await; // dead sync_chapter_content seed_history_analyze_job(&pool, "One Piece", 7).await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; // Unfiltered: all four jobs. let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/history", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 4); // Filter by state. let resp = h .app .clone() .oneshot(get_with_cookie( "/api/v1/admin/crawler/history?state=done", &cookie, )) .await .unwrap(); let body = body_json(resp).await; // Naruto (sync_chapter done) + One Piece (analyze_page done). assert_eq!(body["page"]["total"], 2); // Filter by kind. let resp = h .app .clone() .oneshot(get_with_cookie( "/api/v1/admin/crawler/history?kind=analyze_page", &cookie, )) .await .unwrap(); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 1); assert_eq!(body["items"][0]["manga_title"], "One Piece"); assert_eq!(body["items"][0]["page_number"], 7); assert_eq!(body["items"][0]["chapter_number"], 3); // Search by manga title. let resp = h .app .clone() .oneshot(get_with_cookie( "/api/v1/admin/crawler/history?search=Bleach", &cookie, )) .await .unwrap(); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 1); assert_eq!(body["items"][0]["manga_title"], "Bleach"); assert_eq!(body["items"][0]["state"], "running"); // Pagination: limit clamps the page slice but total reflects the full set. let resp = h .app .clone() .oneshot(get_with_cookie( "/api/v1/admin/crawler/history?limit=1", &cookie, )) .await .unwrap(); let body = body_json(resp).await; assert_eq!(body["items"].as_array().unwrap().len(), 1); assert_eq!(body["page"]["total"], 4); // Admin-gated. let (_u, plain) = register_user(&h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/history", &plain)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } #[sqlx::test(migrations = "./migrations")] async fn job_history_surfaces_sync_manga_payload_title(pool: PgPool) { // A reconcile-enqueued sync_manga job that died has no manga row, so the // history row must fall back to the payload title/url and be searchable. sqlx::query("INSERT INTO crawler_jobs (id, payload, state, last_error) VALUES ($1, $2, 'dead', 'broken-page body signature')") .bind(Uuid::new_v4()) .bind(json!({ "kind": "sync_manga", "source_id": "target", "source_manga_key": "gone-1", "url": "http://x/manga/gone-1", "title": "Ghost Manga", })) .execute(&pool) .await .unwrap(); let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie( "/api/v1/admin/crawler/history?search=ghost", &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 1); let row = &body["items"][0]; assert_eq!(row["kind"], "sync_manga"); assert_eq!(row["manga_title"], serde_json::Value::Null); assert_eq!(row["payload_title"], "Ghost Manga"); assert_eq!(row["source_url"], "http://x/manga/gone-1"); assert_eq!(row["source_key"], "gone-1"); } // --------------------------------------------------------------------------- // Operation metrics: durations, averages, recent-ops log. // --------------------------------------------------------------------------- async fn record_metric( pool: &PgPool, op: &str, manga_id: Option, chapter_id: Option, outcome: &str, duration_ms: i64, items: Option, ) { mangalord::repo::crawl_metrics::record( pool, op, manga_id, chapter_id, outcome, duration_ms, items, None, ) .await .unwrap(); } #[sqlx::test(migrations = "./migrations")] async fn crawler_metrics_summary_and_ops_over_http(pool: PgPool) { // Two chapter ops (one failed) + a cover op. record_metric(&pool, "chapter", None, None, "ok", 6000, Some(20)).await; record_metric(&pool, "chapter", None, None, "failed", 8000, Some(10)).await; record_metric(&pool, "manga_cover", None, None, "ok", 500, None).await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; // Summary: per-type averages + success split. let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/metrics", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; let summary = body["summary"].as_array().unwrap(); let chapter = summary.iter().find(|s| s["op"] == "chapter").unwrap(); assert_eq!(chapter["n"], 2); assert_eq!(chapter["ok"], 1); assert_eq!(chapter["failed"], 1); assert_eq!(chapter["avg_ms"], 7000.0); // Recent-ops log + outcome filter. let resp = h .app .clone() .oneshot(get_with_cookie( "/api/v1/admin/crawler/metrics/ops?outcome=failed", &cookie, )) .await .unwrap(); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 1); assert_eq!(body["items"][0]["op"], "chapter"); assert_eq!(body["items"][0]["outcome"], "failed"); // Admin-gated. let (_u, plain) = register_user(&h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/metrics", &plain)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::FORBIDDEN); } #[sqlx::test(migrations = "./migrations")] async fn job_history_includes_chapter_duration(pool: PgPool) { let (_m, chapter_id) = seed_history_chapter_job(&pool, "Berserk", 12, "done").await; // A chapter metric for that chapter → history row should surface its duration. record_metric(&pool, "chapter", None, Some(chapter_id), "ok", 6100, Some(20)).await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; let resp = h .app .clone() .oneshot(get_with_cookie( "/api/v1/admin/crawler/history?kind=sync_chapter_content", &cookie, )) .await .unwrap(); let body = body_json(resp).await; assert_eq!(body["items"][0]["duration_ms"], 6100); } #[sqlx::test(migrations = "./migrations")] async fn dead_jobs_list_and_requeue_over_http(pool: PgPool) { let job_id = seed_dead_job(&pool, "Bleach").await; let h = harness(pool.clone()); let cookie = seed_admin(&pool, &h.app).await; // List. let resp = h .app .clone() .oneshot(get_with_cookie("/api/v1/admin/crawler/dead-jobs", &cookie)) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; assert_eq!(body["page"]["total"], 1); assert_eq!(body["items"][0]["manga_title"], "Bleach"); // Requeue the single job. let resp = h .app .clone() .oneshot(post_json_with_cookie( "/api/v1/admin/crawler/dead-jobs/requeue", json!({ "scope": "job", "job_id": job_id }), &cookie, )) .await .unwrap(); assert_eq!(resp.status(), StatusCode::OK); let body = body_json(resp).await; assert_eq!(body["requeued"], 1); let state: String = sqlx::query_scalar("SELECT state FROM crawler_jobs WHERE id = $1") .bind(job_id) .fetch_one(&pool) .await .unwrap(); assert_eq!(state, "pending"); }