zwischenstand

.gitignore

@@ -1,3 +1,4 @@
 node_modules
 backend/venv/lib/**
 backend/venv/lib64/*
+backend/dev.db

.vscode/settings.json

@@ -2,5 +2,10 @@
     "cSpell.language": "en,de-DE",
     "cSpell.words": [
         "Chiara"
-    ]
+    ],
+    "python.testing.pytestArgs": [
+        "backend"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
 }

backend/src/backend/internal/authentication.py

@@ -0,0 +1,35 @@
+from sqlmodel import select
+from backend.models.user import User
+from backend.internal.database import get_session
+from sqlmodel import Session
+from fastapi import HTTPException
+from fastapi.security.oauth2 import OAuth2PasswordBearer
+from fastapi import Depends
+from datetime import datetime
+from datetime import timedelta
+
+
+SECRET_KEY = "secret-key"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_DAYS = 360
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + (expires_delta or timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS))
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+async def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        user_code: str = payload.get("sub")
+        if user_code is None:
+            raise HTTPException(status_code=401, detail="Invalid token")
+    except JWTError:
+        raise HTTPException(status_code=401, detail="Invalid token")
+    user = session.exec(select(User).where(User.user_code == user_code))
+    if user is None:
+        raise HTTPException(status_code=401, detail="User not found")
+    return user

backend/src/backend/internal/database.py

@@ -2,6 +2,7 @@ from collections.abc import Generator
 from sqlmodel import Session
 from sqlmodel import SQLModel
 from sqlmodel import create_engine
+
 sqlite_file = "dev.db"
 sqlite_url = f"sqlite:///{sqlite_file}"
 

backend/src/backend/internal/user_code_gen.py

@@ -0,0 +1,5 @@
+import string
+import random
+
+def generate_user_code() -> str:
+    return ''.join(random.choices(string.ascii_uppercase, k=4))

backend/src/backend/main.py

@@ -1,4 +1,5 @@
-from .routers import users
+from sqlmodel import SQLModel
+from .routers import users, comments
 from backend.internal.database import create_db_and_tables
 from fastapi import APIRouter
 from fastapi import FastAPI
@@ -16,3 +17,4 @@ def on_startup():
 router = APIRouter()
 
 app.include_router(users.router)
+app.include_router(comments.router)

backend/src/backend/models/__init__.py

@@ -0,0 +1,2 @@
+from .user import User
+from .comment import Comment

backend/src/backend/models/comment.py

@@ -1,20 +1,22 @@
+from dataclasses import dataclass
+from datetime import datetime
 import uuid
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 from sqlmodel import Relationship
-from sqlmodel import DateTime
 from sqlmodel import Field
 from sqlmodel import SQLModel
 
 if TYPE_CHECKING:
-    from ..models.user import User
+    from backend.models.user import User
 
 class CommentBase(SQLModel):
     pass
 
+@dataclass
 class Comment(CommentBase, table=True):
-    id: uuid.UUID | None = Field(default_factory=uuid.uuid4, primary_key=True)
+    id: uuid.UUID | None = Field(default_factory=uuid.uuid7, primary_key=True)
     slug: str = Field(index=True)
     content: str = Field()
-    date_time: DateTime = Field(default=DateTime())
+    date_time: datetime = Field(default_factory=datetime.now)
     user_id: uuid.UUID = Field(foreign_key="user.id")
-    user: Optional["User"] = Relationship(back_populates="comments")
+    user: "User" = Relationship(back_populates="comments")

backend/src/backend/models/user.py

@@ -1,3 +1,6 @@
+from uuid import uuid7
+from datetime import datetime
+from backend.internal.user_code_gen import generate_user_code
 import uuid
 from typing import TYPE_CHECKING
 from sqlmodel import Field
@@ -5,20 +8,46 @@ from sqlmodel import Relationship
 from sqlmodel import SQLModel
 
 if TYPE_CHECKING:
-    from ..models.comment import Comment
+    from backend.models.comment import Comment
 
 class UserBase(SQLModel):
     pass
 
-class User(UserBase):
+class User(UserBase, table=True):
-    id: uuid.UUID | None = Field(default_factory=uuid.uuid4, primary_key=True)
+    id: uuid.UUID = Field(default_factory=uuid.uuid7, primary_key=True)
     email: str = Field(unique=True)
     name: str = Field(unique=True)
-    user_code: str = Field(unique=True) # TODO: create generation function
+    user_code: str = Field(default_factory=generate_user_code,unique=True)
     subscription: bool = Field()
+    created_at: datetime = Field(default_factory=datetime.now)
+
     comments: list["Comment"] = Relationship(back_populates="user")
 
-class RegisterUser(UserBase):
+    def factory(self,
+        id: uuid.UUID = uuid.uuid7(),
+        email: str = "Email@Test.mail",
+        name: str = "TestName",
+        user_code: str = "WYZZ",
+        subscription: bool = False,
+        ) -> User:
+        self.id=id
+        self.email=email
+        self.name=name
+        self.user_code=user_code
+        self.subscription=subscription
+        return self
+
+class NewUser(UserBase):
     email: str
     name: str
     subscription: bool
+
+    def factory(self,
+        email: str = "Email@Test.mail",
+        name: str = "TestName",
+        subscription: bool = False
+    ) -> NewUser:
+        self.email = email
+        self.name = name
+        self.subscription = subscription
+        return self

backend/src/backend/routers/comments.py

@@ -2,3 +2,23 @@
 # add comment
 # delete comment
 # get comments
+
+from backend.internal.database import get_session
+from fastapi import Depends
+from sqlmodel import Session
+from fastapi import APIRouter
+from ..models.comment import Comment 
+
+router = APIRouter(
+    prefix="/comments",
+    tags=["comments"],
+    dependencies=[],
+    responses={404: {"description": "Not found"}},
+)
+
+@router.post("/")
+def register_user(session: Session = Depends(get_session)):
+    comment = {"test":"test"}
+    comment1 = Comment.model_validate(comment)
+    session.add(comment)
+    session.commit()

backend/src/backend/routers/test_users.py

@@ -1,15 +1,84 @@
-from httpx2 import post
+from backend.models.user import NewUser
-from backend.models.user import RegisterUser
+from sqlmodel import select
+from backend.models.user import User
+from backend.internal.database import get_session
+from sqlmodel import Session
+from sqlmodel import StaticPool
+from sqlmodel import create_engine
+import pytest
+from backend.internal.database import engine
+from sqlmodel import SQLModel
 from backend.main import app
 from fastapi.testclient import TestClient
 
-client = TestClient(app)
+@pytest.fixture(name="session")
+def session_fixture():
+    engine = create_engine(
+        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
+    )
+    SQLModel.metadata.create_all(engine)
+    with Session(engine) as session:
+        yield session
 
-def test_register_user_works():
+
+@pytest.fixture(name="client")  
+def client_fixture(session: Session):  
+    def get_session_override():  
+        return session
+
+    app.dependency_overrides[get_session] = get_session_override  
+
+    client = TestClient(app)  
+    yield client  
+    app.dependency_overrides.clear()  
+
+def test_register_user_saved(session:Session, client: TestClient):
     new_user = {
-        "email": "Test",
+        "email": "Test@Test.test",
         "name": "Test1",
         "subscription": True
     }
     response = client.post(url="/users/", json=new_user)
+    assert response.status_code == 201
+
+    db_user = session.exec(select(User).where(User.email == "Test@Test.test")).first()
+    assert db_user
+    assert db_user.email == "Test@Test.test"
+    assert db_user.name == "Test1"
+    assert db_user.subscription == True
+
+def test_email_not_unique_returns_400(session: Session, client: TestClient):
+    new_user = NewUser(email="NotUnique",name="Müller", subscription=True) # TODO: durch factory ersetzen
+
+    user = User().factory(email="NotUnique")
+    session.add(user)
+    session.commit()
+    response = client.post(url="/users/", json=new_user.model_dump())
+    assert response.status_code == 400
+
+def test_name_not_unique_returns_400(session: Session, client: TestClient):
+    new_user = NewUser(email="wewqe@2ewew",name="Chiara", subscription=True) # TODO: durch factory ersetzen
+
+    user = User().factory(name="Chiara")
+    session.add(user)
+    session.commit()
+    response = client.post(url="/users/", json=new_user.model_dump())
+    assert response.status_code == 400
+
+def test_login_successful(session: Session, client: TestClient):
+    user = User().factory()
+    session.add(user)
+    session.commit()
+    session.refresh(user)
+
+    payload = {
+        "user_code": user.user_code
+    }
+
+    response = client.post(
+        url="/users/login", 
+        data=payload
+    )
+
     assert response.status_code == 200
+    assert response.json() == "token"

backend/src/backend/routers/users.py

@@ -1,13 +1,10 @@
-# CreateUser <- mail, Name
+from fastapi import Form
-# Login Rate Limit! -> token for everything else
+from sqlmodel import select
-# Forgot Code <- email (rate limit)
+from starlette.status import HTTP_400_BAD_REQUEST
-# Delete User <- token
+from fastapi import HTTPException
-# change mail <- token, new mail
+from sqlalchemy.exc import IntegrityError
-# subscribe
+from backend.models.user import User
-# unsubscribe
+from backend.models.user import NewUser
-
-# delete all deactivated users after 48h
-
 from backend.internal.database import get_session
 from fastapi import Depends
 from sqlmodel import Session
@@ -20,6 +17,32 @@ router = APIRouter(
     responses={404: {"description": "Not found"}},
 )
 
-@router.post("/")
+@router.post("/", status_code=201)
-def register_user(session: Session = Depends(get_session)):
+def register_user(new_user: NewUser, session: Session = Depends(get_session)):
-    pass
+    user = User.model_validate(new_user)
+    
+    try:
+        session.add(user)
+        session.commit()
+        # TODO: send mail
+    except IntegrityError as e:
+        session.rollback()
+        raise HTTPException(HTTP_400_BAD_REQUEST)
+
+@router.post("/login") # TODO: rate limit
+def login(user_code: str = Form(), session: Session = Depends(get_session)) -> str:
+    user = session.exec(select(User).where(User.user_code == user_code)).one   
+
+    if not user:
+        raise HTTPException()     
+    return "token"
+
+
+# Login Rate Limit! -> token for everything else
+# Forgot Code <- email (rate limit)
+# Delete User <- token
+# change mail <- token, new mail
+# subscribe
+# unsubscribe
+
+# delete all deactivated users after 48h