Compare commits

..

No commits in common. "418e2a22c92135cb3c615fdfe059004a886ebe50" and "51d7a468cf6c72a38081f7180280165bda016e6c" have entirely different histories.

27 changed files with 35 additions and 229 deletions

3
.gitignore vendored
View file

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

View file

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

BIN
backend/dev.db Normal file

Binary file not shown.

View file

@ -1,35 +0,0 @@
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

View file

@ -2,7 +2,6 @@ from collections.abc import Generator
from sqlmodel import Session from sqlmodel import Session
from sqlmodel import SQLModel from sqlmodel import SQLModel
from sqlmodel import create_engine from sqlmodel import create_engine
sqlite_file = "dev.db" sqlite_file = "dev.db"
sqlite_url = f"sqlite:///{sqlite_file}" sqlite_url = f"sqlite:///{sqlite_file}"
@ -14,4 +13,4 @@ def create_db_and_tables():
def get_session() -> Generator[Session, None, None]: def get_session() -> Generator[Session, None, None]:
with Session(engine) as session: with Session(engine) as session:
yield session yield session

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -1,24 +1,4 @@
# add comment # add comment
# delete comment # delete comment
# get comments # 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()

View file

@ -1,84 +1,15 @@
from backend.models.user import NewUser from httpx2 import post
from sqlmodel import select from backend.models.user import RegisterUser
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 backend.main import app
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@pytest.fixture(name="session") client = TestClient(app)
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 = { new_user = {
"email": "Test@Test.test", "email": "Test",
"name": "Test1", "name": "Test1",
"subscription": True "subscription": True
} }
response = client.post(url="/users/", json=new_user) response = client.post(url="/users/", json=new_user)
assert response.status_code == 201 assert response.status_code == 200
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"

View file

@ -1,10 +1,13 @@
from fastapi import Form # CreateUser <- mail, Name
from sqlmodel import select # Login Rate Limit! -> token for everything else
from starlette.status import HTTP_400_BAD_REQUEST # Forgot Code <- email (rate limit)
from fastapi import HTTPException # Delete User <- token
from sqlalchemy.exc import IntegrityError # change mail <- token, new mail
from backend.models.user import User # subscribe
from backend.models.user import NewUser # unsubscribe
# delete all deactivated users after 48h
from backend.internal.database import get_session from backend.internal.database import get_session
from fastapi import Depends from fastapi import Depends
from sqlmodel import Session from sqlmodel import Session
@ -17,32 +20,6 @@ router = APIRouter(
responses={404: {"description": "Not found"}}, responses={404: {"description": "Not found"}},
) )
@router.post("/", status_code=201) @router.post("/")
def register_user(new_user: NewUser, session: Session = Depends(get_session)): def register_user(session: Session = Depends(get_session)):
user = User.model_validate(new_user) pass
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