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
backend/venv/lib/**
backend/venv/lib64/*
backend/dev.db
backend/venv/lib64/*

View file

@ -2,10 +2,5 @@
"cSpell.language": "en,de-DE",
"cSpell.words": [
"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 SQLModel
from sqlmodel import create_engine
sqlite_file = "dev.db"
sqlite_url = f"sqlite:///{sqlite_file}"
@ -14,4 +13,4 @@ def create_db_and_tables():
def get_session() -> Generator[Session, None, None]:
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, comments
from .routers import users
from backend.internal.database import create_db_and_tables
from fastapi import APIRouter
from fastapi import FastAPI
@ -17,4 +16,3 @@ def on_startup():
router = APIRouter()
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
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Optional
from sqlmodel import Relationship
from sqlmodel import DateTime
from sqlmodel import Field
from sqlmodel import SQLModel
if TYPE_CHECKING:
from backend.models.user import User
from ..models.user import User
class CommentBase(SQLModel):
pass
@dataclass
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)
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: "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
from typing import TYPE_CHECKING
from sqlmodel import Field
@ -8,46 +5,20 @@ from sqlmodel import Relationship
from sqlmodel import SQLModel
if TYPE_CHECKING:
from backend.models.comment import Comment
from ..models.comment import Comment
class UserBase(SQLModel):
pass
class User(UserBase, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid7, primary_key=True)
class User(UserBase):
id: uuid.UUID | None = Field(default_factory=uuid.uuid4, primary_key=True)
email: 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()
created_at: datetime = Field(default_factory=datetime.now)
comments: list["Comment"] = Relationship(back_populates="user")
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):
class RegisterUser(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

View file

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

View file

@ -1,84 +1,15 @@
from backend.models.user import NewUser
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 httpx2 import post
from backend.models.user import RegisterUser
from backend.main import app
from fastapi.testclient import TestClient
@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
client = TestClient(app)
@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):
def test_register_user_works():
new_user = {
"email": "Test@Test.test",
"email": "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"
assert response.status_code == 200

View file

@ -1,10 +1,13 @@
from fastapi import Form
from sqlmodel import select
from starlette.status import HTTP_400_BAD_REQUEST
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from backend.models.user import User
from backend.models.user import NewUser
# CreateUser <- mail, Name
# 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
from backend.internal.database import get_session
from fastapi import Depends
from sqlmodel import Session
@ -17,32 +20,6 @@ router = APIRouter(
responses={404: {"description": "Not found"}},
)
@router.post("/", status_code=201)
def register_user(new_user: NewUser, session: Session = Depends(get_session)):
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
@router.post("/")
def register_user(session: Session = Depends(get_session)):
pass