diff options
Diffstat (limited to '')
| -rw-r--r-- | tests/api/test_archive.py | 388 |
1 files changed, 388 insertions, 0 deletions
diff --git a/tests/api/test_archive.py b/tests/api/test_archive.py new file mode 100644 index 0000000..0ef3425 --- /dev/null +++ b/tests/api/test_archive.py @@ -0,0 +1,388 @@ +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}" + ) |