Initial document CRUD tests

* Test for read_* methods
* Infrastructure for making it work (utils)

Author: jerome-white

backend/app/tests/crud/documents/_utils.py

@@ -0,0 +1,52 @@
+import functools as ft
+from uuid import UUID
+from pathlib import Path
+
+from sqlmodel import Session, delete
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.dialects.sqlite import insert
+
+from app.crud import DocumentCrud
+from app.core.config import settings
+from app.crud.user import get_user_by_email
+from app.models import Document, UserCreate
+
+class Constants:
+    n_documents = 10
+
+@ft.cache
+def get_user_id_by_email(session: Session):
+    user = get_user_by_email(session=session, email=settings.FIRST_SUPERUSER)
+    return user.id
+
+
+@ft.cache
+def int_to_uuid(value):
+    return UUID(int=value)
+
+def rm_documents(session: Session):
+    session.exec(delete(Document))
+    session.commit()
+
+def insert_documents(session: Session, n: int):
+    owner_id = get_user_id_by_email(session)
+
+    crud = DocumentCrud(session)
+    for i in range(n):
+        args = str(int_to_uuid(i)).split('-')
+        fname = Path('/', *args).with_suffix('.xyz')
+        document = Document(
+            id=int_to_uuid(i),
+            owner_id=owner_id,
+            fname=fname.name,
+            object_store_url=fname.as_uri(),
+        )
+        session.add(document)
+        session.commit()
+        session.refresh(document)
+
+        yield document
+
+def insert_document(session: Session):
+    (document, ) = insert_documents(session, 1)
+    return document

backend/app/tests/crud/documents/test_read_many.py

@@ -0,0 +1,69 @@
+import pytest
+from sqlmodel import Session
+
+from app.crud import DocumentCrud
+from app.core.config import settings
+
+from _utils import (
+    Constants,
+    get_user_id_by_email,
+    insert_documents,
+    int_to_uuid,
+    rm_documents,
+)
+
+@pytest.fixture
+def document_collection(db: Session):
+    rm_documents(db)
+    return list(insert_documents(db, Constants.n_documents))
+
+@pytest.fixture(scope='class')
+def clean_db_fixture(db: Session):
+    yield
+    rm_documents(db)
+
+@pytest.mark.usefixtures('clean_db_fixture')
+class TestDatabaseReadMany:
+    def test_number_read_is_expected(
+            self,
+            db: Session,
+            document_collection: list,
+    ):
+        crud = DocumentCrud(db)
+        owner_id = get_user_id_by_email(db)
+        documents = crud.read_many(owner_id)
+
+        assert len(documents) == Constants.n_documents
+
+    def test_deleted_docs_excluded(
+            self,
+            db: Session,
+            document_collection: list,
+    ):
+        assert all(x.deleted_at is None for x in document_collection)
+
+    def test_skip_is_respected(
+            self,
+            db: Session,
+            document_collection: list,
+    ):
+        crud = DocumentCrud(db)
+        owner_id = get_user_id_by_email(db)
+        skip = Constants.n_documents // 2
+        doc_ids = set(x.id for x in crud.read_many(owner_id, skip=skip))
+
+        for i in range(skip, Constants.n_documents):
+            doc = int_to_uuid(i)
+            assert doc in doc_ids
+
+    def test_limit_is_respected(
+            self,
+            db: Session,
+            document_collection: list,
+    ):
+        crud = DocumentCrud(db)
+        owner_id = get_user_id_by_email(db)
+        limit = Constants.n_documents // 2
+        documents = crud.read_many(owner_id, limit=limit)
+
+        assert len(documents) == limit

backend/app/tests/crud/documents/test_read_one.py

@@ -0,0 +1,35 @@
+import pytest
+from sqlmodel import Session
+from sqlalchemy.exc import NoResultFound
+
+from app.crud import DocumentCrud
+
+from _utils import insert_document, int_to_uuid, rm_documents
+
+@pytest.fixture
+def clean_db_fixture(db: Session):
+    rm_documents(db)
+    yield
+    rm_documents(db)
+
+class TestDatabaseReadOne:
+    def test_can_select_valid_id(
+            self,
+            db: Session,
+            clean_db_fixture: None,
+    ):
+        crud = DocumentCrud(db)
+        document = insert_document(db)
+        result = crud.read_one(document.id)
+
+        assert result.id == document.id
+
+    def test_cannot_select_invalid_id(
+            self,
+            db: Session,
+            clean_db_fixture: None,
+    ):
+        crud = DocumentCrud(db)
+        document_id = int_to_uuid(0)
+        with pytest.raises(NoResultFound):
+            crud.read_one(document_id)