refactor: unittest now only test the service layer

before they were wrongfully testing the whole application
with routes and repostiory layer, and that is responsibility of
integration tests

Author: adiffpirate

Makefile

@@ -2,20 +2,15 @@
 
 help:
 	@echo "Available commands:"
-	@echo "  build            - Build the docker container"
 	@echo "  run              - Run the application (development with auto-reload, debug mode)"
+	@echo "  stop             - Stop the application"
 	@echo "  test             - Run all tests"
 	@echo "  test-unit        - Run unit tests"
 	@echo "  test-integration - Run integration tests with fresh database"
-	@echo "  stop             - Stop the application"
 	@echo "  clean            - Clean up Docker containers and volumes"
 
-build:
-	docker build -t python-fastapi-example-oms:dev .
-
 run:
-	$(MAKE) build
-	docker compose up -d
+	docker compose up --build -d
 	@echo "Application running at http://localhost:8000"
 	@echo "Swagger UI is available at http://localhost:8000/docs"
 
@@ -28,9 +23,8 @@ test:
 	$(MAKE) test-integration
 
 test-unit:
-	$(MAKE) build
 	@echo "Running unit tests..."
-	docker compose run --rm app python -m pytest tests/unit/ -v
+	docker run --rm --tty -v $(PWD):/app python-fastapi-example-oms:dev python -m pytest tests/unit/ -v
 
 test-integration:
 	$(MAKE) clean

app/modules/orders/repository.py

@@ -2,17 +2,38 @@
 from . import models
 
 
-def create_order(db: Session, item: str):
-    order = models.Order(item=item)
-    db.add(order)
-    db.commit()
-    db.refresh(order)
-    return order
+class OrderRepository:
+    def __init__(self, db: Session):
+        self._db = db
 
+    def create_order(self, item: str):
+        order = models.Order(item=item)
+        self._db.add(order)
+        self._db.commit()
+        self._db.refresh(order)
+        return order
 
-def get_order(db: Session, order_id: int):
-    return db.query(models.Order).filter(models.Order.id == order_id).first()
+    def get_order(self, order_id: int):
+        return self._db.query(models.Order).filter(models.Order.id == order_id).first()
 
+    def list_orders(self):
+        return self._db.query(models.Order).all()
 
-def list_orders(db: Session):
-    return db.query(models.Order).all()
+    def update_order(self, order_id: int, **kwargs):
+        order = self._db.query(models.Order).filter(models.Order.id == order_id).first()
+        if not order:
+            raise ValueError("Order not found")
+        for field, value in kwargs.items():
+            if value is not None:
+                setattr(order, field, value)
+        self._db.commit()
+        self._db.refresh(order)
+        return order
+
+    def delete_order(self, order_id: int):
+        order = self._db.query(models.Order).filter(models.Order.id == order_id).first()
+        if not order:
+            raise ValueError("Order not found")
+        self._db.delete(order)
+        self._db.commit()
+        return True

app/modules/orders/routes.py

@@ -2,25 +2,45 @@
 from sqlalchemy.orm import Session
 
 from app.db.session import get_db
-from . import service, schemas
+from . import service, schemas, repository
 
 router = APIRouter()
 
-# TODO: Require user authentication when calling these paths
+
+def get_orders_repository(db: Session = Depends(get_db)):
+    return repository.OrderRepository(db)
+
 
 @router.post("/", response_model=schemas.OrderRead)
-def create_order(payload: schemas.OrderCreate, db: Session = Depends(get_db)):
-    return service.create_order(db, payload.item)
+def create_order(payload: schemas.OrderCreate, repo: repository.OrderRepository = Depends(get_orders_repository)):
+    return service.create_order(repo, payload.item)
 
 
 @router.get("/{order_id}", response_model=schemas.OrderRead)
-def get_order(order_id: int, db: Session = Depends(get_db)):
+def get_order(order_id: int, repo: repository.OrderRepository = Depends(get_orders_repository)):
     try:
-        return service.get_order(db, order_id)
+        return service.get_order(repo, order_id)
     except ValueError:
         raise HTTPException(status_code=404, detail="Order not found")
 
 
 @router.get("/", response_model=list[schemas.OrderRead])
-def list_orders(db: Session = Depends(get_db)):
-    return service.list_orders(db)
+def list_orders(repo: repository.OrderRepository = Depends(get_orders_repository)):
+    return service.list_orders(repo)
+
+
+@router.put("/{order_id}", response_model=schemas.OrderRead)
+def update_order(order_id: int, payload: schemas.OrderUpdate, repo: repository.OrderRepository = Depends(get_orders_repository)):
+    try:
+        updates = {k: v for k, v in payload.model_dump(exclude_unset=True).items()}
+        return service.update_order(repo, order_id, **updates)
+    except ValueError:
+        raise HTTPException(status_code=404, detail="Order not found")
+
+
+@router.delete("/{order_id}", status_code=204)
+def delete_order(order_id: int, repo: repository.OrderRepository = Depends(get_orders_repository)):
+    try:
+        service.delete_order(repo, order_id)
+    except ValueError:
+        raise HTTPException(status_code=404, detail="Order not found")

app/modules/orders/schemas.py

@@ -6,6 +6,12 @@ class OrderCreate(BaseModel):
     item: str
 
 
+class OrderUpdate(BaseModel):
+    """Partial update schema for orders. Only provided fields are applied."""
+    item: str | None = None
+    status: OrderStatus | None = None
+
+
 class OrderRead(BaseModel):
     id: int
     item: str

app/modules/orders/service.py

@@ -1,17 +1,38 @@
-from sqlalchemy.orm import Session
-from . import repository, models
+from . import repository
 
 
-def create_order(db: Session, item: str):
-    return repository.create_order(db, item)
+def create_order(repo: repository.OrderRepository, item: str):
+    return repo.create_order(item)
 
 
-def get_order(db: Session, order_id: int):
-    order = repository.get_order(db, order_id)
+def get_order(repo: repository.OrderRepository, order_id: int):
+    order = repo.get_order(order_id)
     if not order:
         raise ValueError("Order not found")
     return order
 
 
-def list_orders(db: Session):
-    return repository.list_orders(db)
+def list_orders(repo: repository.OrderRepository):
+    return repo.list_orders()
+
+
+def update_order(repo: repository.OrderRepository, order_id: int, **kwargs):
+    """Update an order with the provided fields. Raises ValueError if not found."""
+    order = repo.get_order(order_id)
+    if not order:
+        raise ValueError("Order not found")
+    for key, value in kwargs.items():
+        if value is not None:
+            setattr(order, key, value)
+    repo._db.commit()
+    repo._db.refresh(order)
+    return order
+
+
+def delete_order(repo: repository.OrderRepository, order_id: int):
+    """Delete an order. Raises ValueError if not found."""
+    order = repo.get_order(order_id)
+    if not order:
+        raise ValueError("Order not found")
+    repo._db.delete(order)
+    repo._db.commit()

app/modules/users/repository.py

@@ -1,42 +1,38 @@
-from sqlalchemy.orm import Session
-from . import models
 import bcrypt
 import hashlib
+from sqlalchemy.orm import Session
+from . import models
 
 
-def create_user(db: Session, username: str, password: str):
-    # TODO: Add password strength check, should be long (20+ chars) OR complex (with uppercase, lowercase, numbers and symbols)
-
-    # Hash the password with bcrypt
-    # bcrypt has a 72-byte limit, so we'll hash long passwords with SHA-256 first
-    if len(password) > 72:
-        password = hashlib.sha256(password.encode()).hexdigest()
-
-    # Hash with bcrypt
-    hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
-
-    # Create user object
-    user = models.User(username=username, hashed_password=hashed_password.decode('utf-8'))
+class UserRepository:
+    def __init__(self, db: Session):
+        self._db = db
 
-    # Add to database
-    db.add(user)
-    db.commit()
-    db.refresh(user)
-    return user
+    def create_user(self, username: str, password: str):
+        hashed_password = self._hash_password(password)
+        user = models.User(username=username, hashed_password=hashed_password)
+        self._db.add(user)
+        self._db.commit()
+        self._db.refresh(user)
+        return user
 
+    def get_user_by_username(self, username: str):
+        return self._db.query(models.User).filter(models.User.username == username).first()
 
-def get_user_by_username(db: Session, username: str):
-    return db.query(models.User).filter(models.User.username == username).first()
+    def get_user_by_id(self, user_id: int):
+        return self._db.query(models.User).filter(models.User.id == user_id).first()
 
+    def _hash_password(self, password: str) -> str:
+        if len(password) > 72:
+            password = hashlib.sha256(password.encode()).hexdigest()
+        return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
 
-def get_user_by_id(db: Session, user_id: int):
-    return db.query(models.User).filter(models.User.id == user_id).first()
+    @staticmethod
+    def verify_password(plain_password: str, hashed_password: str) -> bool:
+        if len(plain_password) > 72:
+            plain_password = hashlib.sha256(plain_password.encode()).hexdigest()
+        return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8'))
 
 
 def verify_password(plain_password: str, hashed_password: str) -> bool:
-    # Handle long passwords by hashing them first
-    if len(plain_password) > 72:
-        plain_password = hashlib.sha256(plain_password.encode()).hexdigest()
-
-    # Verify password
-    return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8'))
+    return UserRepository.verify_password(plain_password, hashed_password)

app/modules/users/routes.py

@@ -2,31 +2,32 @@
 from sqlalchemy.orm import Session
 
 from app.db.session import get_db
-from . import service, schemas
+from . import service, schemas, repository
 
 router = APIRouter()
 
 
+def get_users_repository(db: Session = Depends(get_db)):
+    return repository.UserRepository(db)
+
+
 @router.post("/register", response_model=schemas.UserRead)
-def register_user(payload: schemas.UserCreate, db: Session = Depends(get_db)):
+def register_user(payload: schemas.UserCreate, repo: repository.UserRepository = Depends(get_users_repository)):
     try:
-        return service.create_user(db, payload.username, payload.password)
+        return service.create_user(repo, payload.username, payload.password)
     except HTTPException:
-        raise  # Re-raise HTTP exceptions
-    except Exception as e:
+        raise
+    except Exception:
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail="Error creating user"
         )
 
 
 @router.post("/login", response_model=schemas.UserToken)
-def login_user(payload: schemas.UserAuthenticate, db: Session = Depends(get_db)):
-    user = service.authenticate_user(db, payload.username, payload.password)
-
-    # In a real application, you would generate a JWT token here
-    # For now, we'll return a simple token structure
+def login_user(payload: schemas.UserAuthenticate, repo: repository.UserRepository = Depends(get_users_repository)):
+    user = service.authenticate_user(repo, payload.username, payload.password)
     return schemas.UserToken(
         access_token=f"fake-jwt-token-for-{user.username}",
         token_type="bearer"
-    )
+    )

app/modules/users/service.py

@@ -1,32 +1,23 @@
-from sqlalchemy.orm import Session
 from . import repository, schemas
 from fastapi import HTTPException, status
 
-# TODO: Evaluate if the logic to check for user existance should actually be here
 
-def create_user(db: Session, username: str, password: str):
-    # Check if user already exists
-    existing_user = repository.get_user_by_username(db, username)
+def create_user(repo: repository.UserRepository, username: str, password: str):
+    existing_user = repo.get_user_by_username(username)
     if existing_user:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="Username already registered"
         )
+    return repo.create_user(username, password)
 
-    # Create new user
-    return repository.create_user(db, username, password)
 
-
-def authenticate_user(db: Session, username: str, password: str):
-    # Get user from database
-    user = repository.get_user_by_username(db, username)
-
-    # Check if user exists and password is correct
-    if not user or not repository.verify_password(password, user.hashed_password):
+def authenticate_user(repo: repository.UserRepository, username: str, password: str):
+    user = repo.get_user_by_username(username)
+    if not user or not repo.verify_password(password, user.hashed_password):
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Incorrect username or password",
             headers={"WWW-Authenticate": "Bearer"},
         )
-
     return user