path: root/tests/api/test_archive.py

import os
from datetime import datetime as dt
from pathlib import Path

import hircine.config
import hircine.db as database
import hircine.thumbnailer as thumb
import pytest
from conftest import DB, Response
from hircine.db.models import Archive, Comic, Image, Page
from sqlalchemy import select


@pytest.fixture
def query_archive(execute_id):
    query = """
    query archive($id: Int!) {
        archive(id: $id) {
            __typename
            ... on FullArchive {
                id
                name
                createdAt
                mtime
                size
                path
                pageCount
                organized
                comics {
                    __typename
                    id
                }
                cover {
                    __typename
                    id
                }
                pages {
                    __typename
                    id
                    image {
                        __typename
                        id
                    }
                }
            }
            ... on Error {
                message
            }
            ... on IDNotFoundError {
                id
            }
        }
    }
    """

    return execute_id(query)


@pytest.fixture
def query_archives(execute):
    query = """
    query archives {
        archives {
            __typename
            count
            edges {
                id
                name
                size
                pageCount
                organized
                cover {
                    __typename
                    id
                }
            }
        }
    }
    """

    return execute(query)


@pytest.fixture
def update_archives(execute_update):
    mutation = """
    mutation updateArchives($ids: [Int!]!, $input: UpdateArchiveInput!) {
        updateArchives(ids: $ids, input: $input) {
            __typename
            ... on Success {
                message
            }
            ... on Error {
                message
            }
            ... on PageRemoteError {
                id
                archiveId
            }
            ... on IDNotFoundError {
                id
            }
        }
    }
    """

    return execute_update(mutation)


@pytest.fixture
def delete_archives(execute_delete):
    mutation = """
    mutation deleteArchives($ids: [Int!]!) {
        deleteArchives(ids: $ids) {
            __typename
            ... on Success {
                message
            }
            ... on Error {
                message
            }
            ... on IDNotFoundError {
                id
            }
        }
    }
    """

    return execute_delete(mutation)


def assert_image_matches(obj, model):
    assert obj["__typename"] == "Image"
    assert obj["id"] == model.id


def assert_page_matches(obj, model):
    assert obj["__typename"] == "Page"
    assert obj["id"] == model.id


@pytest.mark.anyio
async def test_query_archive(query_archive, gen_archive):
    archive = next(gen_archive)
    pages = archive.pages

    await DB.add(archive)

    response = Response(await query_archive(archive.id))
    response.assert_is("FullArchive")

    assert response.id == archive.id
    assert response.name == archive.name
    assert dt.fromisoformat(response.createdAt) == archive.created_at
    assert dt.fromisoformat(response.mtime) == archive.mtime
    assert response.size == archive.size
    assert response.path == archive.path
    assert response.comics == []
    assert response.pageCount == archive.page_count
    assert response.organized == archive.organized
    assert_image_matches(response.cover, pages[0].image)

    assert len(response.pages) == len(pages)

    page_iter = iter(sorted(pages, key=lambda page: page.index))
    for page in response.pages:
        matching_page = next(page_iter)
        assert_page_matches(page, matching_page)
        assert_image_matches(page["image"], matching_page.image)


@pytest.mark.anyio
async def test_query_archive_sorts_pages(query_archive, gen_jumbled_archive):
    archive = await DB.add(next(gen_jumbled_archive))

    response = Response(await query_archive(archive.id))
    response.assert_is("FullArchive")

    page_iter = iter(sorted(archive.pages, key=lambda page: page.index))
    for page in response.pages:
        matching_page = next(page_iter)
        assert_page_matches(page, matching_page)
        assert_image_matches(page["image"], matching_page.image)


@pytest.mark.anyio
async def test_query_archive_fails_not_found(query_archive):
    response = Response(await query_archive(1))
    response.assert_is("IDNotFoundError")
    assert response.id == 1
    assert response.message == "Archive ID not found: '1'"


@pytest.mark.anyio
async def test_query_archives(query_archives, gen_archive):
    archives = await DB.add_all(*gen_archive)

    response = Response(await query_archives())
    response.assert_is("ArchiveFilterResult")

    assert response.count == len(archives)
    assert isinstance((response.edges), list)
    assert len(response.edges) == len(archives)

    edges = iter(response.edges)
    for archive in sorted(archives, key=lambda a: a.name):
        edge = next(edges)
        assert edge["id"] == archive.id
        assert edge["name"] == archive.name
        assert edge["size"] == archive.size
        assert edge["pageCount"] == archive.page_count
        assert_image_matches(edge["cover"], archive.cover)


@pytest.fixture
def gen_archive_with_files(tmpdir, monkeypatch, gen_archive):
    content_dir = os.path.join(tmpdir, "content/")
    object_dir = os.path.join(tmpdir, "objects/")
    os.mkdir(content_dir)
    os.mkdir(object_dir)

    dirs = hircine.config.DirectoryStructure(scan=content_dir, objects=object_dir)
    monkeypatch.setattr(hircine.config, "dir_structure", dirs)

    archive = next(gen_archive)

    archive_path = Path(os.path.join(content_dir, "archive.zip"))
    archive_path.touch()
    archive.path = str(archive_path)

    img_paths = []
    for page in archive.pages:
        for suffix in ["full", "thumb"]:
            img_path = Path(thumb.object_path(object_dir, page.image.hash, suffix))
            os.makedirs(os.path.dirname(img_path), exist_ok=True)
            img_path.touch()

            img_paths.append(img_path)

    yield archive, content_dir, object_dir, img_paths


@pytest.mark.anyio
async def test_delete_archive(delete_archives, gen_archive_with_files):
    archive, content_dir, object_dir, img_paths = gen_archive_with_files
    archive_path = archive.path

    archive = await DB.add(archive)
    page_ids = [page.id for page in archive.pages]
    image_ids = [page.image.id for page in archive.pages]

    response = Response(await delete_archives(archive.id))
    response.assert_is("DeleteSuccess")

    archive = await DB.get(Archive, archive.id)
    assert archive is None

    async with database.session() as s:
        db_pages = (await s.scalars(select(Page).where(Page.id.in_(page_ids)))).all()
        db_images = (
            await s.scalars(select(Image).where(Image.id.in_(image_ids)))
        ).all()

    assert db_pages == []
    assert db_images == []

    assert os.path.exists(archive_path) is False
    for img_path in img_paths:
        assert os.path.exists(img_path) is False


@pytest.mark.anyio
async def test_delete_archive_deletes_images_only_when_necessary(
    delete_archives, gen_archive_with_files, gen_archive
):
    archive, content_dir, object_dir, img_paths = gen_archive_with_files
    archive_path = archive.path

    archive = await DB.add(archive)
    page_ids = [page.id for page in archive.pages]
    image_ids = [page.image.id for page in archive.pages]

    another = next(gen_archive)
    another.pages = [
        Page(path="foo", index=1, image_id=id, archive=another) for id in image_ids
    ]
    another.cover = archive.cover
    await DB.add(another)

    response = Response(await delete_archives(archive.id))
    response.assert_is("DeleteSuccess")

    archive = await DB.get(Archive, archive.id)
    assert archive is None

    async with database.session() as s:
        db_pages = (await s.scalars(select(Page).where(Page.id.in_(page_ids)))).all()
        db_images = (
            await s.scalars(select(Image.id).where(Image.id.in_(image_ids)))
        ).all()

    assert db_pages == []
    assert db_images == image_ids

    assert os.path.exists(archive_path) is False
    for img_path in img_paths:
        assert os.path.exists(img_path) is True


@pytest.mark.anyio
async def test_delete_archive_cascades_on_comic(
    delete_archives, gen_archive_with_files
):
    archive, *_ = gen_archive_with_files
    comic = Comic(
        id=1,
        title="Hic Sunt Dracones",
        archive=archive,
        cover=archive.cover,
        pages=archive.pages,
    )

    comic = await DB.add(comic)

    response = Response(await delete_archives(comic.archive.id))
    response.assert_is("DeleteSuccess")

    archive = await DB.get(Archive, archive.id)
    assert archive is None

    comic = await DB.get(Comic, comic.id)
    assert comic is None


@pytest.mark.anyio
async def test_update_archives(update_archives, gen_archive):
    old_archive = await DB.add(next(gen_archive))

    response = Response(
        await update_archives(
            old_archive.id,
            {"cover": {"id": old_archive.pages[1].id}, "organized": True},
        )
    )
    response.assert_is("UpdateSuccess")

    archive = await DB.get(Archive, old_archive.id)

    assert archive.cover_id == old_archive.pages[1].image.id
    assert archive.organized is True


@pytest.mark.anyio
async def test_update_archive_fails_archive_not_found(update_archives, gen_archive):
    archive = await DB.add(next(gen_archive))

    response = Response(
        await update_archives(100, {"cover": {"id": archive.pages[1].id}})
    )
    response.assert_is("IDNotFoundError")
    assert response.id == 100
    assert response.message == "Archive ID not found: '100'"


@pytest.mark.anyio
async def test_update_archive_cover_fails_page_not_found(update_archives, gen_archive):
    archive = await DB.add(next(gen_archive))

    response = Response(await update_archives(archive.id, {"cover": {"id": 100}}))
    response.assert_is("IDNotFoundError")
    assert response.id == 100
    assert response.message == "Page ID not found: '100'"


@pytest.mark.anyio
async def test_update_archive_cover_fails_page_remote(update_archives, gen_archive):
    archive = await DB.add(next(gen_archive))
    another = await DB.add(next(gen_archive))
    remote_id = another.pages[0].id

    response = Response(await update_archives(archive.id, {"cover": {"id": remote_id}}))
    response.assert_is("PageRemoteError")
    assert response.id == remote_id
    assert response.archiveId == another.id
    assert (
        response.message
        == f"Page ID {remote_id} comes from remote archive ID {another.id}"
    )