added tests and fixing user model

app/models/user.py

@@ -1,6 +1,7 @@
 from sqlalchemy import Column, Integer, String,Boolean
 from app.database.base import Base
 from sqlalchemy.orm import relationship
+from app.models.task import Task
 
 class User(Base):
     __tablename__ = "users"

app/services/user_service.py

@@ -23,7 +23,11 @@ def create_user(db: Session, user: UserCreate):
         email=user.email, 
         hashed_password=get_password_hash(user.password)
     )
-    db.add(db_user)
-    db.commit()
-    db.refresh(db_user)
-    return db_user
+    try:
+        db.add(db_user)
+        db.commit()
+        db.refresh(db_user)
+        return db_user
+    except Exception as e:
+        db.rollback()
+        raise e

requirements.txt

@@ -1,30 +1,48 @@
+alembic==1.13.2
 annotated-types==0.7.0
 anyio==4.4.0
+bcrypt==4.1.3
 certifi==2024.7.4
+cffi==1.16.0
 click==8.1.7
+cryptography==42.0.8
 dnspython==2.6.1
+ecdsa==0.19.0
 email_validator==2.2.0
 fastapi==0.111.0
 fastapi-cli==0.0.4
+greenlet==3.0.3
 h11==0.14.0
 httpcore==1.0.5
 httptools==0.6.1
 httpx==0.27.0
 idna==3.7
+iniconfig==2.0.0
 Jinja2==3.1.4
+Mako==1.3.5
 markdown-it-py==3.0.0
 MarkupSafe==2.1.5
 mdurl==0.1.2
 orjson==3.10.6
+packaging==24.1
+passlib==1.7.4
+pluggy==1.5.0
+pyasn1==0.6.0
+pycparser==2.22
 pydantic==2.8.2
 pydantic_core==2.20.1
 Pygments==2.18.0
+pytest==8.2.2
 python-dotenv==1.0.1
+python-jose==3.3.0
 python-multipart==0.0.9
 PyYAML==6.0.1
 rich==13.7.1
+rsa==4.9
 shellingham==1.5.4
+six==1.16.0
 sniffio==1.3.1
+SQLAlchemy==2.0.31
 starlette==0.37.2
 typer==0.12.3
 typing_extensions==4.12.2
@@ -33,10 +51,3 @@ uvicorn==0.30.1
 uvloop==0.19.0
 watchfiles==0.22.0
 websockets==12.0
-fastapi
-uvicorn[standard]
-SQLAlchemy
-pydantic
-alembic
-passlib[bcrypt]
-python-jose[cryptography]

tests/test_auth.py

@@ -0,0 +1,40 @@
+import pytest
+from datetime import datetime, timedelta
+from jose import jwt
+from fastapi import HTTPException, status
+from sqlalchemy.orm import Session
+from app.services.auth import create_access_token, get_current_user
+from app.models.user import User
+from app.database.base import Base
+from app.services.user_service import get_user_by_email
+import os
+
+os.environ["SECRET_KEY"] = "test_secret_key"
+os.environ["ALGORITHM"] = "HS256"
+
+@pytest.fixture(scope="module")
+def db():
+    from sqlalchemy import create_engine
+    engine = create_engine('sqlite:///:memory:')
+    Base.metadata.create_all(bind=engine)
+    connection = engine.connect()
+    yield connection
+    connection.close()
+
+def mock_get_current_user(db: Session, token: str):
+    payload = jwt.decode(token, os.getenv("SECRET_KEY"), algorithms=[os.getenv("ALGORITHM")])
+    email = payload.get("sub")
+    if email is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
+    user = get_user_by_email(db, email=email)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
+    return user
+
+def test_create_access_token():
+    # test data
+    user_data = {"sub": "[email protected]"}
+    access_token = create_access_token(user_data)
+
+    # Validate token
+    assert isinstance(access_token, str)

tests/test_tasks_service.py

@@ -0,0 +1,89 @@
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from app.models.task import Task
+from app.schemas.task import TaskCreate, TaskUpdate
+from app.database.base import Base
+from app.services.task_service import create_task, get_tasks, get_task_by_id, update_task, delete_task
+
+DATABASE_URL = "sqlite:///./test.db"
+engine = create_engine(DATABASE_URL)
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base.metadata.create_all(bind=engine)
+
+@pytest.fixture(scope="module")
+def db():
+    connection = engine.connect()
+    transaction = connection.begin()
+    session = TestingSessionLocal(bind=connection)
+    yield session
+    session.close()
+    transaction.rollback()
+    connection.close()
+
+def test_get_tasks(db):
+    user_id = 1  
+    task_data_1 = TaskCreate(title="Task 1", description="Task 1 description")
+    task_data_2 = TaskCreate(title="Task 2", description="Task 2 description")
+    create_task(db, task_data_1, user_id)
+    create_task(db, task_data_2, user_id)
+
+
+    tasks = get_tasks(db, user_id)
+
+    assert len(tasks) == 2
+    assert tasks[0].title == "Task 1"
+    assert tasks[1].title == "Task 2"
+
+def test_create_task(db):
+    user_id = 1  
+    task_data = TaskCreate(title="Test Task", description="Testing task creation")
+
+    created_task = create_task(db, task_data, user_id)
+
+    assert created_task.id is not None
+    assert created_task.title == "Test Task"
+    assert created_task.description == "Testing task creation"
+    assert created_task.user_id == user_id
+
+
+
+def test_get_task_by_id(db):
+    # Test data
+    user_id = 1 
+    task_data = TaskCreate(title="Test Task", description="Testing get task by id")
+    created_task = create_task(db, task_data, user_id)
+
+    fetched_task = get_task_by_id(db, created_task.id, user_id)
+
+    assert fetched_task is not None
+    assert fetched_task.title == "Test Task"
+    assert fetched_task.description == "Testing get task by id"
+
+
+def test_update_task(db):
+    user_id = 1 
+    task_data = TaskCreate(title="Initial Task", description="Initial description")
+    created_task = create_task(db, task_data, user_id)
+
+    updated_data = TaskUpdate(title="Updated Task", description="Updated description")
+
+    updated_task = update_task(db, created_task.id, updated_data, user_id)
+
+    assert updated_task is not None
+    assert updated_task.id == created_task.id
+    assert updated_task.title == "Updated Task"
+    assert updated_task.description == "Updated description"
+
+def test_delete_task(db):
+    user_id = 1  
+    task_data = TaskCreate(title="Task to delete", description="Task to delete description")
+    created_task = create_task(db, task_data, user_id)
+
+    deleted_task = delete_task(db, created_task.id, user_id)
+
+    assert deleted_task is not None
+    assert deleted_task.id == created_task.id
+
+    fetched_task = get_task_by_id(db, created_task.id, user_id)
+    assert fetched_task is None  

tests/test_user_service.py

@@ -0,0 +1,44 @@
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from app.services.user_service import create_user, get_user_by_email, verify_password
+from app.schemas.user import UserCreate
+from app.database.base import Base
+from sqlalchemy.exc import IntegrityError
+
+DATABASE_URL = "sqlite:///./test.db"
+engine = create_engine(DATABASE_URL)
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base.metadata.create_all(bind=engine)
+
+@pytest.fixture(scope="function", autouse=True)
+def db():
+    connection = engine.connect()
+    transaction = connection.begin()
+    session = TestingSessionLocal(bind=connection)
+    yield session
+    session.close()
+    transaction.rollback()
+    connection.close()
+
+def test_create_user(db):
+    user_data = UserCreate(username="testuser", email="[email protected]", password="testpassword")
+    user = create_user(db, user_data)
+    assert user.username == "testuser"
+    assert user.email == "[email protected]"
+    assert verify_password("testpassword", user.hashed_password)
+
+def test_get_user_by_email(db):
+    user_data = UserCreate(username="testuser3", email="[email protected]", password="testpassword")
+    user = create_user(db, user_data)
+    fetched_user = get_user_by_email(db, email="[email protected]")
+    assert fetched_user.username == "testuser3"
+    assert fetched_user.email == "[email protected]"
+
+def test_create_duplicate_user(db):
+    user_data = UserCreate(username="testuser2", email="[email protected]", password="testpassword")
+    create_user(db, user_data)
+
+    with pytest.raises(IntegrityError):
+        duplicate_user_data = UserCreate(username="testuser2", email="[email protected]", password="testpassword")
+        create_user(db, duplicate_user_data)