From 41f3c80b4adf9f0d4facacbbb4750350f3eb1f5f Mon Sep 17 00:00:00 2001 From: benoit74 Date: Tue, 11 Nov 2025 15:15:30 +0000 Subject: [PATCH] Handle books location, and update them when adding book to title --- backend/src/cms_backend/db/book.py | 46 ++- backend/src/cms_backend/db/models.py | 20 + .../versions/add_book_location_table.py | 43 ++ backend/src/cms_backend/processors/book.py | 5 +- backend/src/cms_backend/processors/title.py | 51 ++- .../processors/zimfarm_notification.py | 52 ++- backend/src/cms_backend/routes/books.py | 34 +- backend/src/cms_backend/schemas/orms.py | 14 + backend/src/cms_backend/utils/filename.py | 151 +++++++ backend/tests/conftest.py | 39 +- backend/tests/processors/test_book.py | 18 +- .../test_book_location_integration.py | 387 ++++++++++++++++++ backend/tests/processors/test_title.py | 99 ++++- .../processors/test_zimfarm_notification.py | 97 ++++- backend/tests/utils/__init__.py | 0 backend/tests/utils/test_filename.py | 210 ++++++++++ frontend/src/types/book.ts | 10 + frontend/src/views/BookView.vue | 58 +++ 18 files changed, 1283 insertions(+), 51 deletions(-) create mode 100644 backend/src/cms_backend/migrations/versions/add_book_location_table.py create mode 100644 backend/src/cms_backend/utils/filename.py create mode 100644 backend/tests/processors/test_book_location_integration.py create mode 100644 backend/tests/utils/__init__.py create mode 100644 backend/tests/utils/test_filename.py diff --git a/backend/src/cms_backend/db/book.py b/backend/src/cms_backend/db/book.py index ec5c7fd..a1ab21f 100644 --- a/backend/src/cms_backend/db/book.py +++ b/backend/src/cms_backend/db/book.py @@ -3,7 +3,7 @@ from sqlalchemy.orm import Session as OrmSession -from cms_backend.db.models import Book, ZimfarmNotification +from cms_backend.db.models import Book, BookLocation, WarehousePath, ZimfarmNotification from cms_backend.utils.datetime import getnow @@ -53,3 +53,47 @@ def create_book( ) return book + + +def create_book_location( + session: OrmSession, + *, + book: Book, + warehouse_path_id: UUID, + filename: str, + status: str = "current", +) -> BookLocation: + """Create a new book location. + + Args: + session: SQLAlchemy session + book: Book instance + warehouse_path_id: ID of the warehouse path + filename: Filename in warehouse + status: Location status ('current' or 'target'), defaults to 'current' + + Returns: + Created BookLocation instance + """ + # Get warehouse path info for event message + warehouse_path = session.get(WarehousePath, warehouse_path_id) + if not warehouse_path: + raise ValueError(f"WarehousePath with id {warehouse_path_id} not found") + + warehouse_name = warehouse_path.warehouse.name + folder_name = warehouse_path.folder_name + + location = BookLocation( + book_id=book.id, + status=status, + filename=filename, + ) + location.warehouse_path_id = warehouse_path_id + session.add(location) + book.locations.append(location) + book.events.append( + f"{getnow()}: added {status} location: {filename} in {warehouse_name}: " + f"{folder_name} ({warehouse_path_id})" + ) + + return location diff --git a/backend/src/cms_backend/db/models.py b/backend/src/cms_backend/db/models.py index 82ee929..2566db9 100644 --- a/backend/src/cms_backend/db/models.py +++ b/backend/src/cms_backend/db/models.py @@ -112,6 +112,12 @@ class Book(Base): back_populates="book" ) + locations: Mapped[list["BookLocation"]] = relationship( + back_populates="book", + cascade="all, delete-orphan", + init=False, + ) + Index( "idx_book_status_pending_processing", @@ -205,3 +211,17 @@ class WarehousePath(Base): warehouse: Mapped["Warehouse"] = relationship( back_populates="warehouse_paths", init=False ) + + +class BookLocation(Base): + __tablename__ = "book_location" + book_id: Mapped[UUID] = mapped_column(ForeignKey("book.id"), primary_key=True) + warehouse_path_id: Mapped[UUID] = mapped_column( + ForeignKey("warehouse_path.id"), primary_key=True, init=False + ) + status: Mapped[str] = mapped_column(primary_key=True) # 'current' or 'target' + + filename: Mapped[str] + + book: Mapped["Book"] = relationship(back_populates="locations", init=False) + warehouse_path: Mapped["WarehousePath"] = relationship(init=False) diff --git a/backend/src/cms_backend/migrations/versions/add_book_location_table.py b/backend/src/cms_backend/migrations/versions/add_book_location_table.py new file mode 100644 index 0000000..f9cfe4d --- /dev/null +++ b/backend/src/cms_backend/migrations/versions/add_book_location_table.py @@ -0,0 +1,43 @@ +"""Add book_location table to store current and target file locations + +Revision ID: add_book_location_table +Revises: title_warehouse_paths +Create Date: 2025-11-11 00:00:00.000000 + +""" + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects.postgresql import UUID + +# revision identifiers, used by Alembic. +revision = "add_book_location_table" +down_revision = "title_warehouse_paths" +branch_labels = None +depends_on = None + + +def upgrade(): + # Create the book_location table with composite primary key + op.create_table( + "book_location", + sa.Column("book_id", UUID(), nullable=False), + sa.Column("warehouse_path_id", UUID(), nullable=False), + sa.Column("status", sa.String(), nullable=False), + sa.Column("filename", sa.String(), nullable=False), + sa.ForeignKeyConstraint( + ["book_id"], ["book.id"], name="fk_book_location_book_id_book" + ), + sa.ForeignKeyConstraint( + ["warehouse_path_id"], + ["warehouse_path.id"], + name="fk_book_location_warehouse_path_id_warehouse_path", + ), + sa.PrimaryKeyConstraint( + "book_id", "warehouse_path_id", "status", name="pk_book_location" + ), + ) + + +def downgrade(): + op.drop_table("book_location") diff --git a/backend/src/cms_backend/processors/book.py b/backend/src/cms_backend/processors/book.py index 3a74dcf..2cf1e86 100644 --- a/backend/src/cms_backend/processors/book.py +++ b/backend/src/cms_backend/processors/book.py @@ -46,8 +46,7 @@ def check_book_qa(book: Book) -> bool: def get_matching_title(session: ORMSession, book: Book) -> Title | None: try: - name = str(book.zim_metadata.get("Name")) - if not name: + if not book.name: book.events.append( f"{getnow()}: no title can be found because name is missing" ) @@ -55,7 +54,7 @@ def get_matching_title(session: ORMSession, book: Book) -> Title | None: return None title = get_title_by_name_and_producer_or_none( - session, name=name, producer_unique_id=book.producer_unique_id + session, name=book.name, producer_unique_id=book.producer_unique_id ) if not title: diff --git a/backend/src/cms_backend/processors/title.py b/backend/src/cms_backend/processors/title.py index 1e8142e..59f3a39 100644 --- a/backend/src/cms_backend/processors/title.py +++ b/backend/src/cms_backend/processors/title.py @@ -1,15 +1,24 @@ +from sqlalchemy import select +from sqlalchemy.orm import Session as OrmSession + from cms_backend import logger -from cms_backend.db.models import Book, Title +from cms_backend.db.book import create_book_location +from cms_backend.db.models import Book, Title, TitleWarehousePath from cms_backend.utils.datetime import getnow +from cms_backend.utils.filename import compute_target_filename -def add_book_to_title(book: Book, title: Title): +def add_book_to_title(session: OrmSession, book: Book, title: Title): try: - name = book.zim_metadata.get("Name") - if not name: - raise Exception("book Name is missing") - if not isinstance(name, str): - raise Exception(f"book Name is expected to be a string, found {type(name)}") + # Retrieve name from book.name directly + if not book.name: + raise Exception("book name is missing or invalid") + + # Validate book.date is also present and valid + if not book.date: + raise Exception("book date is missing or invalid") + + name = book.name title.books.append(book) book.events.append(f"{getnow()}: book added to title {title.id}") @@ -35,6 +44,34 @@ def add_book_to_title(book: Book, title: Title): ) title.producer_display_url = book.producer_display_url + # Compute target filename once for this book + target_filename = compute_target_filename( + session, + name=name, + flavour=book.flavour, + date=book.date, + ) + + # Determine which warehouse paths to use based on title.in_prod + path_type = "prod" if title.in_prod else "dev" + + # Get all warehouse paths for this title and path_type + stmt = select(TitleWarehousePath).where( + TitleWarehousePath.title_id == title.id, + TitleWarehousePath.path_type == path_type, + ) + target_warehouse_paths = session.scalars(stmt).all() + + # Create target locations for each applicable warehouse path + for title_warehouse_path in target_warehouse_paths: + create_book_location( + session=session, + book=book, + warehouse_path_id=title_warehouse_path.warehouse_path_id, + filename=target_filename, + status="target", + ) + except Exception as exc: book.events.append( f"{getnow()}: error encountered while adding to title {title.id}\n{exc}" diff --git a/backend/src/cms_backend/processors/zimfarm_notification.py b/backend/src/cms_backend/processors/zimfarm_notification.py index 7310fa5..12e40cd 100644 --- a/backend/src/cms_backend/processors/zimfarm_notification.py +++ b/backend/src/cms_backend/processors/zimfarm_notification.py @@ -1,10 +1,11 @@ from typing import cast +from sqlalchemy import select from sqlalchemy.orm import Session as ORMSession from cms_backend import logger -from cms_backend.db.book import create_book -from cms_backend.db.models import ZimfarmNotification +from cms_backend.db.book import create_book, create_book_location +from cms_backend.db.models import Warehouse, WarehousePath, ZimfarmNotification from cms_backend.processors.book import check_book_qa, get_matching_title from cms_backend.processors.title import add_book_to_title from cms_backend.utils.datetime import getnow @@ -27,7 +28,9 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification) "size", "metadata", "zimcheck", - "url", + "warehouse_name", + "folder_name", + "filename", "producer", ] if key not in notification.content @@ -67,6 +70,37 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification) notification.status = "bad_notification" return + # Look up warehouse path by warehouse_name and folder_name + warehouse_name = notification.content.get("warehouse_name") + folder_name = notification.content.get("folder_name") + filename = notification.content.get("filename") + + # Validate filename is a non-empty string + if not isinstance(filename, str) or not filename: + notification.events.append( + f"{getnow()}: filename must be a non-empty string, got " + f"{type(filename).__name__}: {filename}" + ) + notification.status = "bad_notification" + return + + stmt = ( + select(WarehousePath) + .join(Warehouse) + .where( + Warehouse.name == warehouse_name, + WarehousePath.folder_name == folder_name, + ) + ) + warehouse_path = session.scalars(stmt).one_or_none() + + if not warehouse_path: + notification.events.append( + f"{getnow()}: warehouse path not found: {warehouse_name}/{folder_name}" + ) + notification.status = "bad_notification" + return + book = create_book( session=session, book_id=notification.id, @@ -80,6 +114,16 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification) producer_display_url=producer["displayUrl"], producer_unique_id=producer["uniqueId"], ) + + # Create current book location + create_book_location( + session=session, + book=book, + warehouse_path_id=warehouse_path.id, + filename=filename, + status="current", + ) + notification.status = "processed" if not check_book_qa(book): @@ -90,7 +134,7 @@ def process_notification(session: ORMSession, notification: ZimfarmNotification) if not title: return - add_book_to_title(book, title) + add_book_to_title(session, book, title) except Exception as exc: notification.events.append( diff --git a/backend/src/cms_backend/routes/books.py b/backend/src/cms_backend/routes/books.py index 804b6ed..aca35e6 100644 --- a/backend/src/cms_backend/routes/books.py +++ b/backend/src/cms_backend/routes/books.py @@ -10,7 +10,12 @@ from cms_backend.routes.fields import LimitFieldMax200, NotEmptyString, SkipField from cms_backend.routes.models import ListResponse, calculate_pagination_metadata from cms_backend.schemas import BaseModel -from cms_backend.schemas.orms import BookFullSchema, BookLightSchema, ProducerSchema +from cms_backend.schemas.orms import ( + BookFullSchema, + BookLightSchema, + BookLocationSchema, + ProducerSchema, +) router = APIRouter(prefix="/books", tags=["books"]) @@ -59,6 +64,31 @@ async def get_book( db_book = db_get_book(session=session, book_id=book_id) + # Separate current and target locations + current_locations = [ + BookLocationSchema( + warehouse_path_id=location.warehouse_path_id, + warehouse_name=location.warehouse_path.warehouse.name, + folder_name=location.warehouse_path.folder_name, + filename=location.filename, + status=location.status, + ) + for location in db_book.locations + if location.status == "current" + ] + + target_locations = [ + BookLocationSchema( + warehouse_path_id=location.warehouse_path_id, + warehouse_name=location.warehouse_path.warehouse.name, + folder_name=location.warehouse_path.folder_name, + filename=location.filename, + status=location.status, + ) + for location in db_book.locations + if location.status == "target" + ] + return BookFullSchema( id=db_book.id, title_id=db_book.title_id, @@ -78,4 +108,6 @@ async def get_book( display_url=db_book.producer_display_url, unique_id=db_book.producer_unique_id, ), + current_locations=current_locations, + target_locations=target_locations, ) diff --git a/backend/src/cms_backend/schemas/orms.py b/backend/src/cms_backend/schemas/orms.py index ab33521..568e3c7 100644 --- a/backend/src/cms_backend/schemas/orms.py +++ b/backend/src/cms_backend/schemas/orms.py @@ -72,6 +72,18 @@ class ProducerSchema(BaseModel): unique_id: str +class BookLocationSchema(BaseModel): + """ + Schema for book location information + """ + + warehouse_path_id: UUID + warehouse_name: str + folder_name: str + filename: str + status: str # 'current' or 'target' + + class BookLightSchema(BaseModel): """ Schema for reading a book model with some fields @@ -94,6 +106,8 @@ class BookFullSchema(BookLightSchema): zim_metadata: dict[str, Any] events: list[str] producer: ProducerSchema + current_locations: list[BookLocationSchema] + target_locations: list[BookLocationSchema] class WarehousePathSchema(BaseModel): diff --git a/backend/src/cms_backend/utils/filename.py b/backend/src/cms_backend/utils/filename.py new file mode 100644 index 0000000..869e984 --- /dev/null +++ b/backend/src/cms_backend/utils/filename.py @@ -0,0 +1,151 @@ +"""Utilities for computing and managing book target filenames.""" + +from sqlalchemy import select +from sqlalchemy.orm import Session as OrmSession + +from cms_backend.db.models import BookLocation + +PERIOD_LENGTH = 7 + + +def get_next_suffix(current_suffix: str) -> str: + """ + Get the next suffix in sequence. + + Suffixes progress: "" -> "a" -> "b" -> ... -> "z" -> "aa" -> "ab" -> ... + This is essentially a base-26 number system using lowercase letters. + + Args: + current_suffix: The current suffix (e.g., "a", "z", "aa", "ab") + + Returns: + The next suffix in sequence + + Examples: + - "" -> "a" + - "a" -> "b" + - "z" -> "aa" + - "ab" -> "ac" + - "az" -> "ba" + - "zz" -> "aaa" + """ + if not current_suffix: + return "a" + + # Convert suffix to list of chars for easier manipulation + chars = list(current_suffix) + + # Increment from right to left (like a base-26 number) + carry = True + for i in range(len(chars) - 1, -1, -1): + if carry: + if chars[i] == "z": + chars[i] = "a" + # carry continues + else: + chars[i] = chr(ord(chars[i]) + 1) + carry = False + break + + # If we still have carry, we need an additional character + if carry: + chars.insert(0, "a") + + return "".join(chars) + + +def compute_target_filename( + session: OrmSession, *, name: str, flavour: str | None, date: str +) -> str: + """ + Compute target filename: {name}[_{flavour}]_{period}[suffix] + + Period is YYYY-MM from book.date (format: YYYY-MM-DD), with suffix for multiple + books in the same month: + - YYYY-MM (no suffix, if available) + - YYYY-MMa, YYYY-MMb, ..., YYYY-MMz (single letter) + - YYYY-MMaa, YYYY-MMab, ... (multiple letters) + + Finds the last suffix already in use and generates the next one. + Queries ALL book locations (any status) with filenames starting with base pattern. + + Important edge cases: + - Books with same name but different flavours (including no flavour) never collide + - If YYYY-MM and YYYY-MMa exist but YYYY-MMb was deleted, next is YYYY-MMc + (finds last suffix, not collision) + + Args: + session: SQLAlchemy session + name: Book name + flavour: Book flavour (optional) + date: Book date (format: YYYY-MM-DD) + + Returns: + Target filename including .zim extension + + Raises: + ValueError: If date is missing or has invalid format + """ + if not date: + raise ValueError("Book date is required to compute target filename") + + # Extract YYYY-MM from book.date (format: YYYY-MM-DD) + if len(date) < PERIOD_LENGTH: + raise ValueError(f"Book date must have at least YYYY-MM format, got: {date}") + + base_period = date[:PERIOD_LENGTH] # "2024-01-15" -> "2024-01" + + # Build base filename pattern + # IMPORTANT: name with no flavour vs name with flavour never collide + if flavour: + base_name = f"{name}_{flavour}" + else: + base_name = name + + # Base pattern for this book (without .zim extension) + base_pattern = f"{base_name}_{base_period}" + + # Query all locations where filename starts with this pattern + # Check ALL locations regardless of status (current or target) + stmt = select(BookLocation.filename).where( + BookLocation.filename.like(f"{base_pattern}%") + ) + existing_filenames = list(session.scalars(stmt).all()) + + # If no existing files, use base pattern (no suffix) + if not existing_filenames: + return f"{base_pattern}.zim" + + # Parse existing filenames to find the highest suffix + # Examples: + # - "foo_2024-01.zim" -> no suffix (empty string) + # - "foo_2024-01a.zim" -> "a" + # - "foo_2024-01ab.zim" -> "ab" + + suffixes: list[str] = [] + for filename in existing_filenames: + # Remove .zim extension + name_part = filename.rsplit(".zim", 1)[0] + + # Extract suffix after base_pattern + if name_part == base_pattern: + # No suffix + suffixes.append("") + elif name_part.startswith(base_pattern): + suffix = name_part[len(base_pattern) :] + # Validate it's only lowercase letters + if suffix.isalpha() and suffix.islower(): + suffixes.append(suffix) + + # If no valid suffixes found, use base pattern + if not suffixes: + return f"{base_pattern}.zim" + + # Sort suffixes: "", "a", "b", ..., "z", "aa", "ab", ..., "zz", "aaa", ... + suffixes.sort(key=lambda s: (len(s), s)) + + # Get the last suffix and compute the next one + last_suffix = suffixes[-1] + next_suffix = get_next_suffix(last_suffix) + + return f"{base_pattern}{next_suffix}.zim" diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 5f7f0c4..b06da76 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -12,6 +12,7 @@ from cms_backend.db.models import ( Base, Book, + BookLocation, Title, TitleWarehousePath, Warehouse, @@ -148,12 +149,14 @@ def create_title( create_warehouse_path: Callable[..., WarehousePath], ) -> Callable[..., Title]: def _create_title( + *, name: str = "test_en_all", producer_unique_id: str | None = None, producer_display_name: str | None = None, producer_display_url: str | None = None, dev_warehouse_path_ids: list[UUID] | None = None, prod_warehouse_path_ids: list[UUID] | None = None, + in_prod: bool = False, ) -> Title: title = Title( name=name, @@ -165,8 +168,10 @@ def _create_title( ) title.producer_display_name = producer_display_name title.producer_display_url = producer_display_url + title.in_prod = in_prod - # Create default warehouse paths if not provided + # Create default warehouse paths if not provided (None means create default) + # Empty list means explicitly no paths if dev_warehouse_path_ids is None: dev_warehouse_path = create_warehouse_path() dev_warehouse_path_ids = [dev_warehouse_path.id] @@ -254,3 +259,35 @@ def warehouse_path( warehouse: Warehouse, ) -> WarehousePath: return create_warehouse_path(warehouse=warehouse) + + +@pytest.fixture +def create_book_location( + dbsession: OrmSession, + create_book: Callable[..., Book], + create_warehouse_path: Callable[..., WarehousePath], +) -> Callable[..., BookLocation]: + def _create_book_location( + book: Book | None = None, + warehouse_path: WarehousePath | None = None, + filename: str | None = None, + status: str = "current", + ) -> BookLocation: + if book is None: + book = create_book() + if warehouse_path is None: + warehouse_path = create_warehouse_path() + if filename is None: + filename = "test_file.zim" + + location = BookLocation( + book_id=book.id, + status=status, + filename=filename, + ) + location.warehouse_path_id = warehouse_path.id + dbsession.add(location) + dbsession.flush() + return location + + return _create_book_location diff --git a/backend/tests/processors/test_book.py b/backend/tests/processors/test_book.py index 99fd178..791d7fc 100644 --- a/backend/tests/processors/test_book.py +++ b/backend/tests/processors/test_book.py @@ -114,9 +114,7 @@ def test_get_matching_title_found( ): """Get matching title for a given book - title exist""" - book = create_book( - zim_metadata={"Name": title.name}, producer_unique_id=title.producer_unique_id - ) + book = create_book(name=title.name, producer_unique_id=title.producer_unique_id) assert len(book.events) == 0 assert len(title.events) == 0 matching_title = get_matching_title(dbsession, book=book) @@ -134,7 +132,7 @@ def test_get_matching_title_not_found( book_name = "test2_fr_all" assert book_name != title.name - book = create_book(zim_metadata={"Name": book_name}) + book = create_book(name=book_name) assert len(book.events) == 0 assert len(title.events) == 0 matching_title = get_matching_title(dbsession, book=book) @@ -154,7 +152,7 @@ def test_get_matching_title_no_name( book_name = "" assert book_name != title.name - book = create_book(zim_metadata={"Name": book_name}) + book = create_book(name=book_name) assert len(book.events) == 0 assert len(title.events) == 0 matching_title = get_matching_title(dbsession, book=book) @@ -172,17 +170,15 @@ def test_get_matching_title_bad_error( ): """Get matching title for a given book - bad error occurs""" - book_name = "" - assert book_name != title.name - book = create_book(zim_metadata={"Name": book_name}) + book = create_book(name=title.name) assert len(book.events) == 0 assert len(title.events) == 0 # simulate a very bad error by dropping an expected property (and adding it back so # that SQLAlchemy does not choke) - save_metadata = book.zim_metadata - del book.zim_metadata + save_producer_unique_id = book.producer_unique_id + del book.producer_unique_id matching_title = get_matching_title(dbsession, book=book) - book.zim_metadata = save_metadata + book.producer_unique_id = save_producer_unique_id assert matching_title is None assert any( event diff --git a/backend/tests/processors/test_book_location_integration.py b/backend/tests/processors/test_book_location_integration.py new file mode 100644 index 0000000..0f35a09 --- /dev/null +++ b/backend/tests/processors/test_book_location_integration.py @@ -0,0 +1,387 @@ +"""Integration tests for book location workflow.""" + +from collections.abc import Callable +from typing import Any + +import pytest +from sqlalchemy.orm import Session as OrmSession + +from cms_backend.db.models import ( + Book, + BookLocation, + Title, + Warehouse, + WarehousePath, + ZimfarmNotification, +) +from cms_backend.processors.zimfarm_notification import process_notification + +GOOD_ZIM_METADATA = { + "Name": "test_en_all", + "Title": "Test Title", + "Creator": "Test", + "Publisher": "Test", + "Date": "2024-01-15", + "Description": "Test Description", + "Language": "eng", + "Flavour": None, +} + +GOOD_PRODUCER = { + "displayName": "farm.openzim.org: test_en_all", + "displayUrl": "https://farm.openzim.org/recipes/test_en_all", + "uniqueId": "550e8400-e29b-41d4-a716-446655440000", +} + + +@pytest.fixture +def warehouse_setup( + dbsession: OrmSession, # noqa: ARG001 + create_warehouse: Callable[..., Warehouse], + create_warehouse_path: Callable[..., WarehousePath], +) -> dict[str, Any]: + """Set up warehouse and warehouse paths for testing.""" + dev_warehouse = create_warehouse(name="dev-warehouse") + prod_warehouse = create_warehouse(name="prod-warehouse") + + dev_path = create_warehouse_path( + warehouse=dev_warehouse, + folder_name="dev-zim", + ) + prod_path = create_warehouse_path( + warehouse=prod_warehouse, + folder_name="prod-zim", + ) + + return { + "dev_warehouse": dev_warehouse, + "prod_warehouse": prod_warehouse, + "dev_path": dev_path, + "prod_path": prod_path, + } + + +@pytest.fixture +def good_notification_content( + warehouse_setup: dict[str, Any], # noqa: ARG001 +) -> dict[str, Any]: + """Create good notification content with warehouse info.""" + return { + "article_count": 100, + "media_count": 50, + "size": 1024000, + "metadata": GOOD_ZIM_METADATA, + "zimcheck": {"status": "pass"}, + "warehouse_name": "dev-warehouse", + "folder_name": "dev-zim", + "filename": "test_en_all_2024-01-15.zim", + "producer": GOOD_PRODUCER, + } + + +class TestBookLocationCreation: + """Test book location creation during notification processing.""" + + def test_book_gets_current_location_from_notification( + self, + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + good_notification_content: dict[str, Any], + ): + """Book created from notification should have current location.""" + notification = create_zimfarm_notification(content=good_notification_content) + assert notification.status == "pending" + + process_notification(dbsession, notification) + + dbsession.flush() + assert notification.status == "processed" + assert notification.book is not None + + book = notification.book + current_locations = [loc for loc in book.locations if loc.status == "current"] + + assert len(current_locations) == 1 + assert current_locations[0].filename == "test_en_all_2024-01-15.zim" + assert current_locations[0].warehouse_path.warehouse.name == "dev-warehouse" + assert current_locations[0].warehouse_path.folder_name == "dev-zim" + + def test_notification_without_warehouse_path_fails( + self, + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + ): + """Notification with non-existent warehouse path should fail.""" + bad_content = { + "article_count": 100, + "media_count": 50, + "size": 1024000, + "metadata": GOOD_ZIM_METADATA, + "zimcheck": {"status": "pass"}, + "warehouse_name": "non-existent-warehouse", + "folder_name": "non-existent-folder", + "filename": "test.zim", + "producer": GOOD_PRODUCER, + } + notification = create_zimfarm_notification(content=bad_content) + + process_notification(dbsession, notification) + + dbsession.flush() + assert notification.status == "bad_notification" + assert any("warehouse path not found" in event for event in notification.events) + + +class TestTargetLocationCreation: + """Test target location creation when book is added to title.""" + + def test_target_locations_created_for_dev_title( + self, + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + create_title: Callable[..., Title], + warehouse_setup: dict[str, Any], + good_notification_content: dict[str, Any], + ): + """When book is added to dev title, target locations should use dev paths.""" + dev_path = warehouse_setup["dev_path"] + + # Create dev title + create_title( + name="test_en_all", + producer_unique_id=GOOD_PRODUCER["uniqueId"], + dev_warehouse_path_ids=[dev_path.id], + prod_warehouse_path_ids=[], + in_prod=False, + ) + + notification = create_zimfarm_notification(content=good_notification_content) + process_notification(dbsession, notification) + + dbsession.flush() + assert notification.book is not None + + book = notification.book + target_locations = [loc for loc in book.locations if loc.status == "target"] + + assert len(target_locations) == 1 + assert target_locations[0].warehouse_path_id == dev_path.id + # Target filename should be computed + assert "2024-01" in target_locations[0].filename + assert target_locations[0].filename.endswith(".zim") + + def test_target_locations_created_for_prod_title( + self, + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + create_title: Callable[..., Title], + warehouse_setup: dict[str, Any], + good_notification_content: dict[str, Any], + ): + """When book is added to prod title, target locations should use prod paths.""" + prod_path = warehouse_setup["prod_path"] + + # Create prod title + create_title( + name="test_en_all", + producer_unique_id=GOOD_PRODUCER["uniqueId"], + dev_warehouse_path_ids=[], + prod_warehouse_path_ids=[prod_path.id], + in_prod=True, + ) + + notification = create_zimfarm_notification(content=good_notification_content) + process_notification(dbsession, notification) + + dbsession.flush() + assert notification.book is not None + + book = notification.book + target_locations = [loc for loc in book.locations if loc.status == "target"] + + assert len(target_locations) == 1 + assert target_locations[0].warehouse_path_id == prod_path.id + + def test_multiple_target_locations_for_multiple_paths( + self, + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + create_title: Callable[..., Title], + warehouse_setup: dict[str, Any], + create_warehouse_path: Callable[..., WarehousePath], + good_notification_content: dict[str, Any], + ): + """Title with multiple warehouse paths + + This should create multiple target locations. + """ + dev_path_1 = warehouse_setup["dev_path"] + dev_path_2 = create_warehouse_path( + warehouse=warehouse_setup["dev_warehouse"], + folder_name="dev-zim-backup", + ) + + # Create title with multiple dev paths + create_title( + name="test_en_all", + producer_unique_id=GOOD_PRODUCER["uniqueId"], + dev_warehouse_path_ids=[dev_path_1.id, dev_path_2.id], + prod_warehouse_path_ids=[], + in_prod=False, + ) + + notification = create_zimfarm_notification(content=good_notification_content) + process_notification(dbsession, notification) + + dbsession.flush() + assert notification.book is not None + + book = notification.book + target_locations = [loc for loc in book.locations if loc.status == "target"] + + assert len(target_locations) == 2 + target_warehouse_path_ids = {loc.warehouse_path_id for loc in target_locations} + assert target_warehouse_path_ids == {dev_path_1.id, dev_path_2.id} + + +class TestTargetFilenameComputation: + """Test that target filenames are computed correctly.""" + + def test_target_filename_basic_format( + self, + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + create_title: Callable[..., Title], + warehouse_setup: dict[str, Any], + good_notification_content: dict[str, Any], + ): + """Target filename should follow pattern {name}_{period}.zim.""" + dev_path = warehouse_setup["dev_path"] + + create_title( + name="test_en_all", + producer_unique_id=GOOD_PRODUCER["uniqueId"], + dev_warehouse_path_ids=[dev_path.id], + prod_warehouse_path_ids=[], + in_prod=False, + ) + + notification = create_zimfarm_notification(content=good_notification_content) + process_notification(dbsession, notification) + + dbsession.flush() + + assert notification.book + + target_locations = [ + loc for loc in notification.book.locations if loc.status == "target" + ] + + assert len(target_locations) == 1 + # Should match pattern: test_en_all_2024-01.zim + assert target_locations[0].filename == "test_en_all_2024-01.zim" + + def test_target_filename_with_flavour( + self, + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + create_title: Callable[..., Title], + warehouse_setup: dict[str, Any], + ): + """Target filename should include flavour if present.""" + dev_path = warehouse_setup["dev_path"] + + metadata_with_flavour = { + "Name": "wikipedia_en_all", + "Title": "Wikipedia EN All", + "Creator": "Test", + "Publisher": "Test", + "Date": "2024-02-20", + "Description": "Test Description", + "Language": "eng", + "Flavour": "maxi", + } + + content = { + "article_count": 100, + "media_count": 50, + "size": 1024000, + "metadata": metadata_with_flavour, + "zimcheck": {"status": "pass"}, + "warehouse_name": "dev-warehouse", + "folder_name": "dev-zim", + "filename": "wikipedia_en_all_maxi_2024-02-20.zim", + "producer": GOOD_PRODUCER, + } + + create_title( + name="wikipedia_en_all", + producer_unique_id=GOOD_PRODUCER["uniqueId"], + dev_warehouse_path_ids=[dev_path.id], + prod_warehouse_path_ids=[], + in_prod=False, + ) + + notification = create_zimfarm_notification(content=content) + process_notification(dbsession, notification) + + dbsession.flush() + + assert notification.book + + target_locations = [ + loc for loc in notification.book.locations if loc.status == "target" + ] + + assert len(target_locations) == 1 + assert target_locations[0].filename == "wikipedia_en_all_maxi_2024-02.zim" + + def test_target_filename_collision_handling( + self, + dbsession: OrmSession, + create_book: Callable[..., Book], + create_book_location: Callable[..., BookLocation], + create_zimfarm_notification: Callable[..., ZimfarmNotification], + create_title: Callable[..., Title], + warehouse_setup: dict[str, Any], + good_notification_content: dict[str, Any], + ): + """Target filename should get suffix if collision exists.""" + dev_path = warehouse_setup["dev_path"] + + create_title( + name="test_en_all", + producer_unique_id=GOOD_PRODUCER["uniqueId"], + dev_warehouse_path_ids=[dev_path.id], + prod_warehouse_path_ids=[], + in_prod=False, + ) + + # Create existing book with target location to cause collision + existing_book = create_book( + name="test_en_all", + date="2024-01-10", + flavour=None, + ) + create_book_location( + book=existing_book, + warehouse_path=dev_path, + filename="test_en_all_2024-01.zim", + status="target", + ) + + # Process new notification for same period + notification = create_zimfarm_notification(content=good_notification_content) + process_notification(dbsession, notification) + + dbsession.flush() + + assert notification.book + + target_locations = [ + loc for loc in notification.book.locations if loc.status == "target" + ] + + assert len(target_locations) == 1 + # Should get letter suffix to avoid collision + assert target_locations[0].filename == "test_en_all_2024-01a.zim" diff --git a/backend/tests/processors/test_title.py b/backend/tests/processors/test_title.py index 90c569e..d03ecbf 100644 --- a/backend/tests/processors/test_title.py +++ b/backend/tests/processors/test_title.py @@ -12,10 +12,10 @@ def test_add_book_to_title_same_name( ): """Add a book to an existing title with same name""" - book = create_book(zim_metadata={"Name": title.name}) + book = create_book(name=title.name, date="2024-01-01") assert len(book.events) == 0 assert len(title.events) == 0 - add_book_to_title(book=book, title=title) + add_book_to_title(session=dbsession, book=book, title=title) dbsession.flush() assert book.title == title assert book.title_id == title.id @@ -35,10 +35,10 @@ def test_add_book_to_title_different_name( book_name = "test2_fr_all" assert book_name != title.name - book = create_book(zim_metadata={"Name": book_name}) + book = create_book(name=book_name, date="2024-01-01") assert len(book.events) == 0 assert len(title.events) == 0 - add_book_to_title(book=book, title=title) + add_book_to_title(session=dbsession, book=book, title=title) dbsession.flush() assert book.title == title assert book.title_id == title.id @@ -64,10 +64,10 @@ def test_add_book_to_title_empty_name( book_name = "" assert book_name != title.name - book = create_book(zim_metadata={"Name": book_name}) + book = create_book(name=book_name, date="2024-01-01") assert len(book.events) == 0 assert len(title.events) == 0 - add_book_to_title(book=book, title=title) + add_book_to_title(session=dbsession, book=book, title=title) dbsession.flush() assert book not in title.books assert book.title is None @@ -90,17 +90,15 @@ def test_add_book_to_title_empty_name( ) -def test_add_book_to_title_bad_name( +def test_add_book_to_title_missing_name( dbsession: OrmSession, create_book: Callable[..., Book], title: Title ): - """Add a book to an existing title with a bad type name""" + """Add a book to an existing title with missing name""" - book_name = 123 # integer instead of string - assert book_name != title.name - book = create_book(zim_metadata={"Name": book_name}) + book = create_book(name=None, date="2024-01-01") assert len(book.events) == 0 assert len(title.events) == 0 - add_book_to_title(book=book, title=title) + add_book_to_title(session=dbsession, book=book, title=title) dbsession.flush() assert book not in title.books assert book.title is None @@ -128,16 +126,16 @@ def test_add_book_to_title_bad_error( ): """Add a book to an existing title which encounters a bad error""" - book = create_book(zim_metadata={"Name": title.name}) + book = create_book(name=title.name, date="2024-01-01") assert len(book.events) == 0 assert len(title.events) == 0 # simulate a very bad error by dropping an expected property (and adding it back so # that SQLAlchemy does not choke) - save_metadata = book.zim_metadata - del book.zim_metadata - add_book_to_title(book=book, title=title) - book.zim_metadata = save_metadata + save_name = book.name + book.name = None + add_book_to_title(session=dbsession, book=book, title=title) + book.name = save_name dbsession.flush() assert book not in title.books @@ -178,7 +176,8 @@ def test_add_book_to_title_updates_producer_fields( # Create a book with complete producer information book = create_book( - zim_metadata={"Name": title.name}, + name=title.name, + date="2024-01-01", producer_unique_id=title.producer_unique_id, producer_display_name="farm.openzim.org: test_en_all", producer_display_url="https://farm.openzim.org/recipes/test_en_all", @@ -189,7 +188,7 @@ def test_add_book_to_title_updates_producer_fields( assert len(book.events) == 0 assert len(title.events) == 0 - add_book_to_title(book=book, title=title) + add_book_to_title(session=dbsession, book=book, title=title) dbsession.flush() # Verify producer fields were updated @@ -205,3 +204,65 @@ def test_add_book_to_title_updates_producer_fields( for event in title.events if re.match(".*: updating title producer_display_url to .*", event) ) + + +def test_add_book_to_title_missing_date( + dbsession: OrmSession, create_book: Callable[..., Book], title: Title +): + """Add a book to an existing title with missing date""" + + book = create_book(name=title.name, date=None) + assert len(book.events) == 0 + assert len(title.events) == 0 + add_book_to_title(session=dbsession, book=book, title=title) + dbsession.flush() + assert book not in title.books + assert book.title is None + assert book.title_id is None + assert [ + event for event in title.events if re.match(".*: book .* added to title", event) + ] == [] + assert [ + event for event in book.events if re.match(".*: book added to title .*", event) + ] == [] + assert any( + event + for event in title.events + if re.match(".*: error encountered while adding book .*", event) + ) + assert any( + event + for event in book.events + if re.match(".*: error encountered while adding to title .*", event) + ) + + +def test_add_book_to_title_empty_date( + dbsession: OrmSession, create_book: Callable[..., Book], title: Title +): + """Add a book to an existing title with empty date""" + + book = create_book(name=title.name, date="") + assert len(book.events) == 0 + assert len(title.events) == 0 + add_book_to_title(session=dbsession, book=book, title=title) + dbsession.flush() + assert book not in title.books + assert book.title is None + assert book.title_id is None + assert [ + event for event in title.events if re.match(".*: book .* added to title", event) + ] == [] + assert [ + event for event in book.events if re.match(".*: book added to title .*", event) + ] == [] + assert any( + event + for event in title.events + if re.match(".*: error encountered while adding book .*", event) + ) + assert any( + event + for event in book.events + if re.match(".*: error encountered while adding to title .*", event) + ) diff --git a/backend/tests/processors/test_zimfarm_notification.py b/backend/tests/processors/test_zimfarm_notification.py index cc5d05e..95bc919 100644 --- a/backend/tests/processors/test_zimfarm_notification.py +++ b/backend/tests/processors/test_zimfarm_notification.py @@ -21,7 +21,9 @@ "size": 1024000, "metadata": GOOD_ZIM_METADATA, "zimcheck": {"status": "pass"}, - "url": "https://example.com/zim/test.zim", + "warehouse_name": "test_warehouse", + "folder_name": "test_folder", + "filename": "test.zim", "producer": GOOD_PRODUCER, } @@ -30,9 +32,15 @@ def test_process_notification_success( dbsession: OrmSession, create_zimfarm_notification: Callable[..., ZimfarmNotification], create_title: Callable[..., Title], + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], ): """Process notification successfully - all steps work""" + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + # Create title with matching producer_unique_id title = create_title( name="test_en_all", producer_unique_id=GOOD_PRODUCER["uniqueId"] @@ -99,7 +107,9 @@ def test_process_notification_success( "size", "metadata", "zimcheck", - "url", + "warehouse_name", + "folder_name", + "filename", "producer", ] ], @@ -137,7 +147,7 @@ def test_process_notification_missing_multiple_mandatory_keys( notification_content = { key: value for key, value in GOOD_NOTIFICATION_CONTENT.items() - if key not in ["article_count", "size", "url"] + if key not in ["article_count", "size", "filename"] } notification = create_zimfarm_notification(content=notification_content) assert len(notification.events) == 0 @@ -153,7 +163,7 @@ def test_process_notification_missing_multiple_mandatory_keys( event for event in notification.events if re.match( - ".*: notification is missing mandatory keys: article_count,size,url", + ".*: notification is missing mandatory keys: article_count,size,filename", event, ) ) @@ -163,9 +173,15 @@ def test_process_notification_qa_check_fails( dbsession: OrmSession, create_zimfarm_notification: Callable[..., ZimfarmNotification], title: Title, # noqa: ARG001 + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], ): """Process notification where book QA check fails due to missing metadata""" + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + # Create notification with metadata missing the Creator field incomplete_metadata = { key: value for key, value in GOOD_ZIM_METADATA.items() if key != "Creator" @@ -206,9 +222,15 @@ def test_process_notification_no_matching_title( dbsession: OrmSession, create_zimfarm_notification: Callable[..., ZimfarmNotification], title: Title, # noqa: ARG001 + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], ): """Process notification where no matching title exists in database""" + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + # Use a different name that doesn't match the existing title different_metadata = {**GOOD_ZIM_METADATA, "Name": "different_title_name"} notification_content = { @@ -252,9 +274,15 @@ def test_process_notification_missing_name( dbsession: OrmSession, create_zimfarm_notification: Callable[..., ZimfarmNotification], title: Title, # noqa: ARG001 + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], ): """Process notification where book metadata has no Name field""" + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + # Remove Name from metadata no_name_metadata = { key: value for key, value in GOOD_ZIM_METADATA.items() if key != "Name" @@ -319,9 +347,15 @@ def test_process_notification_empty_name( dbsession: OrmSession, create_zimfarm_notification: Callable[..., ZimfarmNotification], title: Title, # noqa: ARG001 + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], ): """Process notification where book metadata has empty Name field""" + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + # Set Name to empty string empty_name_metadata = {**GOOD_ZIM_METADATA, "Name": ""} notification_content = { @@ -355,9 +389,15 @@ def test_process_notification_with_existing_books( create_zimfarm_notification: Callable[..., ZimfarmNotification], create_book: Callable[..., Book], create_title: Callable[..., Title], + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], ): """Process notification and add to title that already has books""" + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + # Create title with matching producer_unique_id title = create_title( name="test_en_all", producer_unique_id=GOOD_PRODUCER["uniqueId"] @@ -489,9 +529,15 @@ def test_process_notification_producer_stored_in_book( dbsession: OrmSession, create_zimfarm_notification: Callable[..., ZimfarmNotification], title: Title, # noqa: ARG001 + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], ): """Process notification successfully and verify producer fields are stored""" + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + notification = create_zimfarm_notification(content=GOOD_NOTIFICATION_CONTENT) assert len(notification.events) == 0 assert notification.status == "pending" @@ -506,3 +552,46 @@ def test_process_notification_producer_stored_in_book( assert notification.book.producer_display_name == GOOD_PRODUCER["displayName"] assert notification.book.producer_display_url == GOOD_PRODUCER["displayUrl"] assert notification.book.producer_unique_id == GOOD_PRODUCER["uniqueId"] + + +@pytest.mark.parametrize( + "invalid_filename", + [ + pytest.param(123, id="int"), + pytest.param(None, id="None"), + pytest.param(["file.zim"], id="list"), + pytest.param({"name": "file.zim"}, id="dict"), + pytest.param("", id="empty-string"), + ], +) +def test_process_notification_filename_not_valid_string( + dbsession: OrmSession, + create_zimfarm_notification: Callable[..., ZimfarmNotification], + invalid_filename: Any, + create_warehouse: Callable[..., Any], + create_warehouse_path: Callable[..., Any], +): + """Process notification where filename is not a valid non-empty string""" + + # Create warehouse and warehouse path that match the notification + warehouse = create_warehouse(name="test_warehouse") + create_warehouse_path(folder_name="test_folder", warehouse=warehouse) + + notification_content = { + **GOOD_NOTIFICATION_CONTENT, + "filename": invalid_filename, + } + notification = create_zimfarm_notification(content=notification_content) + assert len(notification.events) == 0 + assert notification.status == "pending" + + process_notification(dbsession, notification) + + dbsession.flush() + assert notification.status == "bad_notification" + assert notification.book is None + assert any( + event + for event in notification.events + if re.match(r".*: filename must be a non-empty string, got \w+", event) + ) diff --git a/backend/tests/utils/__init__.py b/backend/tests/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/utils/test_filename.py b/backend/tests/utils/test_filename.py new file mode 100644 index 0000000..ed8db6b --- /dev/null +++ b/backend/tests/utils/test_filename.py @@ -0,0 +1,210 @@ +"""Tests for filename utilities.""" + +from collections.abc import Callable + +import pytest +from sqlalchemy.orm import Session as OrmSession + +from cms_backend.db.models import BookLocation +from cms_backend.utils.filename import compute_target_filename, get_next_suffix + + +class TestGetNextSuffix: + """Test the get_next_suffix function.""" + + def test_empty_suffix_returns_a(self): + """Empty suffix should return 'a'.""" + assert get_next_suffix("") == "a" + + def test_single_letter_progression(self): + """Single letters should progress through alphabet.""" + assert get_next_suffix("a") == "b" + assert get_next_suffix("b") == "c" + assert get_next_suffix("m") == "n" + assert get_next_suffix("y") == "z" + + def test_z_wraps_to_aa(self): + """'z' should wrap to 'aa'.""" + assert get_next_suffix("z") == "aa" + + def test_double_letter_progression(self): + """Double letters should progress correctly.""" + assert get_next_suffix("aa") == "ab" + assert get_next_suffix("ab") == "ac" + assert get_next_suffix("az") == "ba" + assert get_next_suffix("ba") == "bb" + assert get_next_suffix("zz") == "aaa" + + def test_triple_letter_progression(self): + """Triple letters should progress correctly.""" + assert get_next_suffix("aaa") == "aab" + assert get_next_suffix("aaz") == "aba" + assert get_next_suffix("zzz") == "aaaa" + + +class TestComputeTargetFilename: + """Test the compute_target_filename function.""" + + def test_missing_date_raises_error(self, dbsession: OrmSession): + """Missing date should raise ValueError.""" + with pytest.raises(ValueError, match="Book date is required"): + compute_target_filename(dbsession, name="test", flavour=None, date="") + + def test_invalid_date_format_raises_error(self, dbsession: OrmSession): + """Invalid date format should raise ValueError.""" + with pytest.raises(ValueError, match="at least YYYY-MM"): + compute_target_filename(dbsession, name="test", flavour=None, date="2024") + + def test_basic_filename_without_flavour(self, dbsession: OrmSession): + """Should generate basic filename without flavour.""" + result = compute_target_filename( + dbsession, + name="wikipedia_en_all", + flavour=None, + date="2024-01-15", + ) + assert result == "wikipedia_en_all_2024-01.zim" + + def test_basic_filename_with_flavour(self, dbsession: OrmSession): + """Should generate filename with flavour.""" + result = compute_target_filename( + dbsession, + name="wikipedia_en_all", + flavour="maxi", + date="2024-01-15", + ) + assert result == "wikipedia_en_all_maxi_2024-01.zim" + + def test_no_existing_locations_uses_base(self, dbsession: OrmSession): + """When no existing locations, should use base pattern.""" + # Fresh database has no locations + result = compute_target_filename( + dbsession, + name="new_book", + flavour=None, + date="2024-02-20", + ) + assert result == "new_book_2024-02.zim" + + def test_collision_handling_single_letter( + self, dbsession: OrmSession, create_book_location: Callable[..., BookLocation] + ): + """Should handle collision with single letter suffix.""" + # Create first location (base pattern) + create_book_location( + filename="test_book_2024-03.zim", + status="target", + ) + + # Compute filename for same name/flavour/period + result = compute_target_filename( + dbsession, + name="test_book", + flavour=None, + date="2024-03-10", + ) + assert result == "test_book_2024-03a.zim" + + def test_multiple_collisions_progression( + self, dbsession: OrmSession, create_book_location: Callable[..., BookLocation] + ): + """Should handle multiple collisions in progression.""" + # Create locations with existing suffixes + create_book_location(filename="foo_2024-04.zim", status="target") + create_book_location(filename="foo_2024-04a.zim", status="target") + create_book_location(filename="foo_2024-04b.zim", status="target") + + # Should get next suffix + result = compute_target_filename( + dbsession, + name="foo", + flavour=None, + date="2024-04-10", + ) + assert result == "foo_2024-04c.zim" + + def test_gap_in_suffixes_uses_last( + self, dbsession: OrmSession, create_book_location: Callable[..., BookLocation] + ): + """Should use last suffix even if gaps exist.""" + # Create locations with a gap (a, c exist, b is missing) + create_book_location(filename="bar_2024-05.zim", status="target") + create_book_location(filename="bar_2024-05a.zim", status="target") + # Note: bar_2024-05b.zim is missing + create_book_location(filename="bar_2024-05c.zim", status="target") + + # Should get suffix after 'c', not reuse 'b' + result = compute_target_filename( + dbsession, + name="bar", + flavour=None, + date="2024-05-10", + ) + assert result == "bar_2024-05d.zim" + + def test_double_letter_suffix_progression( + self, dbsession: OrmSession, create_book_location: Callable[..., BookLocation] + ): + """Should progress to double letter suffixes when needed.""" + # Create locations through z + create_book_location(filename="baz_2024-06.zim", status="target") + for letter in "abcdefghijklmnopqrstuvwxyz": + create_book_location(filename=f"baz_2024-06{letter}.zim", status="target") + + # Should wrap to aa + result = compute_target_filename( + dbsession, + name="baz", + flavour=None, + date="2024-06-10", + ) + assert result == "baz_2024-06aa.zim" + + def test_flavour_prevents_collision( + self, dbsession: OrmSession, create_book_location: Callable[..., BookLocation] + ): + """Different flavours should not collide.""" + # Create location for name without flavour + create_book_location(filename="wiki_2024-07.zim", status="target") + + # Same name with flavour should not collide + result = compute_target_filename( + dbsession, + name="wiki", + flavour="maxi", + date="2024-07-10", + ) + assert result == "wiki_maxi_2024-07.zim" + + def test_mixed_status_locations_included( + self, dbsession: OrmSession, create_book_location: Callable[..., BookLocation] + ): + """Should consider both current and target locations.""" + # Create both current and target locations + create_book_location(filename="mixed_2024-08.zim", status="current") + create_book_location(filename="mixed_2024-08a.zim", status="target") + + # Should get suffix after 'a' + result = compute_target_filename( + dbsession, + name="mixed", + flavour=None, + date="2024-08-10", + ) + assert result == "mixed_2024-08b.zim" + + def test_different_period_no_collision( + self, dbsession: OrmSession, create_book_location: Callable[..., BookLocation] + ): + """Different periods should not collide.""" + # Create location for different period + create_book_location(filename="period_2024-08.zim", status="target") + + # Different period should not collide + result = compute_target_filename( + dbsession, + name="period", + flavour=None, + date="2024-09-10", + ) + assert result == "period_2024-09.zim" diff --git a/frontend/src/types/book.ts b/frontend/src/types/book.ts index 0d0826e..273fcf8 100644 --- a/frontend/src/types/book.ts +++ b/frontend/src/types/book.ts @@ -4,6 +4,14 @@ export interface Producer { unique_id: string } +export interface BookLocation { + warehouse_path_id: string + warehouse_name: string + folder_name: string + filename: string + status: string +} + export interface Book { id: string title_id?: string @@ -19,6 +27,8 @@ export interface Book { zim_metadata: Record events: string[] producer: Producer + current_locations: BookLocation[] + target_locations: BookLocation[] } export interface BookLight { diff --git a/frontend/src/views/BookView.vue b/frontend/src/views/BookView.vue index 949807b..a89c760 100644 --- a/frontend/src/views/BookView.vue +++ b/frontend/src/views/BookView.vue @@ -126,6 +126,64 @@ 
{{ JSON.stringify(book.zimcheck_result, null, 2) }}
+ + Current Locations + +
+ + + + Warehouse + Folder + Filename + + + + + {{ location.warehouse_name }} + {{ location.folder_name }} + + {{ location.filename }} + + + + +
+
No current locations
+ + + + Target Locations + +
+ + + + Warehouse + Folder + Filename + + + + + {{ location.warehouse_name }} + {{ location.folder_name }} + + {{ location.filename }} + + + + +
+
No target locations
+ + Events