✅ Added and updated tests
This commit is contained in:
@@ -26,14 +26,10 @@ jobs:
|
||||
- run: pipenv install --dev
|
||||
- name: Prepare tests
|
||||
run: |
|
||||
echo ${{ secrets.DOTENV }} | base64 --decode > .env
|
||||
set -o allexport; source .env; set +o allexport
|
||||
mkdir /tmp/storage/
|
||||
pipenv run python init_db.py
|
||||
docker run --rm -d -p 6379:6379 --name test_redis redis:alpine
|
||||
chmod +x run_tests.sh
|
||||
- name: Run tests
|
||||
run: |
|
||||
pipenv run coverage run -m pytest --asyncio-mode=strict classquiz/tests
|
||||
./run_tests.sh
|
||||
pipenv run coverage xml
|
||||
- name: Report results to DeepSource
|
||||
run: |
|
||||
|
||||
@@ -32,6 +32,8 @@ pytest = "*"
|
||||
pytest-asyncio = "*"
|
||||
flake8 = "*"
|
||||
black = "*"
|
||||
pytest-dependency = "*"
|
||||
pytest-order = "*"
|
||||
|
||||
[requires]
|
||||
python_version = "3.10"
|
||||
@@ -39,4 +41,4 @@ python_version = "3.10"
|
||||
[scripts]
|
||||
format = "black ."
|
||||
lint = "flake8 classquiz"
|
||||
test = "coverage run -m pytest --lf -v --asyncio-mode=strict classquiz/tests"
|
||||
test = "coverage run -m pytest --lf -v --asyncio-mode=strict classquiz/tests"
|
||||
|
||||
Generated
+17
-2
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"_meta": {
|
||||
"hash": {
|
||||
"sha256": "79e50387bf674087fade99e99dd8d2765d5f9153f4f1e2f56d8cfc528d664cf3"
|
||||
"sha256": "3f7ac0662ef409fd47b457bb2c89cea9f25cf779fa9cbe64d16c6ff51b854862"
|
||||
},
|
||||
"pipfile-spec": 6,
|
||||
"requires": {
|
||||
@@ -1508,12 +1508,27 @@
|
||||
"index": "pypi",
|
||||
"version": "==0.18.3"
|
||||
},
|
||||
"pytest-dependency": {
|
||||
"hashes": [
|
||||
"sha256:c2a892906192663f85030a6ab91304e508e546cddfe557d692d61ec57a1d946b"
|
||||
],
|
||||
"index": "pypi",
|
||||
"version": "==0.5.1"
|
||||
},
|
||||
"pytest-order": {
|
||||
"hashes": [
|
||||
"sha256:5dd6b929fbd7eaa6d0ee07586f65c623babb0afe72b4843c5f15055d6b3b1b1f",
|
||||
"sha256:bbe6e63a8e23741ab3e810d458d1ea7317e797b70f9550512d77d6e9e8fd1bbb"
|
||||
],
|
||||
"index": "pypi",
|
||||
"version": "==1.0.1"
|
||||
},
|
||||
"tomli": {
|
||||
"hashes": [
|
||||
"sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc",
|
||||
"sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"
|
||||
],
|
||||
"markers": "python_version >= '3.7'",
|
||||
"markers": "python_version < '3.11'",
|
||||
"version": "==2.0.1"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
import pytest
|
||||
from typing import Generator
|
||||
from fastapi.testclient import TestClient
|
||||
from classquiz import app
|
||||
|
||||
|
||||
class ValueStorage:
|
||||
quiz_id = None
|
||||
value2 = None
|
||||
|
||||
|
||||
example_quiz = {
|
||||
"public": False,
|
||||
"title": "Some test question",
|
||||
"description": "A description",
|
||||
"questions": [
|
||||
{
|
||||
"question": "Is ClassQuiz cool?",
|
||||
"time": 10,
|
||||
"answers": [{"right": True, "answer": "Yes"}, {"right": False, "answer": "No"}],
|
||||
},
|
||||
{
|
||||
"question": "Do you like open source?",
|
||||
"time": 5,
|
||||
"image": "https://i.imgur.com/sSNSy77.png",
|
||||
"answers": [
|
||||
{"right": True, "answer": "Yes"},
|
||||
{"right": False, "answer": "No"},
|
||||
{"right": False, "answer": "Maybe"},
|
||||
],
|
||||
},
|
||||
],
|
||||
}
|
||||
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
|
||||
|
||||
@@ -3,6 +3,7 @@ import pytest
|
||||
from classquiz.kahoot_importer.get import get
|
||||
|
||||
|
||||
@pytest.mark.order("last")
|
||||
@pytest.mark.asyncio
|
||||
async def test_get():
|
||||
await get("1f95eb0b-fcf4-4db2-879b-5418ef75116b")
|
||||
|
||||
@@ -4,6 +4,7 @@ from classquiz.kahoot_importer.search import search
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.order("last")
|
||||
async def test_search():
|
||||
res = await search(query="Python", limit=100)
|
||||
assert len(res.entities) == 100
|
||||
|
||||
@@ -0,0 +1,317 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from redis import Redis
|
||||
from classquiz.db import database, models
|
||||
from classquiz.config import settings
|
||||
from classquiz.tests import test_user_email, test_user_password
|
||||
from classquiz.tests import test_client, example_quiz, ValueStorage
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def start_db():
|
||||
if not database.is_connected:
|
||||
database.connect()
|
||||
|
||||
|
||||
class TestUsers:
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_test_user(self, 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(self, 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(self, 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(self, 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(self, 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(self, 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(self, 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(self, 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(self, 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(self, 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
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forgotten_password(self, test_client):
|
||||
resp = test_client.post("/api/v1/users/forgot-password", json={"email": test_user_email})
|
||||
assert resp.status_code == 200
|
||||
resp = test_client.post("/api/v1/users/forgot-password", json={"email": "ddassad@dsa.ads"})
|
||||
assert resp.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reset_password_with_token(self, 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"]
|
||||
me = test_client.get("/api/v1/users/me", cookies={"access_token": token}).json()
|
||||
redis = Redis().from_url(settings().redis)
|
||||
redis.set("reset_passwd:_1token_", str(me["id"]))
|
||||
redis.set("reset_passwd:_2token_", str(uuid.uuid4()))
|
||||
# test wtith wrong token
|
||||
resp = test_client.post(
|
||||
"/api/v1/users/reset-password", json={"token": "doesnt_exist", "password": "new_password"}
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
resp = test_client.post("/api/v1/users/reset-password", json={"token": "_2token_", "password": "new_password"})
|
||||
assert resp.status_code == 400
|
||||
resp = test_client.post("/api/v1/users/reset-password", json={"token": "_1token_", "password": "new_password"})
|
||||
assert resp.status_code == 200
|
||||
resp = test_client.post(
|
||||
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": "new_password"}
|
||||
)
|
||||
token = resp.cookies["access_token"]
|
||||
test_client.put(
|
||||
"/api/v1/users/password/update",
|
||||
json={"new_password": test_user_password, "old_password": "new_password"},
|
||||
cookies={"access_token": token},
|
||||
)
|
||||
redis.flushdb()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_signout_everywhere(self, 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.delete("/api/v1/users/signout-everywhere", cookies={"access_token": token})
|
||||
assert resp.status_code == 200
|
||||
|
||||
|
||||
class TestUtils:
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_ip_data(self, 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/utils/ip-lookup/1.1.1.1", cookies={"access_token": token})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["query"] == "1.1.1.1"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_qr(self, test_client):
|
||||
resp = test_client.get("/api/v1/utils/qr/12345678")
|
||||
assert resp.status_code == 200
|
||||
assert resp.headers["Content-Type"] == "image/svg+xml"
|
||||
|
||||
|
||||
class TestStats:
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_quiz_count(self, test_client):
|
||||
redis = Redis().from_url(settings().redis)
|
||||
redis.flushdb()
|
||||
for i in range(2):
|
||||
resp = test_client.get("/api/v1/stats/quizzes")
|
||||
assert resp.status_code == 200
|
||||
assert resp.text == str(0)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_count(self, test_client):
|
||||
redis = Redis().from_url(settings().redis)
|
||||
redis.flushdb()
|
||||
for i in range(2):
|
||||
resp = test_client.get("/api/v1/stats/users")
|
||||
assert resp.status_code == 200
|
||||
assert resp.text == str(1)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_combined_count(self, test_client):
|
||||
redis = Redis().from_url(settings().redis)
|
||||
redis.flushdb()
|
||||
for i in range(2):
|
||||
resp = test_client.get("/api/v1/stats/combined")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["quiz_count"] == 0
|
||||
assert resp.json()["user_count"] == 1
|
||||
|
||||
|
||||
class TestQuiz:
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_quiz(self, 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.post("/api/v1/quiz/create", json=example_quiz, cookies={"access_token": token})
|
||||
assert resp.status_code == 200
|
||||
ValueStorage.quiz_id = resp.json()["id"]
|
||||
example_quiz["questions"][1]["image"] = "https://imgur.com/sSNSy77.png"
|
||||
resp = test_client.post("/api/v1/quiz/create", json=example_quiz, cookies={"access_token": token})
|
||||
assert resp.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_quiz_from_id(self, 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(f"/api/v1/quiz/get/{ValueStorage.quiz_id}", cookies={"access_token": token})
|
||||
assert resp.status_code == 200
|
||||
resp = test_client.get(f"/api/v1/quiz/get/dasdsadsadsadsadsa", cookies={"access_token": token})
|
||||
assert resp.status_code == 400
|
||||
resp = test_client.get(
|
||||
f"/api/v1/quiz/get/847c64d3-39f9-4bb7-8f13-fae913f67858", cookies={"access_token": token}
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_quizzes(self, 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/quiz/list", cookies={"access_token": token})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()[0]["id"] == ValueStorage.quiz_id
|
||||
@@ -1,240 +0,0 @@
|
||||
import uuid
|
||||
from typing import Generator
|
||||
|
||||
import pytest
|
||||
from redis import Redis
|
||||
from fastapi.testclient import TestClient
|
||||
from classquiz.db import database, models
|
||||
from classquiz.config import settings
|
||||
|
||||
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
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reset_password_with_token(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"]
|
||||
me = test_client.get("/api/v1/users/me", cookies={"access_token": token}).json()
|
||||
redis = Redis().from_url(settings().redis)
|
||||
redis.set("reset_passwd:_1token_", str(me["id"]))
|
||||
redis.set("reset_passwd:_2token_", str(uuid.uuid4()))
|
||||
# test wtith wrong token
|
||||
resp = test_client.post("/api/v1/users/reset-password", json={"token": "doesnt_exist", "password": "new_password"})
|
||||
assert resp.status_code == 400
|
||||
resp = test_client.post("/api/v1/users/reset-password", json={"token": "_2token_", "password": "new_password"})
|
||||
assert resp.status_code == 400
|
||||
resp = test_client.post("/api/v1/users/reset-password", json={"token": "_1token_", "password": "new_password"})
|
||||
assert resp.status_code == 200
|
||||
resp = test_client.post(
|
||||
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": "new_password"}
|
||||
)
|
||||
token = resp.cookies["access_token"]
|
||||
test_client.put(
|
||||
"/api/v1/users/password/update",
|
||||
json={"new_password": test_user_password, "old_password": "new_password"},
|
||||
cookies={"access_token": f"Bearer {token}"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forgotten_password(test_client):
|
||||
resp = test_client.post("/api/v1/users/forgot-password", json={"email": test_user_email})
|
||||
assert resp.status_code == 200
|
||||
resp = test_client.post("/api/v1/users/forgot-password", json={"email": "ddassad@dsa.ads"})
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_signout_everywhere(test_client):
|
||||
resp = test_client.delete("/api/v1/users/signout-everywhere")
|
||||
assert resp.status_code == 200
|
||||
+29
-5
@@ -1,5 +1,29 @@
|
||||
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 classquiz/tests/test_users.py
|
||||
rm classquiz.db
|
||||
docker container stop test_redis
|
||||
run_tests() {
|
||||
pipenv run coverage run -m pytest -s -v --asyncio-mode=strict classquiz/tests
|
||||
}
|
||||
|
||||
stop() {
|
||||
rm classquiz.db
|
||||
docker container stop test_redis
|
||||
docker container stop test_meili
|
||||
}
|
||||
|
||||
init() {
|
||||
docker run --rm -d -p 6379:6379 --name test_redis redis:alpine
|
||||
docker run -it --rm -d -p 7700:7700 --name test_meili getmeili/meilisearch:latest
|
||||
pipenv run python3 init_db.py
|
||||
}
|
||||
|
||||
case $1 in
|
||||
+) init ;;
|
||||
-) stop ;;
|
||||
a)
|
||||
init
|
||||
run_tests
|
||||
stop
|
||||
;;
|
||||
*)
|
||||
echo "Invalid option: -$OPTARG" >&2
|
||||
exit 1
|
||||
;;
|
||||
esac
|
||||
|
||||
Reference in New Issue
Block a user