Mangalord/backend/tests/analysis_ocr.rs

//! Integration tests for the OCR analysis backend
//! (`analysis::ocr::OcrAnalyzeDispatcher`). A stub OCR engine stands in for
//! `ocrs` (whose `.rten` models aren't shipped to CI), so these pin the
//! storage→OCR→persist wiring: the dispatcher reads the page image, runs the
//! engine, and persists the lines via the shared `persist_analysis` path —
//! landing `page_ocr_text` rows and a populated `search_doc` exactly like the
//! vision backend. Each `#[sqlx::test]` gets a fresh migrated DB.

mod common;

use std::sync::Arc;

use mangalord::analysis::daemon::AnalyzeDispatcher;
use mangalord::analysis::ocr::test_support::StubOcrEngine;
use mangalord::analysis::ocr::OcrAnalyzeDispatcher;
use mangalord::domain::page_analysis::AnalysisStatus;
use mangalord::repo;
use mangalord::storage::{LocalStorage, Storage};
use sqlx::PgPool;
use tempfile::TempDir;
use uuid::Uuid;

/// Seed a manga → chapter → page chain whose page points at `storage_key`,
/// and return the page id.
async fn seed_page(pool: &PgPool, storage_key: &str) -> Uuid {
    let manga_id: Uuid =
        sqlx::query_scalar("INSERT INTO mangas (title) VALUES ('M') RETURNING id")
            .fetch_one(pool)
            .await
            .unwrap();
    let chapter_id: Uuid = sqlx::query_scalar(
        "INSERT INTO chapters (manga_id, number) VALUES ($1, 1) RETURNING id",
    )
    .bind(manga_id)
    .fetch_one(pool)
    .await
    .unwrap();
    sqlx::query_scalar(
        "INSERT INTO pages (chapter_id, page_number, storage_key, content_type) \
         VALUES ($1, 1, $2, 'image/png') RETURNING id",
    )
    .bind(chapter_id)
    .bind(storage_key)
    .fetch_one(pool)
    .await
    .unwrap()
}

fn ocr_dispatcher(
    pool: &PgPool,
    storage: Arc<dyn Storage>,
    lines: &[&str],
) -> OcrAnalyzeDispatcher {
    OcrAnalyzeDispatcher {
        db: pool.clone(),
        storage,
        engine: StubOcrEngine::new(lines),
        max_image_bytes: 8 * 1024 * 1024,
        ocr_permits: Arc::new(tokio::sync::Semaphore::new(1)),
    }
}

#[sqlx::test(migrations = "./migrations")]
async fn dispatch_persists_ocr_lines_and_search_doc(pool: PgPool) {
    let dir = TempDir::new().unwrap();
    let storage: Arc<dyn Storage> = Arc::new(LocalStorage::new(dir.path()));
    let key = "mangas/x/p1.png";
    storage.put(key, &common::fake_png_bytes()).await.unwrap();
    let page_id = seed_page(&pool, key).await;

    let dispatcher = ocr_dispatcher(&pool, Arc::clone(&storage), &["Hello there", "general"]);
    dispatcher.dispatch(page_id).await.unwrap();

    // Two OCR rows, in order, with the recognized text.
    let rows: Vec<(String, i32)> = sqlx::query_as(
        "SELECT text, ord FROM page_ocr_text WHERE page_id = $1 ORDER BY ord",
    )
    .bind(page_id)
    .fetch_all(&pool)
    .await
    .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].0, "Hello there");
    assert_eq!(rows[1].0, "general");

    // The analysis row is `done`, stamped with the ocrs model label, and has a
    // non-empty tsvector so text search works.
    let row = repo::page_analysis::load(&pool, page_id).await.unwrap().unwrap();
    assert_eq!(row.status, AnalysisStatus::Done);
    assert_eq!(row.model.as_deref(), Some("ocrs"));
    let has_doc: bool = sqlx::query_scalar(
        "SELECT search_doc IS NOT NULL AND search_doc != ''::tsvector \
           FROM page_analysis WHERE page_id = $1",
    )
    .bind(page_id)
    .fetch_one(&pool)
    .await
    .unwrap();
    assert!(has_doc, "search_doc must be populated from OCR text");
}

#[sqlx::test(migrations = "./migrations")]
async fn dispatch_rejects_page_image_over_the_byte_cap(pool: PgPool) {
    let dir = TempDir::new().unwrap();
    let storage: Arc<dyn Storage> = Arc::new(LocalStorage::new(dir.path()));
    let key = "mangas/x/big.png";
    // 4 KiB on disk, 1 KiB cap — the streamed read must bail on the cap
    // rather than buffering the whole blob and OCR-ing it.
    storage.put(key, &vec![0u8; 4096]).await.unwrap();
    let page_id = seed_page(&pool, key).await;

    let dispatcher = OcrAnalyzeDispatcher {
        db: pool.clone(),
        storage: Arc::clone(&storage),
        engine: StubOcrEngine::new(&["should not run"]),
        max_image_bytes: 1024,
        ocr_permits: Arc::new(tokio::sync::Semaphore::new(1)),
    };
    let err = dispatcher.dispatch(page_id).await.unwrap_err();
    assert!(
        err.chain().any(|c| c.to_string().contains("cap")),
        "expected an over-cap error, got: {err:#}"
    );

    // A rejected page must not land an analysis row.
    assert!(repo::page_analysis::load(&pool, page_id)
        .await
        .unwrap()
        .is_none());
}

#[sqlx::test(migrations = "./migrations")]
async fn dispatch_missing_page_is_noop(pool: PgPool) {
    let dir = TempDir::new().unwrap();
    let storage: Arc<dyn Storage> = Arc::new(LocalStorage::new(dir.path()));
    // A page id that was never inserted — the dispatcher must treat it as a
    // deleted page and succeed without writing anything.
    let dispatcher = ocr_dispatcher(&pool, storage, &["whatever"]);
    dispatcher.dispatch(Uuid::new_v4()).await.unwrap();
}