✅ Added tests for users

.github/workflows/pytest.yml

@@ -3,9 +3,7 @@ name: PyTest
 on:
   push:
     paths:
-      - "classquiz/tests/**"
+      - "classquiz/**"
-      - "classquiz/kahoot_importer/**"
-      - "classquiz/storage/**"
   workflow_dispatch:
 
 jobs:

Pipfile

@@ -39,3 +39,4 @@ python_version = "3.10"
 [scripts]
 format = "black ."
 lint = "flake8 classquiz"
+test = "coverage run -m pytest --lf -v --asyncio-mode=strict classquiz/tests"

classquiz/routers/users.py

@@ -26,13 +26,7 @@ settings = settings()
 router = APIRouter()
 
 route_user = User.get_pydantic(
-    exclude={
+    exclude={"id": ..., "verified": ..., "verify_key": ..., "created_at": ..., "usersessions": ..., "quizs": ...}
-        "id": ...,
-        "verified": ...,
-        "verify_key": ...,
-        "created_at": ...,
-        "usersessions": ...,
-    }
 )
 
 
@@ -242,7 +236,7 @@ async def reset_password_with_token(reset_password: ResetPassword, response: Res
     return {"message": "Password updated successfully"}
 
 
-@router.get("/sessions/list", response_model=list[UserSession], response_model_exclude={"user", "session_key"})
+@router.get("/sessions/list", response_model=list[UserSession], response_model_exclude={"user", "session_key", "quizs"})
 async def list_sessions(user: User = Depends(get_current_user)):
     sessions = await UserSession.objects.filter(user=user).all()
     return [session.dict() for session in sessions]
@@ -258,7 +252,9 @@ async def delete_session(session_id: str, user: User = Depends(get_current_user)
     return {"message": "Session deleted"}
 
 
-@router.get("/session", response_model=UserSession, response_model_exclude={"user", "session_key"})
+@router.get(
+    "/session", response_model=UserSession, response_model_exclude={"user": ..., "session_key": ..., "quizs": ...}
+)
 async def get_session(user: User = Depends(get_current_user)):
     session = await UserSession.objects.filter(user=user).first()
     return session

classquiz/tests/test_auth.py

@@ -0,0 +1,23 @@
+import pytest
+from classquiz.auth import get_password_hash, verify_password, settings, ALGORITHM, create_access_token
+from jose import JWTError, jwt
+
+test_passwords = ["password", "password123", "12345678", "saddsaasdsad", "dsadasasdasddasasdsdasad"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("password", test_passwords)
+async def test_password_hashes(password):
+    passwd_hash = get_password_hash(password)
+    assert verify_password(password, passwd_hash)
+
+
+@pytest.mark.asyncio
+async def test_jwt_engine():
+    access_token = create_access_token({"sub": "test@test.com"})
+    assert access_token is not None
+    payload = jwt.decode(access_token, settings.secret_key, algorithms=[ALGORITHM])
+    email: str = payload.get("sub")
+    assert email == "test@test.com"
+    with pytest.raises(JWTError):
+        jwt.decode(access_token, "wrong_secret", algorithms=[ALGORITHM])

classquiz/tests/test_kahoot_import.py

@@ -0,0 +1,16 @@
+import pytest
+
+from classquiz.kahoot_importer.import_quiz import _download_image
+
+ddg_robots_txt = b"""a"""
+test_url = (
+    "https://gist.githubusercontent.com/mawoka-myblock/b43f0d888a9e6a25806b3c73e63b658f/raw"
+    "/134f135f99f8f385695304f739667c70b636386a/test-gist"
+)
+
+
+@pytest.mark.asyncio
+async def test_download_image():
+    image = await _download_image(test_url)
+    print(image)
+    assert image == ddg_robots_txt

classquiz/tests/test_kahoot_search.py

@@ -18,5 +18,4 @@ async def test_search():
     await search(query="Internet", limit=100)
     await search(query="Windows", limit=100)
     res2 = await search(query="Python", limit=100)
-    res2_ids = [str(e.card.uuid) for e in res2.entities]
+    [str(e.card.uuid) for e in res2.entities]
-    print(res2_ids)

classquiz/tests/test_users.py

@@ -0,0 +1,196 @@
+from typing import Generator
+
+import pytest
+from fastapi.testclient import TestClient
+from classquiz.db import database, models
+
+
+from classquiz import app
+
+test_user_email = "sth@byom.de"
+test_user_password = "test"
+
+
+@pytest.fixture(scope="module")
+def test_client() -> Generator:
+    with TestClient(app) as testclient:
+        yield testclient
+
+
+def start_db():
+    if not database.is_connected:
+        database.connect()
+
+
+@pytest.mark.asyncio
+async def test_create_test_user(test_client):
+    resp = test_client.post(
+        "/api/v1/users/create", json={"email": test_user_email, "password": test_user_password, "username": "mawoka"}
+    )
+    assert resp.status_code == 200
+    assert resp.json()["email"] == test_user_email
+    resp = test_client.post(
+        "/api/v1/users/create", json={"email": test_user_email, "password": test_user_password, "username": "mawoka"}
+    )
+    assert resp.status_code == 409
+    resp = test_client.post(
+        "/api/v1/users/create",
+        json={"email": "doesntexist@hidsadawadsdaads.ghsxd", "password": test_user_password, "username": "dieter"},
+    )
+    assert resp.status_code == 400
+
+    resp = test_client.post(
+        "/api/v1/users/create",
+        json={
+            "email": "doesntexist@hidsadawadsdaads.ghsxd",
+            "password": test_user_password,
+            "username": "12345678978978978978945632145678",
+        },
+    )
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_verify_email(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    assert resp.status_code == 401
+    user = await models.User.objects.filter(email=test_user_email).get()
+    assert (test_client.get("/api/v1/users/verify/dasadsasdadsasdsaddassad")).status_code == 404
+
+    test_client.get(f"/api/v1/users/verify/{user.verify_key}")
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    assert resp.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_check(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    token = resp.json()["access_token"]
+    resp = test_client.get("/api/v1/users/check", cookies={"access_token": f"Bearer {token}"})
+    assert resp.status_code == 200
+    resp = test_client.get("/api/v1/users/check", cookies={"access_token": "Bearer dasasdasddasadsasdadssadsd"})
+    assert resp.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_me(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    token = resp.json()["access_token"]
+    resp = test_client.get("/api/v1/users/me", cookies={"access_token": f"Bearer {token}"})
+    data = resp.json()
+    assert resp.status_code == 200
+    assert data["verified"] is True
+    assert data["email"] == test_user_email
+    assert data["username"] == "mawoka"
+
+
+@pytest.mark.asyncio
+async def test_rememberme(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    rememberme_cookie = resp.cookies["rememberme_token"]
+    resp = test_client.get("/api/v1/users/token/rememberme", cookies={"rememberme_token": rememberme_cookie})
+    assert resp.cookies["access_token"] is not None
+    assert resp.status_code == 200
+    resp = test_client.get(
+        "/api/v1/users/token/rememberme", cookies={"rememberme_token": "dsahgvjadsvsahgxddsvhgdsvhg"}
+    )
+    assert resp.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_logout(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    rememberme_cookie = resp.cookies["rememberme_token"]
+    access_token = resp.cookies["access_token"]
+    resp = test_client.get("/api/v1/users/me", cookies={"access_token": access_token})
+    assert resp.status_code == 200
+    resp = test_client.get(
+        "/api/v1/users/logout", cookies={"rememberme_cookie": rememberme_cookie}, allow_redirects=False
+    )
+    assert resp.status_code == 302
+    resp = test_client.get("/api/v1/users/me", cookies={"rememberme_cookie": rememberme_cookie})
+    assert resp.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_password_update(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    token = resp.json()["access_token"]
+    resp = test_client.put(
+        "/api/v1/users/password/update",
+        json={"new_password": "new_password", "old_password": test_user_password},
+        cookies={"access_token": f"Bearer {token}"},
+    )
+    assert resp.status_code == 200
+    resp = test_client.put(
+        "/api/v1/users/password/update",
+        json={"new_password": "asdsdadsasdaasd", "old_password": "asdasdsadadsasdsadasdasd"},
+        cookies={"access_token": f"Bearer {token}"},
+    )
+    assert resp.status_code == 400
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": "new_password"}
+    )
+    assert resp.status_code == 200
+    resp = test_client.put(
+        "/api/v1/users/password/update",
+        json={"new_password": test_user_password, "old_password": "new_password"},
+        cookies={"access_token": f"Bearer {token}"},
+    )
+    assert resp.status_code == 200
+    resp1 = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    rememberme_cookie = resp1.cookies["rememberme_token"]
+    response = test_client.get("/api/v1/users/me", cookies={"rememberme_cookie": rememberme_cookie})
+    assert response.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_get_session(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    token = resp.cookies["access_token"]
+    resp = test_client.get("/api/v1/users/session", cookies={"access_token": token})
+    assert resp.status_code == 200
+    assert resp.json()["ip_address"] == "testclient"
+
+
+@pytest.mark.asyncio
+async def test_delete_session(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    token = resp.cookies["access_token"]
+    resp = test_client.get("/api/v1/users/session", cookies={"access_token": token})
+    session_id = resp.json()["id"]
+    resp = test_client.delete("/api/v1/users/sessions/" + str(session_id), cookies={"access_token": token})
+    assert resp.status_code == 200
+    resp = test_client.delete("/api/v1/users/sessions/asdsadasdasdsad", cookies={"access_token": token})
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_list_sessions(test_client):
+    resp = test_client.post(
+        "/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
+    )
+    token = resp.cookies["access_token"]
+    resp = test_client.get("/api/v1/users/sessions/list", cookies={"access_token": token})
+    assert resp.status_code == 200
+    assert len(resp.json()) >= 1

run_tests.sh

@@ -0,0 +1,5 @@
+docker run --rm -d -p 6379:6379 --name test_redis redis:alpine
+python3 init_db.py
+coverage run -m pytest -s -v --asyncio-mode=strict --cache-clear classquiz/tests
+rm classquiz.db
+docker container stop test_redis