moved fastapi to digifeeds/database; add tests for crud methods

aim/crud.py → aim/digifeeds/database/crud.py

@@ -1,5 +1,6 @@
 from sqlalchemy.orm import Session
-from aim import models, schemas
+from aim.digifeeds.database import schemas
+from aim.digifeeds.database import models
 
 def get_item(db: Session, barcode: str):
     return db.query(models.Item).filter(models.Item.barcode == barcode).first()

aim/main.py → aim/digifeeds/database/main.py

@@ -1,8 +1,8 @@
 from fastapi import Depends, FastAPI, HTTPException
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session, sessionmaker
+from aim.digifeeds.database import crud, models, schemas
 from aim.services import S
-from aim import models, schemas, crud
 
 engine = create_engine(S.mysql_database)
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

aim/schemas.py → aim/digifeeds/database/schemas.py

@@ -1,21 +1,17 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ConfigDict
 from datetime import datetime
 
 class ItemStatus(BaseModel):
     name: str = Field(alias="status_name")
     description: str = Field(alias="status_description")
     created_at: datetime
 
-    class Config:
-        from_attributes = True
-        populate_by_name = True
+    model_config = ConfigDict(populate_by_name=True, from_attributes=True)
 
 class ItemBase(BaseModel):
-    barcode: str = Field(alias="item_barcode")
+    model_config = ConfigDict(populate_by_name=True, from_attributes=True)
 
-    class Config:
-        from_attributes = True
-        populate_by_name = True
+    barcode: str = Field(alias="item_barcode")
 
 class Item(ItemBase):
     created_at: datetime

tests/conftest.py

@@ -3,7 +3,7 @@
 import sqlalchemy as sa
 from sqlalchemy import StaticPool, create_engine
 from sqlalchemy.orm import sessionmaker
-from aim.models import Base, load_statuses, Status
+from aim.digifeeds.database.models import Base, load_statuses, Status
 
 from aim.services import S
 

tests/digifeeds/database/test_crud.py

@@ -0,0 +1,66 @@
+from aim.digifeeds.database.crud import (
+    get_item,
+    get_items,
+    add_item,
+    get_status,
+    get_statuses,
+    add_item_status
+)
+from aim.digifeeds.database.schemas import ItemCreate
+
+class TestCrud:
+    def test_get_item(self, db_session):
+        item = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode"))
+        barcode = item.barcode
+        item_in_db = get_item(barcode=barcode, db=db_session)
+        assert(item_in_db.barcode) == "valid_barcode"
+
+    def test_get_item_that_does_not_exist(self, db_session):
+        item_in_db = get_item(barcode="does not exist", db=db_session)
+        assert(item_in_db) == None
+
+    def test_get_items_all(self, db_session):
+        item1 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode"))
+        item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
+        status = get_status(db=db_session, name="in_zephir")
+        add_item_status(db=db_session,item=item1, status=status)
+        items = get_items(db=db_session, in_zephir=None)
+        db_session.refresh(item1)
+        db_session.refresh(item2)
+        assert(items[0]) == item1
+        assert(items[1]) == item2
+
+    def test_get_items_in_zephir(self, db_session):
+        item1 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode"))
+        item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
+        status = get_status(db=db_session, name="in_zephir")
+        add_item_status(db=db_session,item=item1, status=status)
+        items = get_items(db=db_session, in_zephir=True)
+        db_session.refresh(item1)
+        db_session.refresh(item2)
+        assert(len(items)) == 1
+        assert(items[0]) == item1
+
+    def test_get_items_not_in_zephir(self, db_session):
+        item1 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode"))
+        item2 = add_item(db=db_session, item=ItemCreate(barcode="valid_barcode2"))
+        status = get_status(db=db_session, name="in_zephir")
+        add_item_status(db=db_session,item=item1, status=status)
+        items = get_items(db=db_session, in_zephir=False)
+        db_session.refresh(item1)
+        db_session.refresh(item2)
+        assert(len(items)) == 1
+        assert(items[0]) == item2
+
+    def test_get_status_that_exists(self, db_session):
+        status = get_status(db=db_session, name="in_zephir")
+        assert(status.name) == "in_zephir"
+
+    def test_get_status_that_does_not_exist(self, db_session):
+        status = get_status(db=db_session, name="does_not_exist")
+        assert(status) == None 
+
+    def test_get_statuses(self, db_session):
+        statuses = get_statuses(db=db_session)
+        assert(len(statuses)) > 1
+        assert(statuses[0].name) == "in_zephir"

tests/test_models.py → tests/digifeeds/database/test_models.py

@@ -1,4 +1,4 @@
-from aim.models import Item, Status, ItemStatus
+from aim.digifeeds.database.models import Item, Status, ItemStatus
 
 class TestItem:
     def test_item_valid(self, db_session):
@@ -16,13 +16,16 @@ def test_item_statuses(self, db_session):
         status = db_session.query(Status).filter_by(name="in_zephir").first()
         db_session.refresh(item)
         assert(len(item.statuses)) == 0
+
         item_status = ItemStatus(item=item,status=status)
         db_session.add(item_status)
         db_session.commit()
         db_session.refresh(item)
+
         assert item.barcode == "valid_barcode"
         assert(len(item.statuses)) == 1
         assert(item.statuses[0].created_at)
+        assert item.statuses[0].status_name == "in_zephir"