diff --git a/tests/queries/__init__.py b/tests/queries/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/queries/test_author.py b/tests/queries/test_author.py new file mode 100644 index 0000000..4d97445 --- /dev/null +++ b/tests/queries/test_author.py @@ -0,0 +1,38 @@ +from sqlalchemy.orm import Session + +from worblehat.models import Author +from worblehat.queries.author import search_authors_by_name + + +def test_search_authors_by_name_matches_substring_case_insensitively(sql_session: Session) -> None: + tolkien = Author(name="J.R.R. Tolkien") + orwell = Author(name="George Orwell") + sql_session.add_all([tolkien, orwell]) + sql_session.flush() + + result = search_authors_by_name(sql_session, "tolkien") + + assert result == [tolkien] + + +def test_search_authors_by_name_returns_empty_when_no_match(sql_session: Session) -> None: + sql_session.add(Author(name="Roald Dahl")) + sql_session.flush() + + result = search_authors_by_name(sql_session, "nonexistent") + + assert result == [] + + +def test_search_authors_by_name_returns_all_matches(sql_session: Session) -> None: + tolkien = Author(name="J.R.R. Tolkien") + orwell = Author(name="George Orwell") + dahl = Author(name="Roald Dahl") + nesbo = Author(name="Jo Nesbø") + nakamura = Author(name="中村 文則") + sql_session.add_all([tolkien, orwell, dahl, nesbo, nakamura]) + sql_session.flush() + + result = search_authors_by_name(sql_session, "e") + + assert set(result) == {tolkien, orwell, nesbo} diff --git a/tests/queries/test_bookcase.py b/tests/queries/test_bookcase.py new file mode 100644 index 0000000..02ae537 --- /dev/null +++ b/tests/queries/test_bookcase.py @@ -0,0 +1,23 @@ +from sqlalchemy.orm import Session + +from worblehat.models import Bookcase +from worblehat.queries.bookcase import find_bookcase_by_name + + +def test_find_bookcase_by_name_returns_match(sql_session: Session) -> None: + bookcase = Bookcase(name="Bookcase A", description="A bookcase") + sql_session.add(bookcase) + sql_session.flush() + + result = find_bookcase_by_name(sql_session, "Bookcase A") + + assert result is bookcase + + +def test_find_bookcase_by_name_is_case_sensitive(sql_session: Session) -> None: + sql_session.add(Bookcase(name="Bookcase A")) + sql_session.flush() + + result = find_bookcase_by_name(sql_session, "bookcase a") + + assert result is None diff --git a/tests/queries/test_bookcase_item.py b/tests/queries/test_bookcase_item.py new file mode 100644 index 0000000..90b38eb --- /dev/null +++ b/tests/queries/test_bookcase_item.py @@ -0,0 +1,115 @@ +from sqlalchemy.orm import Session + +from worblehat.models import ( + Bookcase, + BookcaseItem, + BookcaseShelf, + MediaType, +) +from worblehat.queries.bookcase_item import ( + find_bookcase_item_by_isbn, + find_bookcase_item_by_name, + list_bookcase_items_by_owner, + search_bookcase_item_owners, + search_bookcase_items_by_title, +) + + +def _make_bookcase_item( + sql_session: Session, + name: str = "Some Book", + isbn: str = "1234567890", + owner: str = "PVV", +) -> BookcaseItem: + media_type = MediaType(name=f"Media type for {name}") + bookcase = Bookcase(name=f"Bookcase for {name}") + shelf = BookcaseShelf(row=0, column=0, bookcase=bookcase) + sql_session.add_all([media_type, bookcase, shelf]) + sql_session.flush() + + item = BookcaseItem(name, isbn, owner) + item.media_type = media_type + item.shelf = shelf + sql_session.add(item) + sql_session.flush() + return item + + +def test_find_bookcase_item_by_isbn_returns_match(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session, isbn="1111111111") + + result = find_bookcase_item_by_isbn(sql_session, "1111111111") + + assert result is item + + +def test_find_bookcase_item_by_isbn_returns_none_when_missing(sql_session: Session) -> None: + _make_bookcase_item(sql_session, isbn="1111111111") + + result = find_bookcase_item_by_isbn(sql_session, "2222222222") + + assert result is None + + +def test_find_bookcase_item_by_name_returns_match(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session, name="Unique Title") + + result = find_bookcase_item_by_name(sql_session, "Unique Title") + + assert result is item + + +def test_find_bookcase_item_by_name_returns_none_when_missing(sql_session: Session) -> None: + _make_bookcase_item(sql_session, name="Unique Title") + + result = find_bookcase_item_by_name(sql_session, "Other Title") + + assert result is None + + +def test_search_bookcase_items_by_title_matches_substring_case_insensitively( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session, name="The Great Gatsby", isbn="1111111111") + _make_bookcase_item(sql_session, name="Moby Dick", isbn="2222222222") + + result = search_bookcase_items_by_title(sql_session, "great gatsby") + + assert result == [item] + + +def test_search_bookcase_items_by_title_returns_empty_when_no_match(sql_session: Session) -> None: + _make_bookcase_item(sql_session, name="The Great Gatsby") + + result = search_bookcase_items_by_title(sql_session, "nonexistent") + + assert result == [] + + +def test_search_bookcase_item_owners_matches_substring_case_insensitively( + sql_session: Session, +) -> None: + _make_bookcase_item(sql_session, name="Book A", isbn="1111111111", owner="alice") + _make_bookcase_item(sql_session, name="Book B", isbn="2222222222", owner="bob") + + result = search_bookcase_item_owners(sql_session, "ALI") + + assert result == ["alice"] + + +def test_search_bookcase_item_owners_returns_distinct_owners(sql_session: Session) -> None: + _make_bookcase_item(sql_session, name="Book A", isbn="1111111111", owner="alice") + _make_bookcase_item(sql_session, name="Book B", isbn="2222222222", owner="alice") + + result = search_bookcase_item_owners(sql_session, "ali") + + assert result == ["alice"] + + +def test_list_bookcase_items_by_owner_returns_only_matching_owner(sql_session: Session) -> None: + item_a = _make_bookcase_item(sql_session, name="Book A", isbn="1111111111", owner="alice") + _make_bookcase_item(sql_session, name="Book B", isbn="2222222222", owner="bob") + + result = list_bookcase_items_by_owner(sql_session, "alice") + + assert result == [item_a] diff --git a/tests/queries/test_bookcase_shelf.py b/tests/queries/test_bookcase_shelf.py new file mode 100644 index 0000000..a23159e --- /dev/null +++ b/tests/queries/test_bookcase_shelf.py @@ -0,0 +1,92 @@ +from sqlalchemy.orm import Session + +from worblehat.models import Bookcase, BookcaseShelf +from worblehat.queries.bookcase_shelf import ( + find_bookcase_shelf, + list_bookcase_shelf_positions, + list_bookcase_shelfs_ordered, +) + + +def test_find_bookcase_shelf_returns_match(sql_session: Session) -> None: + bookcase = Bookcase(name="Bookcase A") + shelf = BookcaseShelf(row=0, column=0, bookcase=bookcase) + sql_session.add_all([bookcase, shelf]) + sql_session.flush() + + result = find_bookcase_shelf(sql_session, bookcase, column=0, row=0) + + assert result is shelf + + +def test_find_bookcase_shelf_returns_none_for_wrong_position(sql_session: Session) -> None: + bookcase = Bookcase(name="Bookcase A") + shelf = BookcaseShelf(row=0, column=0, bookcase=bookcase) + sql_session.add_all([bookcase, shelf]) + sql_session.flush() + + assert find_bookcase_shelf(sql_session, bookcase, column=1, row=0) is None + assert find_bookcase_shelf(sql_session, bookcase, column=0, row=1) is None + + +def test_find_bookcase_shelf_scopes_to_bookcase(sql_session: Session) -> None: + bookcase_a = Bookcase(name="Bookcase A") + bookcase_b = Bookcase(name="Bookcase B") + shelf_a = BookcaseShelf(row=0, column=0, bookcase=bookcase_a) + sql_session.add_all([bookcase_a, bookcase_b, shelf_a]) + sql_session.flush() + + assert find_bookcase_shelf(sql_session, bookcase_b, column=0, row=0) is None + + +def test_list_bookcase_shelfs_ordered(sql_session: Session) -> None: + bookcase_b = Bookcase(name="Bookcase B") + bookcase_a = Bookcase(name="Bookcase A") + sql_session.add_all([bookcase_b, bookcase_a]) + sql_session.flush() + + shelf_b1 = BookcaseShelf(row=1, column=0, bookcase=bookcase_b) + shelf_a2 = BookcaseShelf(row=0, column=2, bookcase=bookcase_a) + shelf_a1 = BookcaseShelf(row=0, column=1, bookcase=bookcase_a) + sql_session.add_all([shelf_b1, shelf_a2, shelf_a1]) + sql_session.flush() + + result = list_bookcase_shelfs_ordered(sql_session) + + assert result == [shelf_a1, shelf_a2, shelf_b1] + + +def test_list_bookcase_shelf_positions_returns_all_by_default(sql_session: Session) -> None: + bookcase = Bookcase(name="Bookcase A") + shelf_1 = BookcaseShelf(row=0, column=0, bookcase=bookcase) + shelf_2 = BookcaseShelf(row=1, column=1, bookcase=bookcase) + sql_session.add_all([bookcase, shelf_1, shelf_2]) + sql_session.flush() + + result = list_bookcase_shelf_positions(sql_session, bookcase) + + assert set(result) == {(0, 0), (1, 1)} + + +def test_list_bookcase_shelf_positions_filters_by_column(sql_session: Session) -> None: + bookcase = Bookcase(name="Bookcase A") + shelf_1 = BookcaseShelf(row=0, column=0, bookcase=bookcase) + shelf_2 = BookcaseShelf(row=1, column=1, bookcase=bookcase) + sql_session.add_all([bookcase, shelf_1, shelf_2]) + sql_session.flush() + + result = list_bookcase_shelf_positions(sql_session, bookcase, column=1) + + assert result == [(1, 1)] + + +def test_list_bookcase_shelf_positions_filters_by_row(sql_session: Session) -> None: + bookcase = Bookcase(name="Bookcase A") + shelf_1 = BookcaseShelf(row=0, column=0, bookcase=bookcase) + shelf_2 = BookcaseShelf(row=1, column=1, bookcase=bookcase) + sql_session.add_all([bookcase, shelf_1, shelf_2]) + sql_session.flush() + + result = list_bookcase_shelf_positions(sql_session, bookcase, row=1) + + assert result == [(1, 1)] diff --git a/tests/queries/test_borrowing.py b/tests/queries/test_borrowing.py new file mode 100644 index 0000000..fd9b046 --- /dev/null +++ b/tests/queries/test_borrowing.py @@ -0,0 +1,152 @@ +from datetime import datetime, timedelta + +from sqlalchemy.orm import Session + +from worblehat.models import ( + Bookcase, + BookcaseItem, + BookcaseItemBorrowing, + BookcaseShelf, + MediaType, +) +from worblehat.queries.borrowing import ( + has_active_borrowing, + list_active_borrowings, + list_active_borrowings_for_item, + list_borrowings_for_isbn, + list_overdue_borrowings, +) + + +def _make_bookcase_item( + sql_session: Session, + name: str = "Some Book", + isbn: str = "1234567890", +) -> BookcaseItem: + """Creates a BookcaseItem along with the MediaType/Bookcase/BookcaseShelf it needs.""" + media_type = MediaType(name=f"Media type for {name}") + bookcase = Bookcase(name=f"Bookcase for {name}") + shelf = BookcaseShelf(row=0, column=0, bookcase=bookcase) + sql_session.add_all([media_type, bookcase, shelf]) + sql_session.flush() + + item = BookcaseItem(name, isbn) + item.media_type = media_type + item.shelf = shelf + sql_session.add(item) + sql_session.flush() + return item + + +def test_list_active_borrowings_excludes_delivered(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + active = BookcaseItemBorrowing("alice", item) + delivered = BookcaseItemBorrowing("bob", item) + delivered.delivered = datetime.now() + sql_session.add_all([active, delivered]) + sql_session.flush() + + result = list_active_borrowings(sql_session) + + assert result == [active] + + +def test_list_active_borrowings_orders_by_end_time(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + later = BookcaseItemBorrowing("alice", item) + later.end_time = datetime.now() + timedelta(days=10) + sooner = BookcaseItemBorrowing("bob", item) + sooner.end_time = datetime.now() + timedelta(days=1) + sql_session.add_all([later, sooner]) + sql_session.flush() + + result = list_active_borrowings(sql_session) + + assert result == [sooner, later] + + +def test_list_active_borrowings_for_item_scopes_to_item(sql_session: Session) -> None: + item_a = _make_bookcase_item(sql_session, name="Book A", isbn="1111111111") + item_b = _make_bookcase_item(sql_session, name="Book B", isbn="2222222222") + borrowing_a = BookcaseItemBorrowing("alice", item_a) + borrowing_b = BookcaseItemBorrowing("bob", item_b) + sql_session.add_all([borrowing_a, borrowing_b]) + sql_session.flush() + + result = list_active_borrowings_for_item(sql_session, item_a) + + assert result == [borrowing_a] + + +def test_has_active_borrowing_true_when_undelivered_borrowing_exists(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + sql_session.add(BookcaseItemBorrowing("alice", item)) + sql_session.flush() + + assert has_active_borrowing(sql_session, "alice", item) is True + + +def test_has_active_borrowing_false_when_delivered(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + borrowing = BookcaseItemBorrowing("alice", item) + borrowing.delivered = datetime.now() + sql_session.add(borrowing) + sql_session.flush() + + assert has_active_borrowing(sql_session, "alice", item) is False + + +def test_has_active_borrowing_false_for_other_user(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + sql_session.add(BookcaseItemBorrowing("alice", item)) + sql_session.flush() + + assert has_active_borrowing(sql_session, "bob", item) is False + + +def test_list_borrowings_for_isbn_scopes_to_isbn(sql_session: Session) -> None: + item_a = _make_bookcase_item(sql_session, name="Book A", isbn="1111111111") + item_b = _make_bookcase_item(sql_session, name="Book B", isbn="2222222222") + borrowing_a = BookcaseItemBorrowing("alice", item_a) + borrowing_b = BookcaseItemBorrowing("bob", item_b) + sql_session.add_all([borrowing_a, borrowing_b]) + sql_session.flush() + + result = list_borrowings_for_isbn(sql_session, "1111111111") + + assert result == [borrowing_a] + + +def test_list_borrowings_for_isbn_orders_by_username(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + borrowing_bob = BookcaseItemBorrowing("bob", item) + borrowing_alice = BookcaseItemBorrowing("alice", item) + sql_session.add_all([borrowing_bob, borrowing_alice]) + sql_session.flush() + + result = list_borrowings_for_isbn(sql_session, item.isbn) + + assert result == [borrowing_alice, borrowing_bob] + + +def test_list_overdue_borrowings_only_returns_undelivered_past_deadline( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + + overdue = BookcaseItemBorrowing("alice", item) + overdue.end_time = datetime.now() - timedelta(days=1) + + not_yet_due = BookcaseItemBorrowing("bob", item) + not_yet_due.end_time = datetime.now() + timedelta(days=1) + + overdue_but_delivered = BookcaseItemBorrowing("carol", item) + overdue_but_delivered.end_time = datetime.now() - timedelta(days=1) + overdue_but_delivered.delivered = datetime.now() + + sql_session.add_all([overdue, not_yet_due, overdue_but_delivered]) + sql_session.flush() + + result = list_overdue_borrowings(sql_session) + + assert result == [overdue] diff --git a/tests/queries/test_borrowing_queue.py b/tests/queries/test_borrowing_queue.py new file mode 100644 index 0000000..d79a263 --- /dev/null +++ b/tests/queries/test_borrowing_queue.py @@ -0,0 +1,100 @@ +from datetime import datetime, timedelta + +from sqlalchemy.orm import Session + +from worblehat.models import ( + Bookcase, + BookcaseItem, + BookcaseItemBorrowingQueue, + BookcaseShelf, + MediaType, +) +from worblehat.queries.borrowing_queue import ( + is_in_borrowing_queue, + list_all_queue_items, + list_pending_queue_items_for_item, +) + + +def _make_bookcase_item( + sql_session: Session, + name: str = "Some Book", + isbn: str = "1234567890", +) -> BookcaseItem: + """Creates a BookcaseItem along with the MediaType/Bookcase/BookcaseShelf it needs.""" + media_type = MediaType(name=f"Media type for {name}") + bookcase = Bookcase(name=f"Bookcase for {name}") + shelf = BookcaseShelf(row=0, column=0, bookcase=bookcase) + sql_session.add_all([media_type, bookcase, shelf]) + sql_session.flush() + + item = BookcaseItem(name, isbn) + item.media_type = media_type + item.shelf = shelf + sql_session.add(item) + sql_session.flush() + return item + + +def test_list_all_queue_items_orders_by_entered_queue_time(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + later = BookcaseItemBorrowingQueue("alice", item) + later.entered_queue_time = datetime.now() + timedelta(hours=1) + sooner = BookcaseItemBorrowingQueue("bob", item) + sooner.entered_queue_time = datetime.now() + sql_session.add_all([later, sooner]) + sql_session.flush() + + result = list_all_queue_items(sql_session) + + assert result == [sooner, later] + + +def test_is_in_borrowing_queue_true_when_present(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + sql_session.add(BookcaseItemBorrowingQueue("alice", item)) + sql_session.flush() + + assert is_in_borrowing_queue(sql_session, "alice", item) is True + + +def test_is_in_borrowing_queue_false_for_other_user(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + sql_session.add(BookcaseItemBorrowingQueue("alice", item)) + sql_session.flush() + + assert is_in_borrowing_queue(sql_session, "bob", item) is False + + +def test_is_in_borrowing_queue_false_when_empty(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + sql_session.add(item) + sql_session.flush() + + assert is_in_borrowing_queue(sql_session, "alice", item) is False + + +def test_list_pending_queue_items_for_item_excludes_available_items(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + pending = BookcaseItemBorrowingQueue("alice", item) + became_available = BookcaseItemBorrowingQueue("bob", item) + became_available.item_became_available_time = datetime.now() + sql_session.add_all([pending, became_available]) + sql_session.flush() + + result = list_pending_queue_items_for_item(sql_session, item) + + assert result == [pending] + + +def test_list_pending_queue_items_for_item_scopes_to_item(sql_session: Session) -> None: + item_a = _make_bookcase_item(sql_session, name="Book A", isbn="1111111111") + item_b = _make_bookcase_item(sql_session, name="Book B", isbn="2222222222") + queue_item_a = BookcaseItemBorrowingQueue("alice", item_a) + queue_item_b = BookcaseItemBorrowingQueue("bob", item_b) + sql_session.add_all([queue_item_a, queue_item_b]) + sql_session.flush() + + result = list_pending_queue_items_for_item(sql_session, item_a) + + assert result == [queue_item_a] diff --git a/tests/queries/test_deadline_daemon.py b/tests/queries/test_deadline_daemon.py new file mode 100644 index 0000000..a5d8434 --- /dev/null +++ b/tests/queries/test_deadline_daemon.py @@ -0,0 +1,344 @@ +from datetime import datetime, timedelta + +from sqlalchemy.orm import Session + +from worblehat.models import ( + Bookcase, + BookcaseItem, + BookcaseItemBorrowing, + BookcaseItemBorrowingQueue, + BookcaseShelf, + DeadlineDaemonLastRunDatetime, + MediaType, +) +from worblehat.queries.deadline_daemon import ( + find_last_run, + find_next_queue_position, + list_close_deadline_borrowings, + list_expiring_queue_positions, + list_newly_available_queue_items, + list_overdue_queue_positions, + list_undelivered_overdue_borrowings, +) + + +def _make_bookcase_item( + sql_session: Session, + name: str = "Some Book", + isbn: str = "1234567890", +) -> BookcaseItem: + """Creates a BookcaseItem along with the MediaType/Bookcase/BookcaseShelf it needs.""" + media_type = MediaType(name=f"Media type for {name}") + bookcase = Bookcase(name=f"Bookcase for {name}") + shelf = BookcaseShelf(row=0, column=0, bookcase=bookcase) + sql_session.add_all([media_type, bookcase, shelf]) + sql_session.flush() + + item = BookcaseItem(name, isbn) + item.media_type = media_type + item.shelf = shelf + sql_session.add(item) + sql_session.flush() + return item + + +def test_find_last_run_returns_none_when_db_is_empty(sql_session: Session) -> None: + assert find_last_run(sql_session) is None + + +def test_find_last_run_returns_the_single_row(sql_session: Session) -> None: + last_run = DeadlineDaemonLastRunDatetime(time=datetime.now()) + sql_session.add(last_run) + sql_session.flush() + + assert find_last_run(sql_session) is last_run + + +def test_list_close_deadline_borrowings_matches_borrowings_ending_in_n_days( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + close = BookcaseItemBorrowing("alice", item) + close.end_time = now + timedelta(days=2) + + far = BookcaseItemBorrowing("bob", item) + far.end_time = now + timedelta(days=20) + + sql_session.add_all([close, far]) + sql_session.flush() + + result = list_close_deadline_borrowings( + sql_session, + day=2, + last_run_datetime=now - timedelta(minutes=1), + current_run_datetime=now + timedelta(minutes=1), + ) + + assert result == [close] + + +def test_list_close_deadline_borrowings_excludes_delivered(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + delivered = BookcaseItemBorrowing("alice", item) + delivered.end_time = now + timedelta(days=2) + delivered.delivered = now + + sql_session.add(delivered) + sql_session.flush() + + result = list_close_deadline_borrowings( + sql_session, + day=2, + last_run_datetime=now - timedelta(minutes=1), + current_run_datetime=now + timedelta(minutes=1), + ) + + assert result == [] + + +def test_list_undelivered_overdue_borrowings_only_returns_undelivered_past_deadline( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + overdue = BookcaseItemBorrowing("alice", item) + overdue.end_time = now - timedelta(days=1) + + not_yet_due = BookcaseItemBorrowing("bob", item) + not_yet_due.end_time = now + timedelta(days=1) + + overdue_but_delivered = BookcaseItemBorrowing("carol", item) + overdue_but_delivered.end_time = now - timedelta(days=1) + overdue_but_delivered.delivered = now + + sql_session.add_all([overdue, not_yet_due, overdue_but_delivered]) + sql_session.flush() + + result = list_undelivered_overdue_borrowings(sql_session, now) + + assert result == [overdue] + + +def test_list_newly_available_queue_items_requires_delivery_in_window(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + borrowing = BookcaseItemBorrowing("alice", item) + borrowing.delivered = now + queue_item = BookcaseItemBorrowingQueue("bob", item) + + sql_session.add_all([borrowing, queue_item]) + sql_session.flush() + + result = list_newly_available_queue_items( + sql_session, + last_run_datetime=now - timedelta(minutes=1), + current_run_datetime=now + timedelta(minutes=1), + ) + + assert result == [queue_item] + + +def test_list_newly_available_queue_items_excludes_delivery_outside_window( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + borrowing = BookcaseItemBorrowing("alice", item) + borrowing.delivered = now - timedelta(days=10) + queue_item = BookcaseItemBorrowingQueue("bob", item) + + sql_session.add_all([borrowing, queue_item]) + sql_session.flush() + + result = list_newly_available_queue_items( + sql_session, + last_run_datetime=now - timedelta(minutes=1), + current_run_datetime=now + timedelta(minutes=1), + ) + + assert result == [] + + +def test_list_newly_available_queue_items_excludes_undelivered_borrowings( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + borrowing = BookcaseItemBorrowing("alice", item) + queue_item = BookcaseItemBorrowingQueue("bob", item) + + sql_session.add_all([borrowing, queue_item]) + sql_session.flush() + + result = list_newly_available_queue_items( + sql_session, + last_run_datetime=now - timedelta(minutes=1), + current_run_datetime=now + timedelta(minutes=1), + ) + + assert result == [] + + +def test_list_newly_available_queue_items_excludes_expired_queue_entries( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + borrowing = BookcaseItemBorrowing("alice", item) + borrowing.delivered = now + queue_item = BookcaseItemBorrowingQueue("bob", item) + queue_item.expired = True + + sql_session.add_all([borrowing, queue_item]) + sql_session.flush() + + result = list_newly_available_queue_items( + sql_session, + last_run_datetime=now - timedelta(minutes=1), + current_run_datetime=now + timedelta(minutes=1), + ) + + assert result == [] + + +def test_list_expiring_queue_positions_matches_positions_in_window(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + sql_session.add(BookcaseItemBorrowing("someone", item)) + + in_window = BookcaseItemBorrowingQueue("alice", item) + in_window.item_became_available_time = now + + out_of_window = BookcaseItemBorrowingQueue("bob", item) + out_of_window.item_became_available_time = now - timedelta(days=100) + + sql_session.add_all([in_window, out_of_window]) + sql_session.flush() + + result = list_expiring_queue_positions( + sql_session, + last_run_datetime=now - timedelta(days=1), + current_run_datetime=now + timedelta(days=1), + ) + + assert result == [in_window] + + +def test_list_expiring_queue_positions_ignores_items_without_a_borrowing( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + queue_item = BookcaseItemBorrowingQueue("alice", item) + queue_item.item_became_available_time = now + sql_session.add(queue_item) + sql_session.flush() + + result = list_expiring_queue_positions( + sql_session, + last_run_datetime=now - timedelta(days=1), + current_run_datetime=now + timedelta(days=1), + ) + + assert result == [] + + +def test_list_overdue_queue_positions_matches_positions_older_than_expiry( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + overdue = BookcaseItemBorrowingQueue("alice", item) + overdue.item_became_available_time = now - timedelta(days=5) + + not_overdue = BookcaseItemBorrowingQueue("bob", item) + not_overdue.item_became_available_time = now + + sql_session.add_all([overdue, not_overdue]) + sql_session.flush() + + result = list_overdue_queue_positions( + sql_session, + queue_position_expiry_days=1, + current_run_datetime=now, + ) + + assert result == [overdue] + + +def test_list_overdue_queue_positions_excludes_already_expired(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + already_expired = BookcaseItemBorrowingQueue("alice", item) + already_expired.item_became_available_time = now - timedelta(days=5) + already_expired.expired = True + + sql_session.add(already_expired) + sql_session.flush() + + result = list_overdue_queue_positions( + sql_session, + queue_position_expiry_days=1, + current_run_datetime=now, + ) + + assert result == [] + + +def test_find_next_queue_position_returns_earliest_pending_entry(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + later = BookcaseItemBorrowingQueue("alice", item) + later.entered_queue_time = now + timedelta(hours=1) + sooner = BookcaseItemBorrowingQueue("bob", item) + sooner.entered_queue_time = now + + sql_session.add_all([later, sooner]) + sql_session.flush() + + result = find_next_queue_position(sql_session, item.uid) + + assert result is sooner + + +def test_find_next_queue_position_skips_entries_that_already_became_available( + sql_session: Session, +) -> None: + item = _make_bookcase_item(sql_session) + now = datetime.now() + + already_available = BookcaseItemBorrowingQueue("alice", item) + already_available.entered_queue_time = now + already_available.item_became_available_time = now + + pending = BookcaseItemBorrowingQueue("bob", item) + pending.entered_queue_time = now + timedelta(hours=1) + + sql_session.add_all([already_available, pending]) + sql_session.flush() + + result = find_next_queue_position(sql_session, item.uid) + + assert result is pending + + +def test_find_next_queue_position_returns_none_when_empty(sql_session: Session) -> None: + item = _make_bookcase_item(sql_session) + + result = find_next_queue_position(sql_session, item.uid) + + assert result is None diff --git a/tests/queries/test_media_type.py b/tests/queries/test_media_type.py new file mode 100644 index 0000000..b30049e --- /dev/null +++ b/tests/queries/test_media_type.py @@ -0,0 +1,25 @@ +import pytest +from sqlalchemy.exc import NoResultFound +from sqlalchemy.orm import Session + +from worblehat.models import MediaType +from worblehat.queries.media_type import find_media_type_by_name + + +def test_find_media_type_by_name_returns_match_case_insensitively(sql_session: Session) -> None: + book = MediaType(name="Book", description="A physical book") + comic = MediaType(name="Comic", description="A comic book") + sql_session.add_all([book, comic]) + sql_session.flush() + + result = find_media_type_by_name(sql_session, "book") + + assert result is book + + +def test_find_media_type_by_name_raises_when_missing(sql_session: Session) -> None: + sql_session.add(MediaType(name="Comic")) + sql_session.flush() + + with pytest.raises(NoResultFound): + find_media_type_by_name(sql_session, "book")