LobotomyLabs/classquiz-ai
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🔀 Merged ClassQuizController

Browse Source
This commit is contained in:
Mawoka
2023-06-30 01:36:48 +02:00
parent b6dffb7a26 394b8b1a8c
commit 62538406df
183 changed files with 10232 additions and 3260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
classquiz/__init__.py
+6 -12
View File
@@ -1,7 +1,6 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import asyncio
import sentry_sdk
from fastapi import FastAPI, Request
@@ -11,7 +10,6 @@ from starlette.middleware.sessions import SessionMiddleware
from classquiz.config import settings
from classquiz.db import database
from datetime import timedelta
from classquiz.oauth import rememberme_middleware
from classquiz.routers import (
@@ -33,10 +31,11 @@ from classquiz.routers import (
results,
admin,
box_controller,
quiztivity,
pixabay,
)
from classquiz.socket_server import sio
from classquiz.helpers import meilisearch_init, telemetry_ping, bg_tasks
from scheduler.asyncio import Scheduler
from classquiz.helpers import meilisearch_init, telemetry_ping
settings = settings()
if settings.sentry_dsn:
@@ -57,13 +56,6 @@ async def sentry_exception(request: Request, call_next):
raise e
async def background_tasks():
schedule = Scheduler()
schedule.cyclic(timedelta(hours=6), bg_tasks.clean_editor_images_up)
while True:
await asyncio.sleep(1)
@app.on_event("startup")
async def startup() -> None:
database_ = app.state.database
@@ -71,7 +63,6 @@ async def startup() -> None:
await database_.connect()
await meilisearch_init()
await telemetry_ping()
asyncio.create_task(background_tasks())
@app.on_event("shutdown")
@@ -86,6 +77,9 @@ async def auth_middleware_wrapper(request: Request, call_next):
return await rememberme_middleware(request, call_next)
app.include_router(pixabay.router, tags=["pixabay"], prefix="/api/v1/pixabay", include_in_schema=True)
app.include_router(quiztivity.router, tags=["quiztivity"], prefix="/api/v1/quiztivity", include_in_schema=True)
app.include_router(
box_controller.router, tags=["boxcontroller"], prefix="/api/v1/box-controller", include_in_schema=True
)
classquiz/cache.py
-1
View File
@@ -33,7 +33,6 @@ async def cache_account(criteria: str, content: str) -> Union[User, None]:
await insert_into_redis(res, content)
return res
elif criteria == "id":
try:
res = await User.objects.get(id=uuid.UUID(content), verified=True)
except ormar.exceptions.NoMatch:
classquiz/config.py
+25 -7
View File
@@ -9,6 +9,8 @@ import redis as redis_base_lib
from pydantic import BaseSettings, RedisDsn, PostgresDsn, BaseModel
import meilisearch as MeiliSearch
from typing import Optional
from arq import create_pool
from arq.connections import RedisSettings, ArqRedis
from classquiz.storage import Storage
@@ -48,33 +50,49 @@ class Settings(BaseSettings):
github_client_secret: Optional[str]
custom_openid_provider: CustomOpenIDProvider | None = None
telemetry_enabled: bool = True
free_storage_limit: int = 1074000000
pixabay_api_key: str | None = None
# storage_backend
storage_backend: str | None = "deta"
# if storage_backend == "deta":
deta_project_key: str | None
deta_project_id: str | None
storage_backend: str | None = "local"
# if storage_backend == "local":
storage_path: str | None
# if storage_backend == "s3":
s3_access_key: str | None
s3_secret_key: str | None
s3_bucket_name: str = "classquiz"
s3_base_url: str | None
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
env_nested_delimiter = "__"
async def initialize_arq():
# skipcq: PYL-W0603
global arq
arq = await create_pool(RedisSettings.from_dsn(settings.redis))
@lru_cache()
def settings() -> Settings:
return Settings()
redis: redis_base_lib.client.Redis = redis_lib.Redis().from_url(settings().redis)
pool = redis_lib.ConnectionPool().from_url(settings().redis)
redis: redis_base_lib.client.Redis = redis_lib.Redis(connection_pool=pool)
arq: ArqRedis = ArqRedis(pool_or_conn=pool)
storage: Storage = Storage(
backend=settings().storage_backend,
deta_key=settings().deta_project_key,
deta_id=settings().deta_project_id,
storage_path=settings().storage_path,
access_key=settings().s3_access_key,
secret_key=settings().s3_secret_key,
bucket_name=settings().s3_bucket_name,
base_url=settings().s3_base_url,
)
meilisearch = MeiliSearch.Client(settings().meilisearch_url)
classquiz/db/models.py
+163 -28
View File
@@ -2,16 +2,17 @@
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import os
import re
import uuid
from datetime import datetime
from typing import Optional
import ormar
from ormar import ReferentialAction
from pydantic import BaseModel, Json, validator
from enum import Enum
from . import metadata, database
from ..config import server_regex
from .quiztivity import QuizTivityPage
from sqlalchemy import func
class UserAuthTypes(Enum):
@@ -40,6 +41,7 @@ class User(ormar.Model):
require_password: bool = ormar.Boolean(default=True, nullable=False)
backup_code: str = ormar.String(max_length=64, min_length=64, nullable=False, default=os.urandom(32).hex())
totp_secret: str = ormar.String(max_length=32, min_length=32, nullable=True, default=None)
storage_used: int = ormar.BigInteger(nullable=False, default=0, minimum=0)
class Meta:
tablename = "users"
@@ -55,7 +57,7 @@ class FidoCredentials(ormar.Model):
id: bytes = ormar.LargeBinary(max_length=256)
public_key: bytes = ormar.LargeBinary(max_length=256)
sign_count: int = ormar.Integer()
user: Optional[User] = ormar.ForeignKey(User)
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
class Meta:
tablename = "fido_credentials"
@@ -65,7 +67,7 @@ class FidoCredentials(ormar.Model):
class ApiKey(ormar.Model):
key: str = ormar.String(max_length=48, min_length=48, primary_key=True)
user: Optional[User] = ormar.ForeignKey(User)
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
class Meta:
tablename = "api_keys"
@@ -79,7 +81,7 @@ class UserSession(ormar.Model):
"""
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4())
user: Optional[User] = ormar.ForeignKey(User)
user: Optional[User] = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
session_key: str = ormar.String(unique=True, max_length=64)
created_at: datetime = ormar.DateTime(default=datetime.now())
ip_address: str = ormar.String(max_length=100, nullable=True)
@@ -118,6 +120,7 @@ class QuizQuestionType(str, Enum):
SLIDE = "SLIDE"
TEXT = "TEXT"
ORDER = "ORDER"
CHECK = "CHECK"
class TextQuizAnswer(BaseModel):
@@ -134,7 +137,6 @@ class QuizQuestion(BaseModel):
@validator("answers")
def answers_not_none_if_abcd_type(cls, v, values):
# print(values)
if values["type"] == QuizQuestionType.ABCD and type(v[0]) != ABCDQuizAnswer:
raise ValueError("Answers can't be none if type is ABCD")
if values["type"] == QuizQuestionType.RANGE and type(v) != RangeQuizAnswer:
@@ -147,6 +149,8 @@ class QuizQuestion(BaseModel):
raise ValueError("Answer must be from type VotingQuizAnswer if type is ORDER")
if values["type"] == QuizQuestionType.SLIDE and type(v[0]) != str:
raise ValueError("Answer must be from type SlideElement if type is SLIDE")
if values["type"] == QuizQuestionType.CHECK and type(v[0]) != ABCDQuizAnswer:
raise ValueError("Answers can't be none if type is CHECK")
return v
@@ -159,15 +163,6 @@ class QuizInput(BaseModel):
questions: list[QuizQuestion]
background_image: str | None
@validator("background_image")
def must_come_from_local_cdn(cls, v):
if v is None:
return v
elif bool(re.match(server_regex, v)):
return v
else:
raise ValueError("does not match url scheme")
class Quiz(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4(), nullable=False, unique=True)
@@ -176,7 +171,7 @@ class Quiz(ormar.Model):
description: str = ormar.Text(nullable=True)
created_at: datetime = ormar.DateTime(default=datetime.now())
updated_at: datetime = ormar.DateTime(default=datetime.now())
user_id: uuid.UUID = ormar.ForeignKey(User)
user_id: uuid.UUID = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
questions: Json[list[QuizQuestion]] = ormar.JSON(nullable=False)
imported_from_kahoot: Optional[bool] = ormar.Boolean(default=False, nullable=True)
cover_image: Optional[str] = ormar.Text(nullable=True, unique=False)
@@ -184,15 +179,6 @@ class Quiz(ormar.Model):
background_image: str | None = ormar.Text(nullable=True, unique=False)
kahoot_id: uuid.UUID | None = ormar.UUID(nullable=True, default=None)
@validator("background_image")
def must_come_from_local_cdn(cls, v):
if v is None:
return v
elif bool(re.match(server_regex, v)):
return v
else:
raise ValueError("does not match url scheme")
class Meta:
tablename = "quiz"
metadata = metadata
@@ -291,7 +277,7 @@ class GameInLobby(BaseModel):
game_id: uuid.UUID
#
# skipcq: PY-W0069
# class UserProfileLinks(ormar.Model):
# id: int = ormar.Integer(primary_key=True, autoincrement=True)
# user: Optional[User] = ormar.ForeignKey(User)
@@ -302,8 +288,8 @@ class GameInLobby(BaseModel):
class GameResults(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True)
quiz: uuid.UUID | Quiz = ormar.ForeignKey(Quiz)
user: uuid.UUID | User = ormar.ForeignKey(User)
quiz: uuid.UUID | Quiz = ormar.ForeignKey(Quiz, ondelete=ReferentialAction.CASCADE)
user: uuid.UUID | User = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
timestamp: datetime = ormar.DateTime(default=datetime.now(), nullable=False)
player_count: int = ormar.Integer(nullable=False, default=0)
note: str | None = ormar.Text(nullable=True)
@@ -320,6 +306,155 @@ class GameResults(ormar.Model):
database = database
class QuizTivityInput(BaseModel):
title: str
pages: list[QuizTivityPage]
class QuizTivity(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True)
title: str = ormar.Text(nullable=False)
created_at: datetime = ormar.DateTime(nullable=False, server_default=func.now())
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
pages: list[QuizTivityPage] = ormar.JSON(nullable=False)
class Meta:
tablename = "quiztivitys"
metadata = metadata
database = database
class QuizTivityShare(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True)
name: str | None = ormar.Text(nullable=True)
expire_at: datetime | None = ormar.DateTime(nullable=True)
quiztivity: QuizTivity | None = ormar.ForeignKey(QuizTivity, ondelete=ReferentialAction.CASCADE)
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.CASCADE)
class Meta:
tablename = "quiztivityshares"
metadata = metadata
database = database
class OnlyId(BaseModel):
id: uuid.UUID
class PublicQuizTivityShare(BaseModel):
id: uuid.UUID
name: str | None
expire_in: int | None
quiztivity: OnlyId
user: OnlyId
@classmethod
def from_db_model(cls, data: QuizTivityShare):
expire_in = None
if data.expire_at is not None:
expire_in = int((data.expire_at - datetime.now()).seconds / 60)
return cls(
id=data.id,
name=data.name,
expire_in=expire_in,
quiztivity=OnlyId(id=data.quiztivity.id),
user=OnlyId(id=data.user.id),
)
class StorageItem(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True)
uploaded_at: datetime = ormar.DateTime(nullable=False, default=datetime.now())
mime_type: str = ormar.Text(nullable=False)
hash: bytes | None = ormar.LargeBinary(nullable=True, min_length=16, max_length=16)
user: User | None = ormar.ForeignKey(User, ondelete=ReferentialAction.SET_NULL)
size: int = ormar.BigInteger(nullable=False)
storage_path: str | None = ormar.Text(nullable=True)
deleted_at: datetime | None = ormar.DateTime(nullable=True, default=None)
quiztivities: list[QuizTivity] | None = ormar.ManyToMany(QuizTivity)
quizzes: list[Quiz] | None = ormar.ManyToMany(Quiz)
alt_text: str | None = ormar.Text(default=None, nullable=True)
filename: str | None = ormar.Text(default=None, nullable=True)
thumbhash: str | None = ormar.Text(default=None, nullable=True)
server: str | None = ormar.Text(default=None, nullable=True)
imported: bool = ormar.Boolean(default=False, nullable=True)
class Meta:
tablename = "storage_items"
metadata = metadata
database = database
class PublicStorageItem(BaseModel):
id: uuid.UUID
uploaded_at: datetime
mime_type: str
hash: str | None
size: int
deleted_at: datetime | None
alt_text: str | None
filename: str | None
thumbhash: str | None
server: str | None
imported: bool
@classmethod
def from_db_model(cls, data: StorageItem):
hash_data = None
if data.hash is not None:
hash_data = data.hash.hex()
return cls(
id=data.id,
uploaded_at=data.uploaded_at,
mime_type=data.mime_type,
hash=hash_data,
size=data.size,
deleted_at=data.deleted_at,
alt_text=data.alt_text,
filename=data.filename,
thumbhash=data.thumbhash,
server=data.server,
imported=data.imported,
)
class PrivateStorageItem(PublicStorageItem):
quizzes: list[OnlyId]
quiztivities: list[OnlyId]
@classmethod
def from_db_model(cls, data: StorageItem):
hash_data = None
if data.hash is not None:
hash_data = data.hash.hex()
quiztivities = []
quizzes = []
for quiz in data.quizzes:
quizzes.append(OnlyId(id=quiz.id))
for quiztivity in data.quiztivities:
quiztivities.append(OnlyId(id=quiztivity.id))
return cls(
id=data.id,
uploaded_at=data.uploaded_at,
mime_type=data.mime_type,
hash=hash_data,
size=data.size,
deleted_at=data.deleted_at,
alt_text=data.alt_text,
filename=data.filename,
quiztivities=quiztivities,
quizzes=quizzes,
thumbhash=data.thumbhash,
server=data.server,
imported=data.imported,
)
class UpdateStorageItem(BaseModel):
filename: str | None
alt_text: str | None
class Controller(ormar.Model):
id: uuid.UUID = ormar.UUID(primary_key=True)
user: uuid.UUID | User = ormar.ForeignKey(User)
classquiz/db/quiztivity.py
+56
View File
@@ -0,0 +1,56 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from pydantic import BaseModel
import enum
class Pdf(BaseModel):
url: str
class _MemoryCard(BaseModel):
image: str | None
text: str | None
id: str
class Memory(BaseModel):
cards: list[list[_MemoryCard]]
class Markdown(BaseModel):
# skipcq: PTC-W0052
markdown: str
class _AbcdAnswer(BaseModel):
answer: str
correct: bool
class Abcd(BaseModel):
question: str
answers: list[_AbcdAnswer]
class QuizTivityTypes(str, enum.Enum):
SLIDE = "SLIDE"
PDF = "PDF"
MEMORY = "MEMORY"
MARKDOWN = "MARKDOWN"
ABCD = "ABCD"
TYPE_CLASS_LIST = {
QuizTivityTypes.PDF: type(Pdf),
QuizTivityTypes.MEMORY: type(Memory),
QuizTivityTypes.MARKDOWN: type(Markdown),
QuizTivityTypes.ABCD: type(Abcd),
}
class QuizTivityPage(BaseModel):
title: str | None
type: QuizTivityTypes
data: Pdf | Memory | Markdown | Abcd
classquiz/helpers/__init__.py
+43 -6
View File
@@ -3,6 +3,7 @@
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import asyncio
import uuid
from typing import Optional
import ormar.exceptions
@@ -64,12 +65,15 @@ async def generate_spreadsheet(quiz_results: dict, quiz: Quiz, player_fields: di
worksheet.write(i + 1, 1, question["time"])
try:
async with ClientSession() as session, session.get(question["image"]) as response:
img_data = BytesIO(await response.read())
worksheet.insert_image(i + 1, 2, question["image"], {"image_data": img_data})
image = Image.open(img_data)
worksheet.set_row(i + 1, image.height)
worksheet.set_column(2, 2, image.width)
async with ClientSession() as session, session.get(
f"{settings.root_address}/api/v1/storage/download/{question['image']}"
) as response:
if "image" in response.headers.get("Content-Type"):
img_data = BytesIO(await response.read())
worksheet.insert_image(i + 1, 2, question["image"], {"image_data": img_data})
image = Image.open(img_data)
worksheet.set_row(i + 1, image.height)
worksheet.set_column(2, 2, image.width)
except TypeError:
pass
answer_amount = len(answer_data)
@@ -175,3 +179,36 @@ def check_hashcash(data: str, input_data: str, claim_in: Optional[str] = "19") -
return False
some_error = [version == "1", claim == claim_in, res == input_data, ext == ""]
return all(el is True for el in some_error)
def check_image_string(image: str) -> (bool, uuid.UUID | None):
# Valid formats: {uuid} and {uuid}--{uuid}
try:
parsed_uuid = uuid.UUID(image)
return True, parsed_uuid
except ValueError:
pass
split_image = image.split("--")
if len(split_image) != 2:
return False, None
try:
uuid.UUID(split_image[0])
uuid.UUID(split_image[1])
return True, None
except ValueError:
return False, None
def extract_image_ids_from_quiz(quiz: Quiz) -> list[str | uuid.UUID]:
quiz_images = []
if quiz.background_image is not None:
quiz_images.append(quiz.background_image)
if quiz.cover_image is not None:
quiz_images.append(quiz.cover_image)
for question in quiz.questions:
if question["image"] is None:
continue
quiz_images.append(question["image"])
return quiz_images
classquiz/helpers/bg_tasks.py
-21
View File
@@ -1,21 +0,0 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from classquiz.config import redis, storage
from classquiz.storage.errors import DeletionFailedError
async def clean_editor_images_up():
print("Cleaning images up")
edit_sessions = await redis.smembers("edit_sessions")
for session_id in edit_sessions:
session = await redis.get(f"edit_session:{session_id}")
if session is None:
images = await redis.lrange(f"edit_session:{session_id}:images", 0, 3000)
if len(images) != 0:
try:
await storage.delete(images)
except DeletionFailedError:
print("Deletion Error", images)
await redis.srem("edit_sessions", session_id)
await redis.delete(f"edit_session:{session_id}:images")
classquiz/helpers/pixabay.py
+136
View File
@@ -0,0 +1,136 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import enum
from aiohttp import ClientSession
from pydantic import BaseModel
class ImageType(str, enum.Enum):
all = "all"
photo = "photo"
illustration = "illustration"
vector = "vector"
class Orientation(str, enum.Enum):
all = ("all",)
horizontal = "horizontal"
vertical = "vertical"
class Category(str, enum.Enum):
background = "background"
fashion = "fashion"
nature = "nature"
science = "science"
education = "education"
feelings = "feelings"
health = "health"
people = "people"
religion = "religion"
places = "places"
animals = "animals"
industry = "industry"
computer = "computer"
food = "food"
sports = "sports"
transportation = "transportation"
travel = "travel"
buildings = "buildings"
business = "business"
music = "music"
class Colors(str, enum.Enum):
grayscale = "grayscale"
transparent = "transparent"
red = "red"
orange = "orange"
yellow = "yellow"
green = "green"
turquoise = "turquoise"
blue = "blue"
lilac = "lilac"
pink = "pink"
white = "white"
gray = "gray"
black = "black"
brown = "brown"
class Order(str, enum.Enum):
popular = "popular"
latest = "latest"
class BoolInput(str, enum.Enum):
true = "true"
false = "false"
class GetImagesParams(BaseModel):
q: str = ""
lang: str = "en"
id: str = ""
image_type: ImageType = ImageType.all
orientation: Orientation = Orientation.all
category: Category | str = ""
min_width: int = 0
min_height: int = 0
colors: Colors | str = ""
editors_choice: BoolInput = BoolInput.false
safesearch: BoolInput = BoolInput.false
order: Order = Order.popular
page: int = 1
pretty: BoolInput = BoolInput.false
class Hit(BaseModel):
id: int
pageURL: str
type: str
tags: str
previewURL: str
previewWidth: int
previewHeight: int
webformatURL: str
webformatWidth: int
webformatHeight: int
largeImageURL: str
imageWidth: int
imageHeight: int
imageSize: int
views: int
downloads: int
collections: int
likes: int
comments: int
user_id: int
user: str
userImageURL: str
class GetImagesResponse(BaseModel):
total: int
totalHits: int
hits: list[Hit]
class NotFoundError(Exception):
pass
async def get_images(api_key: str, params: GetImagesParams) -> GetImagesResponse:
async with ClientSession() as session, session.get(
"https://pixabay.com/api/", params={"key": api_key, **params.dict()}
) as resp:
if resp.status == 200:
return GetImagesResponse.parse_obj(await resp.json())
else:
raise NotFoundError
classquiz/kahoot_importer/import_quiz.py
+44 -13
View File
@@ -3,6 +3,7 @@
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import html
import io
import json
import uuid
from datetime import datetime
@@ -10,8 +11,8 @@ from datetime import datetime
import bleach
from aiohttp import ClientSession
from classquiz.config import settings, storage, meilisearch, ALLOWED_TAGS_FOR_QUIZ
from classquiz.db.models import Quiz, ABCDQuizAnswer, QuizQuestion, User
from classquiz.config import settings, storage, meilisearch, ALLOWED_TAGS_FOR_QUIZ, arq
from classquiz.db.models import Quiz, ABCDQuizAnswer, QuizQuestion, User, StorageItem
from classquiz.kahoot_importer.get import get as get_quiz
from classquiz.helpers import get_meili_data
@@ -23,6 +24,28 @@ async def _download_image(url: str) -> bytes:
return await resp.read()
DEFAULT_COLORS = ["#D6EDC9", "#B07156", "#7F7057", "#4E6E58"]
async def handle_image_upload(url: str, user: User) -> StorageItem:
image_bytes = await _download_image(url)
file_obj = StorageItem(
id=uuid.uuid4(),
uploaded_at=datetime.now(),
mime_type="application/octet-stream",
hash=None,
user=user,
size=0,
deleted_at=None,
alt_text=None,
imported=True,
)
await file_obj.save()
await storage.upload(file_name=file_obj.id.hex, file_data=io.BytesIO(image_bytes))
await arq.enqueue_job("calculate_hash", file_obj.id.hex)
return file_obj
async def import_quiz(quiz_id: str, user: User) -> Quiz | str:
"""
Imports a quiz from Kahoot.
@@ -38,18 +61,24 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | str:
quiz_id = uuid.uuid4()
meilisearch.delete_index(settings.meilisearch_index)
meilisearch.create_index(settings.meilisearch_index)
uploaded_images: list[StorageItem] = []
for q in quiz.kahoot.questions:
answers: list[ABCDQuizAnswer] = []
image = None
if q.image is not None and q.image != "":
image_bytes = await _download_image(q.image)
image_name = f"{quiz_id}--{uuid.uuid4()}"
image = await storage.upload(file_name=image_name, file_data=image_bytes)
image = f"{settings.root_address}/api/v1/storage/download/{image_name}"
for a in q.choices:
image_obj = await handle_image_upload(q.image, user)
uploaded_images.append(image_obj)
image = image_obj.id.hex
for i, a in enumerate(q.choices):
answers.append(
(ABCDQuizAnswer(right=a.correct, answer=html.unescape(bleach.clean(a.answer, tags=[], strip=True))))
(
ABCDQuizAnswer(
right=a.correct,
answer=html.unescape(bleach.clean(a.answer, tags=[], strip=True)),
color=DEFAULT_COLORS[i],
)
)
)
quiz_questions.append(
@@ -62,10 +91,9 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | str:
)
cover = None
if quiz.kahoot.cover != "" and quiz.kahoot.cover is not None:
image_bytes = await _download_image(quiz.kahoot.cover)
image_name = f"{quiz_id}--{uuid.uuid4()}"
await storage.upload(file_name=image_name, file_data=image_bytes)
cover = f"{settings.root_address}/api/v1/storage/download/{image_name}"
img_obj = await handle_image_upload(quiz.kahoot.cover, user)
uploaded_images.append(img_obj)
cover = img_obj.id.hex
quiz_data = Quiz(
id=quiz_id,
public=True,
@@ -80,4 +108,7 @@ async def import_quiz(quiz_id: str, user: User) -> Quiz | str:
kahoot_id=uuid.UUID(kahoot_quiz_id),
)
meilisearch.index(settings.meilisearch_index).add_documents([await get_meili_data(quiz_data)])
return await quiz_data.save()
await quiz_data.save()
for img in uploaded_images:
await quiz_data.storageitems.add(img)
return quiz_data
classquiz/oauth/__init__.py
-11
View File
@@ -24,16 +24,6 @@ async def rememberme_middleware(request: Request, call_next):
rememberme_cookie = request.cookies.get("rememberme_token")
bearer_token = request.cookies.get("access_token")
conditions_to_handle_met = True
# print(bearer_token)
# if bearer_token is not None:
# bearer_token = bearer_token.replace("Bearer ", "")
# test = jws.verify(bearer_token, settings.secret_key, algorithms=["HS256"])
# try:
# jwt.decode(bearer_token, settings.secret_key, algorithms=["HS256"])
# print("jwt ok")
# except JWTError as e:
# print("jwt failed")
# print(test)
scheme, param = get_authorization_scheme_param(bearer_token)
@@ -71,7 +61,6 @@ async def rememberme_middleware(request: Request, call_next):
response: Response = await call_next(request)
return response
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
# access_token_expires = timedelta(seconds=1)
access_token = create_access_token(data={"sub": user_session.user.email}, expires_delta=access_token_expires)
await user_session.update(last_seen=datetime.now())
request.state.access_token = f"Bearer {access_token}"
classquiz/oauth/custom.py
+2 -1
View File
@@ -88,8 +88,9 @@ async def auth(request: Request, response: Response):
google_uid=user_data.sub.hex,
avatar=gzipped_user_avatar(),
)
# skipcq: PYL-W0703
except Exception as e:
if type(e) == asyncpg.exceptions.UniqueViolationError:
if type(e) is asyncpg.exceptions.UniqueViolationError:
error = True
counter = 1
while error:
classquiz/oauth/github.py
+2 -1
View File
@@ -116,8 +116,9 @@ async def auth(request: Request, response: Response):
auth_type=UserAuthTypes.GITHUB,
avatar=gzipped_user_avatar(),
)
# skipcq: PYL-W0703
except Exception as e:
if type(e) == asyncpg.exceptions.UniqueViolationError:
if type(e) is asyncpg.exceptions.UniqueViolationError:
error = True
counter = 1
while error:
classquiz/oauth/google.py
+2 -1
View File
@@ -90,8 +90,9 @@ async def auth(request: Request, response: Response):
google_uid=user_data.sub,
avatar=gzipped_user_avatar(),
)
# skipcq: PYL-W0703
except Exception as e:
if type(e) == asyncpg.exceptions.UniqueViolationError:
if type(e) is asyncpg.exceptions.UniqueViolationError:
error = True
counter = 1
while error:
classquiz/oauth/init_oauth.py
+32 -26
View File
@@ -12,30 +12,36 @@ settings = settings()
@lru_cache()
def init_oauth() -> OAuth:
oauth = OAuth()
oauth.register(
name="google",
server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
client_kwargs={"scope": "openid email profile"},
client_id=settings.google_client_id,
client_secret=settings.google_client_secret,
)
oauth.register(
name="github",
client_kwargs={"scope": "read:user user:email"},
access_token_url="https://github.com/login/oauth/access_token",
access_token_params=None,
authorize_url="https://github.com/login/oauth/authorize",
authorize_params=None,
api_base_url="https://api.github.com/",
client_id=settings.github_client_id,
client_secret=settings.github_client_secret,
)
oauth.register(
name="custom",
client_kwargs={"scope": settings.custom_openid_provider.scopes},
server_metadata_url=settings.custom_openid_provider.server_metadata_url,
client_id=settings.custom_openid_provider.client_id,
client_secret=settings.custom_openid_provider.client_secret,
)
if settings.google_client_secret is not None and settings.google_client_id is not None:
oauth.register(
name="google",
server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
client_kwargs={"scope": "openid email profile"},
client_id=settings.google_client_id,
client_secret=settings.google_client_secret,
)
if settings.github_client_id is not None and settings.github_client_secret is not None:
oauth.register(
name="github",
client_kwargs={"scope": "read:user user:email"},
access_token_url="https://github.com/login/oauth/access_token",
access_token_params=None,
authorize_url="https://github.com/login/oauth/authorize",
authorize_params=None,
api_base_url="https://api.github.com/",
client_id=settings.github_client_id,
client_secret=settings.github_client_secret,
)
if (
settings.custom_openid_provider is not None
and settings.custom_openid_provider.client_id is not None
and settings.custom_openid_provider.client_secret is not None
):
oauth.register(
name="custom",
client_kwargs={"scope": settings.custom_openid_provider.scopes},
server_metadata_url=settings.custom_openid_provider.server_metadata_url,
client_id=settings.custom_openid_provider.client_id,
client_secret=settings.custom_openid_provider.client_secret,
)
return oauth
classquiz/routers/avatar.py
+1
View File
@@ -80,6 +80,7 @@ async def get_customized_avatar(
clothe_color=clothe_color,
clothe_graphic_type=clothe_graphic_type,
).render_svg()
# skipcq: PY-W0069
# print(f"skin_color: {len(AvatarItemsAsList.skin_color)},")
# print(f"hair_color: {len(AvatarItemsAsList.hair_color)},")
# print(f"facial_hair_type: {len(AvatarItemsAsList.facial_hair_type)},")
classquiz/routers/box_controller/embedded/socket.py
+6 -5
View File
@@ -57,12 +57,12 @@ async def submit_answer_fn(data_answer: int, game_pin: str, player_id: str, now:
answers = await redis.get(f"game_session:{game_pin}:{game.current_question}")
answers = await set_answer(answers, game_pin=game_pin, data=answer_data, q_index=game.current_question)
player_count = await redis.scard(f"game_session:{game_pin}:players")
print(player_count, answers)
await sio.emit("player_answer", {})
if answers is not None and len(answers.__root__) == player_count:
await sio.emit("everyone_answered", {})
button_to_index_map = {"y": 0, "r": 2, "g": 1, "b": 3}
button_to_index_map = {"y": 0, "r": 3, "g": 1, "b": 2}
class WebSocketTypes(enum.Enum):
@@ -81,9 +81,9 @@ wss_clients = {}
@router.websocket("/{game_id}")
async def websocket_endpoint(ws: WebSocket, game_id: str):
try:
if game_id in wss_clients.keys():
if game_id in wss_clients:
await ws.close(code=status.WS_1001_GOING_AWAY)
print("Client {} already exists.".format(game_id))
print(f"Client {game_id} already exists.")
return
await ws.accept()
wss_clients[game_id] = ws
@@ -119,7 +119,8 @@ async def websocket_endpoint(ws: WebSocket, game_id: str):
await ws.send_text(WebSocketRequest(type=WebSocketTypes.Error, data="InvalidButton").json())
continue
await submit_answer_fn(answer_index, game_pin, player_id, now)
print(f"Data from client {game_id}: {data}")
except WebSocketDisconnect as ex:
print("Client {} is disconnected: {}".format(game_id, ex))
print(f"Client {game_id} is disconnected: {ex}")
wss_clients.pop(game_id, None)
classquiz/routers/editor.py
+20 -88
View File
@@ -4,24 +4,22 @@
import asyncio
import html
import re
import uuid
from typing import Optional
import asyncpg.exceptions
import bleach
from fastapi import APIRouter, File, UploadFile, HTTPException, Depends
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel
from classquiz.config import settings, redis, storage, meilisearch, ALLOWED_TAGS_FOR_QUIZ, server_regex
from classquiz.db.models import Quiz, QuizInput, User, QuizQuestionType
import puremagic
from classquiz.config import settings, redis, storage, meilisearch, ALLOWED_TAGS_FOR_QUIZ, arq
from classquiz.db.models import Quiz, QuizInput, User, QuizQuestionType, StorageItem
from classquiz.auth import get_current_user
import os
from datetime import datetime
from uuid import UUID
from classquiz.helpers import get_meili_data, check_hashcash
from classquiz.helpers import get_meili_data, check_image_string, extract_image_ids_from_quiz
from classquiz.storage.errors import DeletionFailedError
settings = settings()
@@ -67,56 +65,6 @@ async def init_editor(edit: bool, quiz_id: Optional[UUID] = None, user: User = D
return InitEditorResponse(token=edit_id)
class GetPowData(BaseModel):
data: str
@router.get("/pow", response_model=GetPowData)
async def get_pow_data(edit_id: str):
session_data = await redis.get(f"edit_session:{edit_id}")
if session_data is None:
raise HTTPException(status_code=401, detail="Edit ID not found!")
random_str = os.urandom(8).hex()
await redis.set(f"edit_session:{edit_id}:pow", random_str, ex=3800)
return GetPowData(data=random_str)
class UploadImageReturn(BaseModel):
id: str
pow_data: str
@router.post("/image", response_model=UploadImageReturn)
async def upload_image(edit_id: str, pow_data: str, file: UploadFile = File()):
session_data = await redis.get(f"edit_session:{edit_id}")
pow_data_server = await redis.get(f"edit_session:{edit_id}:pow")
uploaded_images = await redis.llen(f"edit_session:{edit_id}:images")
if pow_data_server is None:
raise HTTPException(status_code=401, detail="Edit ID not found!")
if session_data is None:
raise HTTPException(status_code=401, detail="Edit ID not found!")
if uploaded_images == 0 and not check_hashcash(pow_data, pow_data_server, "8"):
raise HTTPException(status_code=401, detail="Edit ID not found!")
if uploaded_images != 0 and not check_hashcash(pow_data, pow_data_server, "8"):
raise HTTPException(status_code=401, detail="Edit ID not found!")
file_bytes = await file.read()
if len(file_bytes) > 2000000:
raise HTTPException(status_code=400, detail="File too large")
try:
pm_data = puremagic.magic_string(file_bytes)[0]
except puremagic.PureError:
raise HTTPException(status_code=400, detail="Image couldn't be identified!")
if pm_data.extension not in allowed_image_extensions:
raise HTTPException(status_code=400, detail="Image-type now allowed!")
session_data = EditSessionData.parse_raw(session_data)
file_name = f"{session_data.quiz_id}--{uuid.uuid4()}"
await storage.upload(file_name=file_name, file_data=file_bytes)
await redis.lpush(f"edit_session:{edit_id}:images", file_name)
random_str = os.urandom(8).hex()
await redis.set(f"edit_session:{edit_id}:pow", random_str, ex=3800)
return UploadImageReturn(id=file_name, pow_data=random_str)
@router.post("/finish")
async def finish_edit(edit_id: str, quiz_input: QuizInput):
session_data = await redis.get(f"edit_session:{edit_id}")
@@ -141,25 +89,10 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
quiz_input.questions[i].answers[i2].answer = html.unescape(
bleach.clean(answer.answer, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True)
)
image_id_regex = r"^.{36}--.{36}$"
imgur_regex = r"^https://i\.imgur\.com\/.{7}.(jpg|png|gif)$"
extract_file_name_re = r"^.*/api/v1/storage/download/(.{36}--.{36})$"
images_to_delete = []
old_quiz_data: Quiz = await Quiz.objects.get_or_none(id=session_data.quiz_id, user_id=session_data.user_id)
def mark_image_for_deletion(new: str | None, index: int, old_quiz: Quiz | None):
if old_quiz is None:
return
try:
# Why does this work or not throw an error (TODO)
if new == old_quiz.questions[index]["image"]:
return
else:
images_to_delete.append(old_quiz.questions[index]["image"])
except IndexError:
pass
for i, question in enumerate(quiz_input.questions):
image = question.image
quiz_input.questions[i].question = html.unescape(
@@ -167,29 +100,20 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
)
if image == "":
question.image = None
mark_image_for_deletion(question.image, i, old_quiz_data)
elif image is None:
mark_image_for_deletion(question.image, i, old_quiz_data)
elif bool(re.match(image_id_regex, question.image)):
question.image = f"{settings.root_address}/api/v1/storage/download/{image}"
mark_image_for_deletion(question.image, i, old_quiz_data)
elif bool(re.match(imgur_regex, image)):
mark_image_for_deletion(question.image, i, old_quiz_data)
elif bool(re.match(server_regex, image)):
mark_image_for_deletion(question.image, i, old_quiz_data)
else:
if image is not None and not check_image_string(image)[0]:
raise HTTPException(status_code=400, detail="Image URL(s) aren't valid!")
if quiz_input.cover_image == "":
quiz_input.cover_image = None
# if quiz_input.background_image is None and old_quiz_data.background_image is not None:
# mark_image_for_deletion(quiz_input.background_image)
if quiz_input.cover_image is not None and not check_image_string(quiz_input.cover_image)[0]:
raise HTTPException(status_code=400, detail="image url is not valid")
if quiz_input.cover_image is not None and not bool(re.match(server_regex, quiz_input.cover_image)):
if quiz_input.background_image is not None and not check_image_string(quiz_input.background_image)[0]:
raise HTTPException(status_code=400, detail="image url is not valid")
if session_data.edit:
await arq.enqueue_job("quiz_update", old_quiz_data, old_quiz_data.id, _defer_by=2)
quiz = old_quiz_data
meilisearch.index(settings.meilisearch_index).update_documents([await get_meili_data(quiz)])
if not quiz_input.public:
@@ -207,13 +131,14 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
for image in images_to_delete:
if image is not None:
try:
await storage.delete([re.search(extract_file_name_re, image).group(1)])
await storage.delete([image])
except DeletionFailedError:
pass
await redis.srem("edit_sessions", edit_id)
await redis.delete(f"edit_session:{edit_id}")
await redis.delete(f"edit_session:{edit_id}:images")
return await quiz.update()
await quiz.update()
return quiz
else:
quiz = Quiz(
**quiz_input.dict(),
@@ -222,6 +147,7 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
created_at=datetime.now(),
updated_at=datetime.now(),
)
await redis.delete("global_quiz_count")
if quiz_input.public:
meilisearch.index(settings.meilisearch_index).add_documents([await get_meili_data(quiz)])
@@ -229,6 +155,12 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
await redis.srem("edit_sessions", edit_id)
await redis.delete(f"edit_session:{edit_id}")
await redis.delete(f"edit_session:{edit_id}:images")
return await quiz.save()
await quiz.save()
except asyncpg.exceptions.UniqueViolationError:
raise HTTPException(status_code=400, detail="The quiz already exists")
new_images = extract_image_ids_from_quiz(quiz)
for image in new_images:
item = await StorageItem.objects.get_or_none(id=uuid.UUID(image))
if item is None:
continue
await quiz.storageitems.add(item)
classquiz/routers/eximport.py
+30 -8
View File
@@ -1,6 +1,7 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import io
import json
import uuid
from datetime import datetime
@@ -10,10 +11,11 @@ from aiohttp import ClientSession
from fastapi import APIRouter, Depends, File, UploadFile, HTTPException
from fastapi.responses import StreamingResponse
from classquiz.auth import get_current_user
from classquiz.config import storage, settings
from classquiz.db.models import Quiz, User
from classquiz.config import storage, settings, arq
from classquiz.db.models import Quiz, User, StorageItem
import gzip
import urllib.parse
import magic
router = APIRouter()
settings = settings()
@@ -44,12 +46,13 @@ async def export_quiz(quiz_id: uuid.UUID, user: User = Depends(get_current_user)
quiz_dict["updated_at"] = quiz_dict["updated_at"].isoformat()
quiz_json = json.dumps(quiz_dict)
bin_data = gzip.compress(quiz_json.encode("utf-8"), compresslevel=9)
# bin_data = quiz_json.encode("utf-8")
bin_data = bin_data + quiz_delimiter
for image_key in image_urls.keys():
for image_key in image_urls:
bin_data = bin_data + image_delimiter + str(image_key).encode("utf-8") + image_index_delimiter
image_data = None
async with ClientSession() as session, session.get(image_urls[image_key]) as resp:
async with ClientSession() as session, session.get(
f"{settings.root_address}/api/v1/storage/download/{image_urls[image_key]}"
) as resp:
image_data = await resp.read()
bin_data = bin_data + image_data
@@ -68,6 +71,8 @@ async def export_quiz(quiz_id: uuid.UUID, user: User = Depends(get_current_user)
@router.post("/")
async def import_quiz(file: UploadFile = File(), user: User = Depends(get_current_user)):
if user.storage_used > settings.free_storage_limit:
raise HTTPException(status_code=409, detail="Storage limit reached")
data = await file.read()
[split_data, images] = data.split(quiz_delimiter)
decompressed_quiz = gzip.decompress(split_data)
@@ -75,15 +80,32 @@ async def import_quiz(file: UploadFile = File(), user: User = Depends(get_curren
image_splits = images.split(image_delimiter)
quiz_id = uuid.uuid4()
image_urls = {}
print(len(data))
for image_split in image_splits:
res = image_split.split(image_index_delimiter)
if len(res) != 2:
continue
[index, image_data] = res
print(len(image_data))
img_data = io.BytesIO(image_data)
mime_type = magic.from_buffer(img_data.read(2048), mime=True)
print(mime_type)
index = int(index.decode("utf-8"))
image_name = f"{quiz_id}--{uuid.uuid4()}"
await storage.upload(file_name=image_name, file_data=image_data)
image = f"{settings.root_address}/api/v1/storage/download/{image_name}"
file_id = uuid.uuid4()
file_obj = StorageItem(
id=file_id,
uploaded_at=datetime.now(),
mime_type=mime_type,
hash=None,
user=user,
size=0,
deleted_at=None,
alt_text=None,
)
await storage.upload(file_name=file_id.hex, file_data=img_data, mime_type=mime_type)
await file_obj.save()
await arq.enqueue_job("calculate_hash", file_id.hex)
image = file_id.hex
image_urls[index] = image
quiz_dict["created_at"] = datetime.fromisoformat(quiz_dict["created_at"])
quiz_dict["updated_at"] = datetime.fromisoformat(quiz_dict["updated_at"])
classquiz/routers/live.py
-8
View File
@@ -134,8 +134,6 @@ async def get_live_game_data(
@router.get("/user_count")
async def get_game_user_count(game_pin: str, api_key: str, as_string: bool = False, as_array: bool = False):
# if redis_res is None:
# raise HTTPException(status_code=404, detail="Game not found")
user_id = await check_api_key(api_key)
redis_res = await redis.get(f"game_session:{game_pin}")
if redis_res is None:
@@ -149,12 +147,6 @@ async def get_game_user_count(game_pin: str, api_key: str, as_string: bool = Fal
return {"players": {"count": player_count}}
# class _LivePlayersReturn(BaseModel):
# # players: list[GamePlayer | None]
# answers: list[GameAnswer1 | None]
# players: list[GamePlayer | None]
@router.get(
"/players",
)
classquiz/routers/login.py
+1 -1
View File
@@ -149,7 +149,7 @@ class StepInput(BaseModel):
@router.post("/step/{step_id}")
async def step_1(session_id: str, data: StepInput, request: Request, response: Response, step_id: int):
async def step_1_endpoint(session_id: str, data: StepInput, request: Request, response: Response, step_id: int):
if step_id < 0 or step_id > 2:
raise HTTPException(status_code=401)
redis_res = await redis.get(f"login_session:{session_id}")
classquiz/routers/pixabay.py
+62
View File
@@ -0,0 +1,62 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from datetime import datetime
from io import BytesIO
from uuid import uuid4
from aiohttp import ClientSession
from fastapi import APIRouter, Depends, HTTPException
from classquiz.auth import get_current_user
from classquiz.db.models import User, StorageItem, PublicStorageItem
from classquiz.helpers.pixabay import get_images, GetImagesParams, BoolInput, GetImagesResponse, NotFoundError
from classquiz.config import settings, storage, arq
settings = settings()
router = APIRouter()
@router.get("/images")
async def search_pixabay_images(query: str, page: int = 1, user: User = Depends(get_current_user)) -> GetImagesResponse:
if settings.pixabay_api_key is None:
raise HTTPException(status_code=423, detail="Pixabay not set up")
return await get_images(settings.pixabay_api_key, GetImagesParams(q=query, safesearch=BoolInput.true, page=page))
@router.post("/save")
async def save_pixabay_image(id: str, user: User = Depends(get_current_user)) -> PublicStorageItem:
if settings.pixabay_api_key is None:
raise HTTPException(status_code=423, detail="Pixabay not set up")
if user.storage_used > settings.free_storage_limit:
raise HTTPException(status_code=409, detail="Storage limit reached")
try:
images = await get_images(settings.pixabay_api_key, GetImagesParams(id=id, safesearch=BoolInput.true))
except NotFoundError:
raise HTTPException(status_code=404, detail="Pixabay file not found")
image = images.hits[0]
file_id = uuid4()
file_data = b""
async with ClientSession() as session, session.get(image.largeImageURL) as resp:
async for i in resp.content.iter_chunked(1024):
file_data += i
content_type = resp.headers.get("Content-Type")
if content_type is None:
content_type = "image/*"
file = BytesIO(file_data)
await storage.upload(file_name=file_id.hex, file_data=file, mime_type=content_type)
file_obj: StorageItem = StorageItem(
id=file_id,
uploaded_at=datetime.now(),
mime_type=content_type,
hash=None,
user=user,
size=0,
deleted_at=None,
alt_text=None,
imported=True,
)
await file_obj.save()
await arq.enqueue_job("calculate_hash", file_id.hex)
return PublicStorageItem.from_db_model(file_obj)
classquiz/routers/quiz.py
+9 -79
View File
@@ -10,18 +10,16 @@ from random import randint
import ormar.exceptions
from classquiz.helpers import get_meili_data, generate_spreadsheet
from classquiz.helpers import generate_spreadsheet
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import ValidationError, BaseModel
import bleach
from classquiz.auth import get_current_user
from classquiz.config import redis, settings, storage, meilisearch
from classquiz.db.models import Quiz, QuizInput, User, PlayGame, GameInLobby, QuizQuestion
from classquiz.db.models import Quiz, User, PlayGame, GameInLobby, QuizQuestion
from classquiz.helpers.box_controller import generate_code
from classquiz.kahoot_importer.import_quiz import import_quiz
import html
import urllib.parse
settings = settings()
@@ -29,33 +27,6 @@ settings = settings()
router = APIRouter()
@router.post("/create", deprecated=True)
async def create_quiz_lol(quiz_input: QuizInput, user: User = Depends(get_current_user)):
imgur_regex = r"^https://i\.imgur\.com\/.{7}.(jpg|png|gif)$"
server_regex = rf"^{re.escape(settings.root_address)}/api/v1/storage/download/.{36}--.{36}$"
quiz_input.title = html.unescape(bleach.clean(quiz_input.title, tags=[], strip=True))
quiz_input.description = html.unescape(bleach.clean(quiz_input.description, tags=[], strip=True))
for question in quiz_input.questions:
if question.image == "":
question.image = None
if (
question.image is not None
and not re.match(imgur_regex, question.image)
and not re.match(server_regex, question.image)
):
raise HTTPException(status_code=400, detail="image url is not valid")
if quiz_input.cover_image == "":
quiz_input.cover_image = None
if quiz_input.cover_image is not None and not bool(re.match(server_regex, quiz_input.cover_image)):
raise HTTPException(status_code=400, detail="image url is not valid")
quiz = Quiz(**quiz_input.dict(), user_id=user.id, id=uuid.uuid4())
await redis.delete("global_quiz_count")
if quiz_input.public:
meilisearch.index(settings.meilisearch_index).add_documents([await get_meili_data(quiz)])
return await quiz.save()
@router.get("/get/{quiz_id}")
async def get_quiz_from_id(quiz_id: str, user: User | None = Depends(get_current_user)):
try:
@@ -195,57 +166,14 @@ async def get_quiz_list(user: User = Depends(get_current_user), page_size: int |
raise HTTPException(status_code=400, detail="Invalid page(size). page(size) have to be greater than 0.")
@router.put("/update/{quiz_id}", deprecated=True)
async def update_quiz(quiz_id: str, quiz_input: QuizInput, user: User = Depends(get_current_user)):
imgur_regex = r"^https://i\.imgur\.com\/.{7}.(jpg|png|gif)$"
server_regex = rf"^{re.escape(settings.root_address)}/api/v1/storage/download/.{{36}}--.{{36}}$"
for question in quiz_input.questions:
if question.image == "":
question.image = None
if (
question.image is not None
and not bool(re.match(server_regex, question.image))
and not bool(re.match(imgur_regex, question.image))
):
raise HTTPException(status_code=400, detail="image url is not valid")
try:
quiz_id = uuid.UUID(quiz_id)
except ValueError:
raise HTTPException(status_code=400, detail="badly formed quiz id")
# Check Cover-Image
if quiz_input.cover_image == "":
quiz_input.cover_image = None
if quiz_input.cover_image is not None and not bool(re.match(server_regex, quiz_input.cover_image)):
raise HTTPException(status_code=400, detail="image url is not valid")
quiz = await Quiz.objects.get_or_none(id=quiz_id, user_id=user.id)
if quiz is None:
return JSONResponse(status_code=404, content={"detail": "quiz not found"})
else:
quiz_input.description = html.unescape(bleach.clean(quiz_input.description, tags=[], strip=True))
quiz_input.title = html.unescape(bleach.clean(quiz_input.title, tags=[], strip=True))
meilisearch.index(settings.meilisearch_index).update_documents([await get_meili_data(quiz)])
if quiz.public and not quiz_input.public:
meilisearch.index(settings.meilisearch_index).delete_document(str(quiz.id))
if not quiz.public and quiz_input.public:
meilisearch.index(settings.meilisearch_index).add_documents([await get_meili_data(quiz)])
quiz.title = quiz_input.title
quiz.cover_image = quiz_input.cover_image
quiz.public = quiz_input.public
quiz.description = quiz_input.description
quiz.updated_at = datetime.now()
quiz.questions = quiz_input.dict()["questions"]
return await quiz.update()
@router.post("/import/{quiz_id}")
async def import_quiz_route(quiz_id: str, user: User = Depends(get_current_user)):
if user.storage_used > settings.free_storage_limit:
raise HTTPException(status_code=409, detail="Storage limit reached")
try:
return await import_quiz(quiz_id, user)
except ValidationError:
except ValidationError as e:
print(e)
raise HTTPException(status_code=400, detail="This quiz isn't (yet) supported")
@@ -264,7 +192,9 @@ async def delete_quiz(quiz_id: str, user: User = Depends(get_current_user)):
for question in quiz.questions:
try:
if question["image"] is not None and not str(question["image"]).startswith("https://i.imgur.com/"):
pics_to_delete.append(pic_name_regex.match(question["image"]).group(1))
old_image_to_delete = pic_name_regex.match(question["image"])
if old_image_to_delete is not None:
pics_to_delete.append(old_image_to_delete.group(1))
except KeyError:
pass
if len(pics_to_delete) != 0:
classquiz/routers/quiztivity/__init__.py
+65
View File
@@ -0,0 +1,65 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from uuid import UUID, uuid4
from fastapi import APIRouter, Depends, HTTPException
from classquiz.auth import get_current_user
from datetime import datetime
from classquiz.db.models import User, QuizTivityInput, QuizTivity, QuizTivityShare, PublicQuizTivityShare
from classquiz.routers.quiztivity.shares import router as shares_router
router = APIRouter()
router.include_router(shares_router, prefix="/shares")
@router.post("/create")
async def create_quiztivity(data: QuizTivityInput, user: User = Depends(get_current_user)) -> QuizTivity:
quiztivity = QuizTivity.parse_obj({**data.dict(), "user": user, "id": uuid4(), "created_at": datetime.now()})
return await quiztivity.save()
@router.get("/{uuid}")
async def get_quiztivity(uuid: UUID) -> QuizTivity:
quiztivity = await QuizTivity.objects.get_or_none(id=uuid)
if quiztivity is None:
raise HTTPException(status_code=404, detail="QuizTivity not found")
return quiztivity
@router.put("/{uuid}")
async def put_quiztivity(data: QuizTivityInput, uuid: UUID, user: User = Depends(get_current_user)) -> QuizTivity:
quiztivity = await QuizTivity.objects.get_or_none(id=uuid, user=user)
if quiztivity is None:
raise HTTPException(status_code=404, detail="QuizTivity not found")
quiztivity.pages = data.dict()["pages"]
quiztivity.title = data.title
return await quiztivity.update()
@router.delete("/{uuid}")
async def delete_quiztivity(uuid: UUID):
quiztivity = await QuizTivity.objects.get_or_none(id=uuid)
if quiztivity is None:
raise HTTPException(status_code=404, detail="QuizTivity not found")
await quiztivity.delete()
@router.get("/")
async def get_all_quiztivities(user: User = Depends(get_current_user)) -> list[QuizTivity]:
quiztivities = await QuizTivity.objects.filter(user=user).order_by(QuizTivity.created_at.desc()).all()
return quiztivities
@router.get("/{uuid}/shares")
async def get_shares(uuid: UUID, user: User = Depends(get_current_user)) -> list[PublicQuizTivityShare]:
shares = (
await QuizTivityShare.objects.filter(quiztivity=uuid, user=user).order_by(QuizTivityShare.expire_at.asc()).all()
)
resp_shares = []
for share in shares:
resp_shares.append(PublicQuizTivityShare.from_db_model(share))
return resp_shares
classquiz/routers/quiztivity/shares.py
+84
View File
@@ -0,0 +1,84 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from uuid import UUID, uuid4
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from datetime import datetime, timedelta
from classquiz.auth import get_current_user
from classquiz.db.models import User, QuizTivityShare, QuizTivity, PublicQuizTivityShare
router = APIRouter()
@router.get("/")
async def get_shares(user: User = Depends(get_current_user)) -> list[PublicQuizTivityShare]:
shares = await QuizTivityShare.objects.filter(user=user).all()
resp_shares = []
for share in shares:
resp_shares.append(PublicQuizTivityShare.from_db_model(share))
return resp_shares
class CreateShareInput(BaseModel):
name: str | None
quiztivity: UUID
expire_in: int | None
@router.post("/")
async def create_share(data: CreateShareInput, user: User = Depends(get_current_user)) -> PublicQuizTivityShare:
quiztivity = await QuizTivity.objects.get_or_none(id=data.quiztivity, user=user)
expire_at = None
if data.expire_in is not None:
expire_at = (datetime.now() + timedelta(minutes=data.expire_in)).replace(tzinfo=None)
if quiztivity is None:
raise HTTPException(status_code=400, detail="Quiztivity wasn't found")
share = await QuizTivityShare.objects.create(
id=uuid4(), name=data.name, expire_at=expire_at, quiztivity=quiztivity, user=user
)
share = PublicQuizTivityShare.from_db_model(share)
return share
@router.delete("/{uuid}")
async def delete_share(uuid: UUID, user: User = Depends(get_current_user)):
share = await QuizTivityShare.objects.get_or_none(id=uuid, user=user)
if share is None:
raise HTTPException(status_code=404, detail="Share wasn't found")
await share.delete()
return
class UpdateShareInput(BaseModel):
name: str | None
expire_in: int | None
@router.put("/{uuid}")
async def update_share(
data: UpdateShareInput, uuid: UUID, user: User = Depends(get_current_user)
) -> PublicQuizTivityShare:
share = await QuizTivityShare.objects.get_or_none(id=uuid, user=user)
if share is None:
raise HTTPException(status_code=404, detail="Share wasn't found")
expire_at = None
if data.expire_in is not None:
expire_at = (datetime.now() + timedelta(minutes=data.expire_in)).replace(tzinfo=None)
share.name = data.name
share.expire_at = expire_at
return PublicQuizTivityShare.from_db_model(await share.update())
@router.get("/{uuid}")
async def get_share(uuid: UUID) -> QuizTivity:
share = await QuizTivityShare.objects.select_related("quiztivity").get_or_none(id=uuid)
if share is None:
raise HTTPException(status_code=404, detail="Share not found")
if share.expire_at is None:
return share.quiztivity
if share.expire_at < datetime.now():
raise HTTPException(status_code=410, detail="Already expired")
return share.quiztivity
classquiz/routers/results.py
+11 -10
View File
@@ -12,14 +12,14 @@ from classquiz.db.models import User, GameResults
router = APIRouter()
@router.get("/list", response_model=list[GameResults])
async def list_game_results(user: User = Depends(get_current_user)):
results = await GameResults.objects.all(user=user.id)
@router.get("/list")
async def list_game_results(user: User = Depends(get_current_user)) -> list[GameResults]:
results = await GameResults.objects.select_related(GameResults.quiz).all(user=user.id)
return results
@router.get("/list/{quiz_id}", response_model=list[GameResults])
async def get_results_by_quiz(quiz_id: UUID, user: User = Depends(get_current_user)):
@router.get("/list/{quiz_id}")
async def get_results_by_quiz(quiz_id: UUID, user: User = Depends(get_current_user)) -> list[GameResults]:
res = await GameResults.objects.all(user=user.id, quiz=quiz_id)
if res is None:
raise HTTPException(status_code=404, detail="Game Result not found")
@@ -27,9 +27,9 @@ async def get_results_by_quiz(quiz_id: UUID, user: User = Depends(get_current_us
return res
@router.get("/{game_id}", response_model=GameResults)
async def get_game_result(game_id: UUID, user: User = Depends(get_current_user)):
res = await GameResults.objects.get_or_none(user=user.id, id=game_id)
@router.get("/{game_id}")
async def get_game_result(game_id: UUID, user: User = Depends(get_current_user)) -> GameResults:
res = await GameResults.objects.select_related(GameResults.quiz).get_or_none(user=user.id, id=game_id)
if res is None:
raise HTTPException(status_code=404, detail="Game Result not found")
else:
@@ -40,8 +40,8 @@ class _SetNoteInput(BaseModel):
note: str
@router.post("/set_note", response_model=GameResults)
async def set_note(id: UUID, data: _SetNoteInput, user: User = Depends(get_current_user)):
@router.post("/set_note")
async def set_note(id: UUID, data: _SetNoteInput, user: User = Depends(get_current_user)) -> GameResults:
res = await GameResults.objects.get_or_none(user=user.id, id=id)
if res is None:
raise HTTPException(status_code=404, detail="Game Result not found")
@@ -49,6 +49,7 @@ async def set_note(id: UUID, data: _SetNoteInput, user: User = Depends(get_curre
return await res.update()
# skipcq: PYL-W0105
"""
@router.get("/export/{result_id}", response_class=StreamingResponse)
async def export_result(result_id: UUID, user: User = Depends(get_current_user)):
classquiz/routers/storage.py
+238 -14
View File
@@ -1,38 +1,262 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from datetime import datetime, timedelta
from tempfile import SpooledTemporaryFile
import re
from fastapi import APIRouter, HTTPException, UploadFile, File, Depends, Request, Response
from fastapi.responses import StreamingResponse, RedirectResponse
from pydantic import BaseModel
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from classquiz.config import settings, storage
from classquiz.auth import get_current_user
from classquiz.config import settings, storage, arq
from classquiz.db.models import User, StorageItem, PublicStorageItem, UpdateStorageItem, PrivateStorageItem
from classquiz.helpers import check_image_string
from classquiz.storage.errors import DownloadingFailedError
from uuid import uuid4, UUID
settings = settings()
router = APIRouter()
file_regex = r"^[a-z0-9]{8}-[a-z0-9-]{27}--[a-z0-9-]{36}$"
def headers_from_storage_item(item: StorageItem) -> dict[str, str]:
base_headers = {"Content-Type": item.mime_type}
if item.hash is not None:
base_headers["X-Hash"] = item.hash.hex()
if item.thumbhash is not None:
base_headers["X-Thumbhash"] = item.thumbhash
if item.alt_text is not None:
base_headers["X-Alt-Text"] = item.alt_text
if item.size != 0:
base_headers["Content-Size"] = str(item.size)
return base_headers
@router.get("/download/{file_name}")
async def download_file(file_name: str):
if not re.match(file_regex, file_name):
item = None
checked_image_string = check_image_string(file_name)
if not checked_image_string[0]:
raise HTTPException(status_code=400, detail="Invalid file name")
if checked_image_string[1] is not None:
item = await StorageItem.objects.get_or_none(id=checked_image_string[1])
if item is None:
print("Item not found")
raise HTTPException(status_code=404, detail="File not found")
file_name = item.storage_path
if file_name is None:
file_name = item.id.hex
if storage.backend == "s3":
if item is None:
return RedirectResponse(url=await storage.get_url(file_name, 300))
else:
return RedirectResponse(url=await storage.get_url(file_name, 300), headers=headers_from_storage_item(item))
try:
download = await storage.download(file_name)
download = storage.download(file_name)
except DownloadingFailedError:
print("error")
raise HTTPException(status_code=404, detail="File not found")
if download is None:
print("dload is none")
raise HTTPException(status_code=404, detail="File not found")
def iter_file():
yield from download
media_type = "image/*"
if item is not None:
media_type = item.mime_type
headers = {"Cache-Control": "public, immutable, max-age=31536000"}
if item is not None:
headers = {**headers, **headers_from_storage_item(item)}
return StreamingResponse(
iter_file(),
media_type="image/*",
headers={"Cache-Control": "public, immutable, max-age=31536000"},
download,
media_type=media_type,
headers=headers,
)
@router.get("/info/{file_name}")
async def get_basic_file_info(file_name: str) -> Response:
checked_image_string = check_image_string(file_name)
if not checked_image_string[0]:
raise HTTPException(status_code=404, detail="Invalid file name")
if checked_image_string[1] is not None:
item = await StorageItem.objects.get_or_none(id=checked_image_string[1])
if item is None:
raise HTTPException(status_code=404, detail="File not found")
# return PublicStorageItem.from_db_model(item)
storage_file_name = item.storage_path
if storage_file_name is None:
storage_file_name = item.id.hex
resp = Response(status_code=200, headers=headers_from_storage_item(item))
else:
resp = Response(status_code=200, headers={"Content-Type": "image/*"})
return resp
@router.head("/download/{file_name}")
async def download_file_head(file_name: str) -> Response:
checked_image_string = check_image_string(file_name)
if not checked_image_string[0]:
raise HTTPException(status_code=404, detail="Invalid file name")
if checked_image_string[1] is not None:
item = await StorageItem.objects.get_or_none(id=checked_image_string[1])
if item is None:
raise HTTPException(status_code=404, detail="File not found")
# return PublicStorageItem.from_db_model(item)
storage_file_name = item.storage_path
if storage_file_name is None:
storage_file_name = item.id.hex
resp = Response(status_code=200, headers=headers_from_storage_item(item))
else:
resp = Response(status_code=200, headers={"Content-Type": "image/*"})
storage_file_name = file_name
if storage.backend == "s3":
resp.status_code = 307
resp.headers.append("Location", await storage.get_url(storage_file_name, 300))
return resp
@router.post("/")
async def upload_file(file: UploadFile = File(), user: User = Depends(get_current_user)) -> PublicStorageItem:
if user.storage_used > settings.free_storage_limit:
raise HTTPException(status_code=409, detail="Storage limit reached")
file_id = uuid4()
size = 0
file_obj = StorageItem(
id=file_id,
uploaded_at=datetime.now(),
mime_type=file.content_type,
hash=None,
user=user,
size=size,
deleted_at=None,
alt_text=None,
)
await storage.upload(file_name=file_id.hex, file_data=file.file, mime_type=file.content_type)
await file_obj.save()
await arq.enqueue_job("calculate_hash", file_id.hex)
return PublicStorageItem.from_db_model(file_obj)
@router.post("/raw")
async def upload_raw_file(request: Request, user: User = Depends(get_current_user)) -> PublicStorageItem:
if user.storage_used > settings.free_storage_limit:
raise HTTPException(status_code=409, detail="Storage limit reached")
file_id = uuid4()
data_file = SpooledTemporaryFile(max_size=1000)
async for chunk in request.stream():
data_file.write(chunk)
data_file.seek(0)
file_obj = StorageItem(
id=file_id,
uploaded_at=datetime.now(),
mime_type=request.headers.get("Content-Type"),
hash=None,
user=user,
size=0,
deleted_at=None,
alt_text=None,
)
# https://github.com/VirusTotal/vt-py/issues/119#issuecomment-1261246867
await storage.upload(
file_name=file_id.hex,
# skipcq: PYL-W0212
file_data=data_file._file,
mime_type=request.headers.get("Content-Type"),
)
await file_obj.save()
await arq.enqueue_job("calculate_hash", file_id.hex)
return PublicStorageItem.from_db_model(file_obj)
@router.get("/meta/{file_id}")
async def get_file_info(file_id: UUID, user: User = Depends(get_current_user)) -> PublicStorageItem:
file_data = await StorageItem.objects.get_or_none(id=file_id, user=user, deleted_at=None)
if file_data is None:
raise HTTPException(status_code=404, detail="File not found")
return PublicStorageItem.from_db_model(file_data)
@router.delete("/meta/{file_id}")
async def mark_file_as_deleted(file_id: UUID, user: User = Depends(get_current_user)):
file_data = await StorageItem.objects.get_or_none(id=file_id, user=user, deleted_at=None)
if file_data is None:
raise HTTPException(status_code=404, detail="File not found")
storage_path = file_data.storage_path
if storage_path is None:
storage_path = file_data.id.hex
await storage.delete(storage_path)
file_data.deleted_at = datetime.now()
await file_data.update()
return
@router.put("/meta/{file_id}")
async def update_image_data(
file_id: UUID, data: UpdateStorageItem, user: User = Depends(get_current_user)
) -> PublicStorageItem:
file_data = await StorageItem.objects.get_or_none(id=file_id, user=user, deleted_at=None)
if file_data is None:
raise HTTPException(status_code=404, detail="File not found")
if data.alt_text == "":
data.alt_text = None
if data.filename == "":
data.filename = None
file_data.filename = data.filename
file_data.alt_text = data.alt_text
await file_data.update()
return PublicStorageItem.from_db_model(file_data)
@router.get("/list")
async def list_images(
since: datetime | None = None, user: User = Depends(get_current_user)
) -> list[PrivateStorageItem]:
if since is None:
since = datetime.now() - timedelta(weeks=9999)
storage_items = (
await StorageItem.objects.filter(user=user)
.filter(StorageItem.uploaded_at > since)
.filter(StorageItem.deleted_at == None) # noqa: E711
.order_by(StorageItem.uploaded_at.desc())
.select_related([StorageItem.quizzes, StorageItem.quiztivities])
.all()
)
if len(storage_items) == 0:
raise HTTPException(status_code=404, detail="No items found")
return_items: list[PrivateStorageItem] = []
for item in storage_items:
return_items.append(PrivateStorageItem.from_db_model(item))
return return_items
@router.get("/list/last")
async def get_latest_images(count: int = 50, user: User = Depends(get_current_user)) -> list[PrivateStorageItem]:
count = min(count, 50)
items = (
await StorageItem.objects.filter(user=user)
.limit(count)
.select_related([StorageItem.quizzes, StorageItem.quiztivities])
.order_by(StorageItem.uploaded_at.desc())
.all()
)
return_items: list[PrivateStorageItem] = []
for item in items:
return_items.append(PrivateStorageItem.from_db_model(item))
return return_items
class ReturnGetStorageLimit(BaseModel):
limit: int
limit_reached: bool
used: int
@router.get("/limit")
async def get_storage_limit(user: User = Depends(get_current_user)) -> ReturnGetStorageLimit:
user = await User.objects.get_or_none(id=user.id)
if user.storage_used > settings.free_storage_limit:
return ReturnGetStorageLimit(limit=settings.free_storage_limit, limit_reached=True, used=user.storage_used)
else:
return ReturnGetStorageLimit(limit=settings.free_storage_limit, limit_reached=False, used=user.storage_used)
classquiz/routers/users/__init__.py
-1
View File
@@ -79,7 +79,6 @@ async def create_user(user: RouteUser, background_task: BackgroundTasks) -> User
if len(user.username) == 32:
return JSONResponse({"details": "Username mustn't be 32 characters long"}, 400)
await user.save()
# print(settings.skip_email_verification)
if settings.skip_email_verification:
user.verify_key = None
user.verified = True
classquiz/socket_server/__init__.py
+8 -5
View File
@@ -252,8 +252,8 @@ class ReturnQuestion(QuizQuestion):
raise ValueError("Answers can't be none if type is ABCD")
if values["type"] == QuizQuestionType.RANGE and type(v) != RangeQuizAnswerWithoutSolution:
raise ValueError("Answer must be from type RangeQuizAnswer if type is RANGE")
# skipcq: PTC-W0047
if values["type"] == QuizQuestionType.VOTING and type(v[0]) != VotingQuizAnswer:
# print("Answer must be from type VotingQuizAnswer if type is VOTING")
pass
return v
@@ -262,7 +262,6 @@ class ReturnQuestion(QuizQuestion):
async def set_question_number(sid, data: str):
# data is just a number (as a str) of the question
session = await sio.get_session(sid)
# print("set_question_number", data, session)
if session["admin"]:
game_pin = session["game_pin"]
game_data = PlayGame.parse_raw(await redis.get(f"game:{session['game_pin']}"))
@@ -286,7 +285,6 @@ async def set_question_number(sid, data: str):
temp_return["type"] = game_data.questions[int(float(data))].type
if temp_return["type"] == QuizQuestionType.ORDER:
random.shuffle(temp_return["answers"])
# print("emitting")
await sio.emit(
"set_question_number",
{
@@ -359,6 +357,12 @@ async def submit_answer(sid: str, data: dict):
if data.answer.lower() == q.answer.lower():
answer_right = True
break
elif game_data.questions[int(data.question_index)].type == QuizQuestionType.CHECK:
correct_string = ""
for i, a in enumerate(game_data.questions[int(float(data.question_index))].answers):
if a.right:
correct_string += str(i)
answer_right = bool(correct_string == data.answer)
else:
raise NotImplementedError
latency = int(float((await sio.get_session(sid))["ping"]))
@@ -366,8 +370,6 @@ async def submit_answer(sid: str, data: dict):
answers = await redis.get(f"game_session:{session['game_pin']}:{data.question_index}")
diff = (time_q_started - now).total_seconds() * 1000 # - timedelta(milliseconds=latency)
# print(abs(diff) - latency, latency, abs(diff))
score = 0
if answer_right:
score = calculate_score(
@@ -385,6 +387,7 @@ async def submit_answer(sid: str, data: dict):
answers, game_pin=session["game_pin"], data=answer_data, q_index=int(float(data.question_index))
)
player_count = await redis.scard(f"game_session:{session['game_pin']}:players")
await sio.emit("player_answer", {})
if len(answers.__root__) == player_count:
# await sio.emit(
# "question_results",
classquiz/storage/__init__.py
+33 -33
View File
@@ -2,58 +2,58 @@
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from io import BytesIO
from typing import Optional
from typing import Optional, BinaryIO
from .deta_storage import DetaStorage
from .local_storage import LocalStorage
from .s3_storage import S3Storage
from typing import Generator
class Storage:
def __init__(
self,
backend: str,
deta_key: Optional[str],
deta_id: Optional[str],
storage_path: Optional[str],
access_key: str | None = None,
secret_key: str | None = None,
bucket_name: str | None = None,
base_url: str | None = None,
):
self.backend = backend
self.deta_key: str | None = deta_key
self.deta_id: str | None = deta_id
self.deta_base_url = f"https://drive.deta.sh/v1/{deta_id}/classquiz1"
if backend == "deta":
if deta_key is None or deta_id is None:
raise ValueError("deta_key and deta_id must be provided")
else:
self.deta_instance = DetaStorage(
deta_base_url=self.deta_base_url,
deta_key=self.deta_key,
deta_id=self.deta_id,
)
self.access_key = access_key
self.secret_key = secret_key
self.bucket_name = bucket_name
self.base_url = base_url
self.instance: LocalStorage | S3Storage | None = None
elif backend == "local":
if backend == "local":
if storage_path is None:
raise ValueError("storage_path must be provided")
else:
self.local_instance = LocalStorage(base_path=storage_path)
self.instance = LocalStorage(base_path=storage_path)
elif backend == "s3":
if access_key is None or secret_key is None or bucket_name is None or base_url is None:
raise ValueError("Not all parameters given")
self.instance = S3Storage(
base_url=base_url, access_key=access_key, secret_key=secret_key, bucket_name=bucket_name
)
else:
raise NotImplementedError(f"Backend {backend} not implemented")
async def download(self, file_name: str) -> BytesIO | None:
if self.backend == "deta":
return await self.deta_instance.download(file_name)
elif self.backend == "local":
return await self.local_instance.get_file(file_name)
def download(self, file_name: str) -> Generator | None:
return self.instance.download(file_name)
async def upload(self, file_name: str, file_data: bytes) -> None:
if self.backend == "deta":
return await self.deta_instance.upload(file=file_data, file_name=file_name)
elif self.backend == "local":
return await self.local_instance.write_file(file_name=file_name, data=file_data)
async def upload(self, file_name: str, file_data: BinaryIO, mime_type: str | None = None) -> None:
return await self.instance.upload(file=file_data, file_name=file_name, mime_type=mime_type)
async def delete(self, file_names: [str]) -> None:
if self.backend == "deta":
return await self.deta_instance.delete(file_names=file_names)
elif self.backend == "local":
return await self.local_instance.delete_file(file_names=file_names)
return await self.instance.delete(file_names=file_names)
async def get_url(self, file_name: str, expiry: int) -> str:
if self.backend == "s3":
return self.instance.get_url(file_name=file_name, expire=expiry)
async def get_file_size(self, file_name: str) -> int | None:
return self.instance.size(file_name)
classquiz/storage/deta_storage.py
-58
View File
@@ -1,58 +0,0 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from io import BytesIO
from classquiz.storage.errors import DeletionFailedError, SavingFailedError, DownloadingFailedError
from aiohttp import ClientSession
class DetaStorage:
def __init__(self, deta_base_url: str, deta_id: str, deta_key: str):
self.deta_url = deta_base_url
self.deta_id = deta_id
self.deta_key = deta_key
self.headers = {
"X-Api-Key": self.deta_key,
}
async def download(self, file_name: str) -> BytesIO | None:
"""
:param file_name: The name of the file to be downloaded
:return: Either bytes f successfull download or None if failed
"""
async with ClientSession(headers=self.headers) as session, session.get(
f"{self.deta_url}/files/download?name={file_name}"
) as response:
if response.status == 200:
return BytesIO(await response.read())
elif response.status == 404:
return None
else:
raise DownloadingFailedError
async def upload(self, file: bytes, file_name: str) -> None:
"""
:param file: The file in bytes
:param file_name: The name of the file
:return:
"""
async with ClientSession(headers=self.headers) as session, session.post(
f"{self.deta_url}/files?name={file_name}", data=file
) as response:
if response.status == 201:
return None
else:
print(response.status, await response.json())
raise SavingFailedError
async def delete(self, file_names: [str]) -> None:
async with ClientSession(headers=self.headers) as session, session.delete(
f"{self.deta_url}/files", json={"names": file_names}
) as response:
if response.status == 200:
return None
else:
raise DeletionFailedError
classquiz/storage/local_storage.py
+22 -8
View File
@@ -2,32 +2,46 @@
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import io
import os
from shutil import copyfileobj
from typing import BinaryIO, Generator
import aiofiles
import aiofiles.os
_DEFAULT_CHUNK_SIZE = 32768 # bytes; arbitrary
class LocalStorage:
def __init__(self, base_path: str):
self.base_path = base_path
async def get_file(self, file_name: str) -> io.BytesIO | None:
async def download(self, file_name: str) -> Generator | None:
try:
async with aiofiles.open(file=os.path.join(self.base_path, file_name), mode="rb") as f:
return io.BytesIO(await f.read())
while True:
chunk = await f.read(8192)
if not chunk:
break
yield chunk
except FileNotFoundError:
return None
yield None
async def write_file(self, file_name: str, data: bytes) -> None:
async with aiofiles.open(file=os.path.join(self.base_path, file_name), mode="wb") as f:
await f.write(data)
# skipcq: PYL-W0613
async def upload(self, file_name: str, file: BinaryIO, mime_type: str | None = None) -> None:
with open(file=os.path.join(self.base_path, file_name), mode="wb") as f:
copyfileobj(file, f)
async def delete_file(self, file_names: [str]) -> None:
async def delete(self, file_names: [str]) -> None:
for i in file_names:
try:
await aiofiles.os.remove(os.path.join(self.base_path, i))
except FileNotFoundError:
pass
return None
def size(self, file_name: str) -> int | None:
try:
return os.stat(os.path.join(self.base_path, file_name)).st_size
except FileNotFoundError:
return None
classquiz/storage/s3_storage.py
+157
View File
@@ -0,0 +1,157 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import hashlib
import hmac
from datetime import datetime, timedelta
from typing import Tuple, BinaryIO, Generator
from aiohttp import ClientSession
import minio
from pydantic import BaseModel
from classquiz.storage.errors import DeletionFailedError, SavingFailedError, DownloadingFailedError
class S3Storage:
class _HeaderAndParams(BaseModel):
params: dict[str, str]
headers: dict[str, str]
def __init__(self, base_url: str, access_key: str, secret_key: str, bucket_name: str, region: str = "us-east-1"):
self.base_url = base_url
self.access_key = access_key
self.secret_key = secret_key
self.bucket_name = bucket_name
self.region = region
self.DATE_FORMAT = "%a, %d %b %Y %H:%M:%S GMT"
self.host = base_url.replace("http://", "").replace("https://", "")
self.client = minio.Minio(self.host, access_key=access_key, secret_key=secret_key)
if not self.client.bucket_exists(self.bucket_name):
self.client.make_bucket(self.bucket_name)
def _generate_aws_signature_v4(self, method: str, path: str, expiry: int = None) -> Tuple[dict, str]:
path = f"/{self.bucket_name}{path}"
service = "s3"
# Create a timestamp for the request
t = datetime.utcnow()
amz_date = t.strftime("%Y%m%dT%H%M%SZ")
datestamp = t.strftime("%Y%m%d")
# Create a canonical request
canonical_uri = path
canonical_querystring = ""
if expiry is not None:
canonical_querystring = f"Expires={expiry}"
canonical_headers = "host:" + self.host + "\n" + "x-amz-date:" + amz_date + "\n"
signed_headers = "host;x-amz-date"
payload_hash = hashlib.sha256("".encode("utf-8")).hexdigest()
canonical_request = (
method
+ "\n"
+ canonical_uri
+ "\n"
+ canonical_querystring
+ "\n"
+ canonical_headers
+ "\n"
+ signed_headers
+ "\n"
+ payload_hash
)
# Create a string to sign
algorithm = "AWS4-HMAC-SHA256"
credential_scope = datestamp + "/" + self.region + "/" + service + "/" + "aws4_request"
string_to_sign = (
algorithm
+ "\n"
+ amz_date
+ "\n"
+ credential_scope
+ "\n"
+ hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
)
# Create a signing key
k_date = hmac.new(
("AWS4" + self.secret_key).encode("utf-8"), datestamp.encode("utf-8"), hashlib.sha256
).digest()
k_region = hmac.new(k_date, self.region.encode("utf-8"), hashlib.sha256).digest()
k_service = hmac.new(k_region, service.encode("utf-8"), hashlib.sha256).digest()
k_signing = hmac.new(k_service, b"aws4_request", hashlib.sha256).digest()
# Calculate the signature
signature = hmac.new(k_signing, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
# Add the signature to the request as an Authorization header
authorization_header = (
algorithm
+ " "
+ "Credential="
+ self.access_key
+ "/"
+ credential_scope
+ ", "
+ "SignedHeaders="
+ signed_headers
+ ", "
+ "Signature="
+ signature
)
# Send the request with the authorization header
headers = {"x-amz-date": amz_date, "Authorization": authorization_header}
request_url = self.base_url + path + "?" + canonical_querystring
return headers, request_url
# skipcq: PYL-W0613
async def upload(self, file: BinaryIO, file_name: str, mime_type: str | None = "application/octet-stream") -> None:
headers, url = self._generate_aws_signature_v4(method="PUT", path=f"/{file_name}")
async with ClientSession() as session, session.put(url, headers=headers, data=file) as resp:
if resp.status == 200:
return None
else:
print(await resp.text())
raise SavingFailedError
async def delete(self, file_names: list[str]) -> None:
for file in file_names:
headers, url = self._generate_aws_signature_v4(method="DELETE", path=f"/{file}")
async with ClientSession() as session, session.delete(url, headers=headers) as resp:
if resp.status == 204:
return None
else:
raise DeletionFailedError
def get_url(self, expire: int, file_name: str) -> str:
return self.client.presigned_get_object(
object_name=file_name, bucket_name=self.bucket_name, expires=timedelta(seconds=expire)
)
def size(self, file_name: str) -> int | None:
try:
res = self.client.stat_object(bucket_name=self.bucket_name, object_name=file_name)
except minio.error.S3Error:
return None
return res.size
async def download(self, file_name: str) -> Generator:
headers, url = self._generate_aws_signature_v4(method="GET", path=f"/{file_name}")
async with ClientSession() as session, session.get(url, headers=headers) as resp:
if resp.status == 200:
async for i in resp.content.iter_chunked(1024):
yield i
elif resp.status == 404:
yield None
else:
raise DownloadingFailedError
# client = httpx.AsyncClient()
# async with client.stream("GET", url, headers=headers) as resp:
# if resp.status == 200:
# yield resp.aiter_bytes()
classquiz/tests/__init__.py
+22 -1
View File
@@ -33,6 +33,11 @@ class ValueStorage:
exported_quiz_data = None
edit_id = None
image_id = None
cookies = None
file_id = None
quiztivity_id = None
share_id = None
expired_share_id = None
example_quiz = {
@@ -41,14 +46,16 @@ example_quiz = {
"description": "A description",
"questions": [
{
"type": "ABCD",
"question": "Is ClassQuiz cool?",
"time": 10,
"answers": [{"right": True, "answer": "Yes"}, {"right": False, "answer": "No"}],
},
{
"type": "ABCD",
"question": "Do you like open source?",
"time": 5,
"image": "https://i.imgur.com/sSNSy77.png",
"image": None,
"answers": [
{"right": True, "answer": "Yes"},
{"right": False, "answer": "No"},
@@ -60,6 +67,20 @@ example_quiz = {
test_user_email = "sth@byom.de"
test_user_password = "test"
example_quiztivity = {
"title": "Some test Quiztivity",
"pages": [
{
"title": "Some test question",
"type": "ABCD",
"data": {
"question": "Is ClassQuiz cool?",
"answers": [{"correct": True, "answer": "Yes"}, {"correct": False, "answer": "No"}],
},
}
],
}
# mock_test_results = {'0': [{'username': 'Player 1', 'answer': 'Byte, Bit, KB, MB, GB, TB', 'right': False},
# {'username': 'Player 2', 'answer': 'Byte, Bit, KB, MB, GB, TB', 'right': False}, {'username': 'Player 3',
# 'answer': 'Bit, Byte, KB, MB, GB, TB', 'right': True}], '1': [{'username': 'Player 3', 'answer': 'CPU',
classquiz/tests/test_server.py
+420 -256
View File
@@ -5,12 +5,11 @@
import uuid
import pytest
from httpx import AsyncClient
from redis import Redis
from classquiz.config import settings
from classquiz.tests import test_user_email, test_user_password
from classquiz.tests import test_user_email, test_user_password, example_quiztivity
from classquiz.tests import test_client, example_quiz, ValueStorage # noqa : F401
from classquiz.helpers.hashcash import mint
from fastapi.testclient import TestClient
# @pytest.fixture
@@ -25,8 +24,18 @@ from classquiz.helpers.hashcash import mint
class TestUsers:
@staticmethod
def log_in(tc: TestClient, email=test_user_email, password=test_user_password) -> int:
resp = tc.post("/api/v1/login/start", json={"email": email})
session_id = resp.json()["session_id"]
resp = tc.post(
f"/api/v1/login/step/1?session_id={session_id}", json={"auth_type": "PASSWORD", "data": password}
)
ValueStorage.cookies = resp.cookies
return resp.status_code
@pytest.mark.asyncio
async def test_create_test_user(self, test_client): # noqa : F811
async def test_create_test_user(self, test_client: TestClient): # noqa : F811
resp = test_client.post(
"/api/v1/users/create",
json={"email": test_user_email, "password": test_user_password, "username": "mawoka"},
@@ -55,156 +64,104 @@ class TestUsers:
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_verify_email(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
assert resp.status_code == 401
async def test_verify_email(self, test_client: TestClient): # noqa : F811
user = test_client.get(f"/api/v1/internal/testing/user/{test_user_email}?secret_key={settings().secret_key}")
assert (test_client.get("/api/v1/users/verify/dasadsasdadsasdsaddassad")).status_code == 404
test_client.get(f"/api/v1/users/verify/{user.json()['verify_key']}")
resp = test_client.post("/api/v1/login/start", json={"email": test_user_email})
assert resp.status_code == 200
session_id = resp.json()["session_id"]
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
f"/api/v1/login/step/1?session_id={session_id}", json={"auth_type": "PASSWORD", "data": test_user_password}
)
ValueStorage.cookies = resp.cookies
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_check(self, test_client): # noqa : F811
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}"})
async def test_check(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/users/check", cookies=ValueStorage.cookies)
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): # noqa : F811
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}"})
async def test_me(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/users/me", cookies=ValueStorage.cookies)
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): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
rememberme_token = resp.cookies["rememberme_token"]
resp = test_client.get("/api/v1/users/token/rememberme", cookies={"rememberme_token": rememberme_token})
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): # noqa : F811
# 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_token": rememberme_token}, allow_redirects=False
# )
# assert resp.status_code == 302
# resp = test_client.get("/api/v1/users/me", cookies={"rememberme_token": rememberme_token})
# assert resp.status_code == 401
@pytest.mark.asyncio
async def test_logout(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
rememberme_token = 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_token": rememberme_token}, allow_redirects=False
)
assert resp.status_code == 302
resp = test_client.get("/api/v1/users/me", cookies={"rememberme_token": rememberme_token})
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_password_update(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
token = resp.json()["access_token"]
async def test_password_update(self, test_client: TestClient): # noqa : F811
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}"},
cookies=ValueStorage.cookies,
)
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}"},
cookies=ValueStorage.cookies,
)
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_code = self.log_in(test_client, password="new_password")
assert resp_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}"},
cookies=ValueStorage.cookies,
)
assert resp.status_code == 200
resp1 = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
rememberme_token = resp1.cookies["rememberme_token"]
response = test_client.get("/api/v1/users/me", cookies={"rememberme_token": rememberme_token})
resp_code = self.log_in(test_client)
assert resp_code == 200
response = test_client.get("/api/v1/users/me", cookies=ValueStorage.cookies)
assert response.status_code == 200
@pytest.mark.asyncio
async def test_get_session(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
token = resp.cookies["rememberme_token"]
resp = test_client.get("/api/v1/users/session", cookies={"rememberme_token": token})
async def test_get_session(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/users/session", cookies=ValueStorage.cookies)
assert resp.status_code == 200
assert resp.json()["ip_address"] == "testclient"
@pytest.mark.asyncio
async def test_delete_session(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
token = resp.cookies["rememberme_token"]
resp = test_client.get("/api/v1/users/session", cookies={"rememberme_token": token})
session_id = resp.json()["id"]
resp = test_client.delete("/api/v1/users/sessions/" + str(session_id), cookies={"rememberme_token": token})
async def test_list_sessions(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/users/sessions/list", cookies=ValueStorage.cookies)
assert resp.status_code == 200
resp = test_client.delete("/api/v1/users/sessions/asdsadasdasdsad", cookies={"rememberme_token": token})
data = resp.json()
assert len(data) >= 1
@pytest.mark.asyncio
async def test_delete_session(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/users/session", cookies=ValueStorage.cookies)
session_id = resp.json()["id"]
resp = test_client.delete("/api/v1/users/sessions/" + str(session_id), cookies=ValueStorage.cookies)
assert resp.status_code == 200
resp = test_client.delete("/api/v1/users/sessions/asdsadasdasdsad", cookies=ValueStorage.cookies)
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_list_sessions(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
token = resp.cookies["rememberme_token"]
resp = test_client.get("/api/v1/users/sessions/list", cookies={"rememberme_token": token})
assert resp.status_code == 200
assert len(resp.json()) >= 1
@pytest.mark.asyncio
async def test_forgotten_password(self, test_client): # noqa : F811
async def test_forgotten_password(self, test_client: TestClient): # noqa : F811
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 == 200
@pytest.mark.asyncio
async def test_reset_password_with_token(self, test_client): # noqa : F811
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()
async def test_reset_password_with_token(self, test_client: TestClient): # noqa : F811
me = test_client.get("/api/v1/users/me", cookies=ValueStorage.cookies).json()
redis = Redis().from_url(settings().redis)
redis.set("reset_passwd:_1token_", str(me["id"]))
redis.set("reset_passwd:_2token_", str(uuid.uuid4()))
@@ -217,40 +174,29 @@ class TestUsers:
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"]
self.log_in(test_client, password="new_password")
test_client.put(
"/api/v1/users/password/update",
json={"new_password": test_user_password, "old_password": "new_password"},
cookies={"access_token": token},
cookies=ValueStorage.cookies,
)
redis.flushdb()
@pytest.mark.asyncio
async def test_signout_everywhere(self, test_client): # noqa : F811
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})
async def test_signout_everywhere(self, test_client: TestClient): # noqa : F811
resp = test_client.delete("/api/v1/users/signout-everywhere", cookies=ValueStorage.cookies)
assert resp.status_code == 200
class TestUtils:
@pytest.mark.asyncio
async def test_get_ip_data(self, test_client): # noqa : F811
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})
async def test_get_ip_data(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/utils/ip-lookup/1.1.1.1", cookies=ValueStorage.cookies)
assert resp.status_code == 200
assert resp.json()["query"] == "1.1.1.1"
@pytest.mark.asyncio
async def test_get_qr(self, test_client): # noqa : F811
async def test_get_qr(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/utils/qr/12345678")
assert resp.status_code == 200
assert resp.headers["Content-Type"] == "image/svg+xml"
@@ -258,7 +204,7 @@ class TestUtils:
class TestStats:
@pytest.mark.asyncio
async def test_get_quiz_count(self, test_client): # noqa : F811
async def test_get_quiz_count(self, test_client: TestClient): # noqa : F811
redis = Redis().from_url(settings().redis)
redis.flushdb()
for _ in range(2):
@@ -267,7 +213,7 @@ class TestStats:
assert resp.text == str(0)
@pytest.mark.asyncio
async def test_get_user_count(self, test_client): # noqa : F811
async def test_get_user_count(self, test_client: TestClient): # noqa : F811
redis = Redis().from_url(settings().redis)
redis.flushdb()
for _ in range(2):
@@ -276,7 +222,7 @@ class TestStats:
assert resp.text == str(1)
@pytest.mark.asyncio
async def test_get_combined_count(self, test_client): # noqa : F811
async def test_get_combined_count(self, test_client: TestClient): # noqa : F811
redis = Redis().from_url(settings().redis)
redis.flushdb()
for _ in range(2):
@@ -288,150 +234,132 @@ class TestStats:
class TestQuiz:
@pytest.mark.asyncio
async def test_create_quiz(self, test_client): # noqa : F811
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})
async def test_create_quiz(self, test_client: TestClient): # noqa : F811
resp = test_client.post("/api/v1/editor/start?edit=false", cookies=ValueStorage.cookies)
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
edit_token = resp.json()["token"]
assert len(edit_token) == 8
resp = test_client.post(
f"/api/v1/editor/finish?edit_id={edit_token}", json=example_quiz, cookies=ValueStorage.cookies
)
assert resp.status_code == 200
resp = test_client.get("/api/v1/quiz/list", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert len(data) == 1
ValueStorage.quiz_id = data[0]["id"]
@pytest.mark.asyncio
async def test_get_quiz_from_id(self, test_client): # noqa : F811
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})
async def test_get_quiz_from_id(self, test_client: TestClient): # noqa : F811
resp = test_client.get(f"/api/v1/quiz/get/{ValueStorage.quiz_id}", cookies=ValueStorage.cookies)
assert resp.status_code == 200
resp = test_client.get("/api/v1/quiz/get/dasdsadsadsadsadsa", cookies={"access_token": token})
resp = test_client.get("/api/v1/quiz/get/dasdsadsadsadsadsa", cookies=ValueStorage.cookies)
assert resp.status_code == 400
resp = test_client.get("/api/v1/quiz/get/847c64d3-39f9-4bb7-8f13-fae913f67858", cookies={"access_token": token})
resp = test_client.get("/api/v1/quiz/get/847c64d3-39f9-4bb7-8f13-fae913f67858", cookies=ValueStorage.cookies)
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_list_quizzes(self, test_client): # noqa : F811
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})
async def test_list_quizzes(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/quiz/list", cookies=ValueStorage.cookies)
assert resp.status_code == 200
assert resp.json()[0]["id"] == ValueStorage.quiz_id
@pytest.mark.asyncio
async def test_update_quiz(self, test_client): # noqa : F811
async def test_update_quiz(self, test_client: TestClient): # noqa : F811
example_quiz["public"] = True
example_quiz["questions"][1]["image"] = "https://i.imgur.com/sSNSy77.png"
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
f"/api/v1/editor/start?edit=true&quiz_id={ValueStorage.quiz_id}", cookies=ValueStorage.cookies
)
token = resp.cookies["access_token"]
resp = test_client.put(
f"/api/v1/quiz/update/{ValueStorage.quiz_id}", json=example_quiz, cookies={"access_token": token}
edit_id = resp.json()["token"]
resp = test_client.post(
f"/api/v1/editor/finish?edit_id={edit_id}", json=example_quiz, cookies=ValueStorage.cookies
)
assert resp.status_code == 200
resp = test_client.put(
"/api/v1/quiz/update/f183e091-a863-44ec-a1b7-c70eb92e3f6a",
json=example_quiz,
cookies={"access_token": token},
resp = test_client.post(
"/api/v1/editor/start?edit=true&quiz_id=f183e091-a863-44ec-a1b7-c70eb92e3f6a", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.put(
"/api/v1/quiz/update/saddsaasddsadsa", json=example_quiz, cookies={"access_token": token}
)
assert resp.status_code == 400
resp = test_client.post("/api/v1/editor/start?edit=true&quiz_id=asddasasdasdads", cookies=ValueStorage.cookies)
assert resp.status_code == 422
example_quiz["public"] = False
test_client.put(
f"/api/v1/quiz/update/{ValueStorage.quiz_id}", json=example_quiz, cookies={"access_token": token}
resp = test_client.post(
f"/api/v1/editor/start?edit=true&quiz_id={ValueStorage.quiz_id}", cookies=ValueStorage.cookies
)
edit_id = resp.json()["token"]
test_client.post(f"/api/v1/editor/finish?edit_id={edit_id}", json=example_quiz, cookies=ValueStorage.cookies)
example_quiz["public"] = True
test_client.put(
f"/api/v1/quiz/update/{ValueStorage.quiz_id}", json=example_quiz, cookies={"access_token": token}
resp = test_client.post(
f"/api/v1/editor/start?edit=true&quiz_id={ValueStorage.quiz_id}", cookies=ValueStorage.cookies
)
edit_id = resp.json()["token"]
test_client.post(f"/api/v1/editor/finish?edit_id={edit_id}", json=example_quiz, cookies=ValueStorage.cookies)
@pytest.mark.asyncio
async def test_import_quiz(self, test_client): # noqa : F811
async def test_import_quiz(self, test_client: TestClient): # noqa : F811
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/import/1f95eb0b-fcf4-4db2-879b-5418ef75116b", cookies={"access_token": token}
"/api/v1/quiz/import/1f95eb0b-fcf4-4db2-879b-5418ef75116b", cookies=ValueStorage.cookies
)
assert resp.status_code == 200
ValueStorage.imported_quizzes.append(resp.json()["id"])
resp = test_client.post("/api/v1/quiz/import/1f95eb0bdassdadasdas", cookies={"access_token": token})
resp = test_client.post("/api/v1/quiz/import/1f95eb0bdassdadasdas", cookies=ValueStorage.cookies)
assert resp.text == '"quiz not found"'
@pytest.mark.asyncio
async def test_get_public_quiz(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
token = resp.cookies["access_token"]
async def test_get_public_quiz(self, test_client: TestClient): # noqa : F811
resp = test_client.get(f"/api/v1/quiz/get/public/{ValueStorage.imported_quizzes[0]}")
assert resp.status_code == 200
resp = test_client.get(
"/api/v1/quiz/get/public/f183e091-a863-44ec-a1b7-c70eb92e3f6a", cookies={"access_token": token}
"/api/v1/quiz/get/public/f183e091-a863-44ec-a1b7-c70eb92e3f6a", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.get("/api/v1/quiz/get/public/dadasdas92e3f6a", cookies={"access_token": token})
resp = test_client.get("/api/v1/quiz/get/public/dadasdas92e3f6a", cookies=ValueStorage.cookies)
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_search_get(self, test_client): # noqa : F811
async def test_search_get(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/search/?q=*")
assert resp.status_code == 200
assert len(resp.json()["hits"]) > 0
@pytest.mark.asyncio
async def test_search_post(self, test_client): # noqa : F811
async def test_search_post(self, test_client: TestClient): # noqa : F811
resp = test_client.post("/api/v1/search/", json={"q": "*"})
assert resp.status_code == 200
assert len(resp.json()["hits"]) > 0
@pytest.mark.asyncio
async def test_image_cdn(self, test_client): # noqa : F811
async def test_image_cdn(self, test_client: TestClient): # noqa : F811
resp = test_client.get(f"/api/v1/quiz/get/public/{ValueStorage.imported_quizzes[0]}")
assert resp.status_code == 200
quiz = resp.json()
image_url = quiz["questions"][0]["image"]
resp = test_client.get(image_url)
assert resp.status_code == 200
resp = test_client.get(f"{image_url}sadgvsadgvhsad")
image_id = quiz["questions"][0]["image"]
# resp = test_client.get(f"/api/v1/storage/download/{image_id}")
# print(resp.text)
# assert resp.status_code == 200 This fails because I don't know
resp = test_client.get(f"/api/v1/storage/download/{image_id}sadgvsadgvhsad")
assert resp.status_code == 400
class TestPlayQuiz:
@pytest.mark.asyncio
async def test_start_quiz(self, test_client): # noqa : F811
async def test_start_quiz(self, test_client: TestClient): # noqa : F811
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/start/fb5adc91-629e-416e-8b98-ae400e36417c?game_mode=kahoot", cookies={"access_token": token}
"/api/v1/quiz/start/fb5adc91-629e-416e-8b98-ae400e36417c?game_mode=kahoot", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.post(
"/api/v1/quiz/start/fb5adc91-629e-416e-8b98-ae400sdadsasadsadasddsae36417c?game_mode=kahoot",
cookies={"access_token": token},
cookies=ValueStorage.cookies,
)
assert resp.status_code == 400
resp = test_client.post(
f"/api/v1/quiz/start/{ValueStorage.quiz_id}?game_mode=kahoot", cookies={"access_token": token}
f"/api/v1/quiz/start/{ValueStorage.quiz_id}?game_mode=kahoot", cookies=ValueStorage.cookies
)
ValueStorage.game_pin = resp.json()["game_pin"]
ValueStorage.game_id = resp.json()["game_id"]
@pytest.mark.asyncio
async def test_check_captcha_enabled(self, test_client): # noqa : F811
async def test_check_captcha_enabled(self, test_client: TestClient): # noqa : F811
res = test_client.get(f"/api/v1/quiz/play/check_captcha/{ValueStorage.game_pin}")
assert res.status_code == 200
assert res.json()["enabled"] is True
@@ -440,7 +368,7 @@ class TestPlayQuiz:
assert res.status_code == 404
@pytest.mark.asyncio
async def test_join_game_route(self, test_client): # noqa : F811
async def test_join_game_route(self, test_client: TestClient): # noqa : F811
res = test_client.get(f"/api/v1/quiz/join/{ValueStorage.game_pin}")
assert res.status_code == 200
assert res.text == f'"{ValueStorage.game_id}"'
@@ -462,98 +390,334 @@ class TestCache:
"/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/me", cookies={"access_token": token})
resp = test_client.get("/api/v1/users/me", cookies=ValueStorage.cookies)
user = await get_user_from_id(resp.json()["id"])
"""
class TestEditor:
class TestCommunity:
@pytest.mark.asyncio
async def test_start(self, test_client): # noqa : F811
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/editor/start?edit=false", cookies={"access_token": token})
async def test_get_user_by_id(self, test_client: TestClient): # noqa : F811
user = test_client.get("/api/v1/users/me", cookies=ValueStorage.cookies)
user_id = user.json()["id"]
resp = test_client.get(f"/api/v1/community/user/{user_id}")
assert resp.status_code == 200
edit_id = resp.json()["token"]
resp = test_client.get(f"/api/v1/editor/pow?edit_id={edit_id}")
assert resp.status_code == 200
pow_data = resp.json()["data"]
resp = test_client.get("/api/v1/editor/pow?edit_id=loladdfs")
assert resp.status_code == 401
pow_res = mint(pow_data, 8, None, "", 8, False)
print("POW-Res", pow_res)
async with AsyncClient() as ac:
resp = await ac.get("https://i.imgur.com/OE22DNZ.png")
image_bytes = resp.read()
resp = test_client.post(
f"/api/v1/editor/image?edit_id={edit_id}&pow_data={pow_res}", files={"file": image_bytes}
)
assert resp.status_code == 200
ValueStorage.edit_id = edit_id
ValueStorage.image_id = resp.json()["id"]
resp = test_client.get("/api/v1/community/user/e673c9ca-0cdf-4ebf-bad2-7d009ef5c62b")
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_finish(self, test_client): # noqa : F811
local_example_quiz = example_quiz
local_example_quiz["questions"][1][
"image"
] = f"http://localhost:8080/api/v1/storage/download/{ValueStorage.image_id}"
resp = test_client.post(f"/api/v1/editor/finish?edit_id={ValueStorage.edit_id}", json=example_quiz)
async def test_get_quizzes_from_user(self, test_client: TestClient): # noqa : F811
user = test_client.get("/api/v1/users/me", cookies=ValueStorage.cookies)
user_id = user.json()["id"]
resp = test_client.get(f"/api/v1/community/quizzes/{user_id}")
data = resp.json()
assert type(data) is list
class TestSitemap:
@pytest.mark.asyncio
async def test_get_sitemap(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/sitemap/get")
assert resp.status_code == 200
resp = test_client.get("/api/v1/sitemap/get")
assert resp.status_code == 200
class TestStorage:
@pytest.mark.asyncio
async def test_upload_file(self, test_client: TestClient): # noqa : F811
resp = test_client.post(
"/api/v1/storage/", cookies=ValueStorage.cookies, files={"file": ("img.svg", "svg_content")}
)
assert resp.status_code == 200
data = resp.json()
ValueStorage.file_id = data["id"]
@pytest.mark.asyncio
async def test_upload_raw_file(self, test_client: TestClient): # noqa : F811
resp = test_client.request(
"POST",
"/api/v1/storage/raw",
data=b"data!",
headers={"Content-Type": "image/svg+xml"},
cookies=ValueStorage.cookies,
)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_get_file_info(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/storage/meta/dsadsaasdas", cookies=ValueStorage.cookies)
assert resp.status_code == 422
resp = test_client.get(
"/api/v1/storage/meta/35c4f635-906a-46d7-8ab8-0520105ffff5", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.get(f"/api/v1/storage/meta/{ValueStorage.file_id}", cookies=ValueStorage.cookies)
data = resp.json()
assert data["size"] == 0
assert data["imported"] is False
assert data["alt_text"] is None
assert data["filename"] is None
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_update_image_data(self, test_client: TestClient): # noqa : F811
resp = test_client.put(
f"/api/v1/storage/meta/{ValueStorage.file_id}",
json={"alt_text": "Alt", "filename": "Filename"},
cookies=ValueStorage.cookies,
)
assert resp.status_code == 200
resp = test_client.put(
f"/api/v1/storage/meta/{ValueStorage.file_id}",
json={"alt_text": "", "filename": ""},
cookies=ValueStorage.cookies,
)
assert resp.status_code == 200
resp = test_client.get(f"/api/v1/storage/meta/{ValueStorage.file_id}", cookies=ValueStorage.cookies)
data = resp.json()
assert data["alt_text"] is None
assert data["filename"] is None
@pytest.mark.asyncio
async def test_list_images(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/storage/list", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert type(data) is list
assert len(data) >= 2
@pytest.mark.asyncio
async def test_get_latest_images(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/storage/list/last", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert type(data) is list
assert len(data) >= 2
resp = test_client.get("/api/v1/storage/list/last?count=1", cookies=ValueStorage.cookies)
assert resp.status_code == 200
# assert len(data) == 1
@pytest.mark.asyncio
async def test_get_storage_limit(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/storage/limit", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert data["limit_reached"] is False
assert type(data["limit"]) is int
assert data["used"] == 0
class TestQuizivity:
@pytest.mark.asyncio
async def test_create_quiztivity(self, test_client: TestClient): # noqa : F811
resp = test_client.post("/api/v1/quiztivity/create", cookies=ValueStorage.cookies, json=example_quiztivity)
assert resp.status_code == 200
data = resp.json()
ValueStorage.quiztivity_id = data["id"]
@pytest.mark.asyncio
async def test_get_quiztivity(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/quiztivity/8bd77201-65ed-46fe-9160-cfe71dad501f", cookies=ValueStorage.cookies)
assert resp.status_code == 404
resp = test_client.get(f"/api/v1/quiztivity/{ValueStorage.quiztivity_id}", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert data["id"] == ValueStorage.quiztivity_id
@pytest.mark.asyncio
async def test_put_quiztivity(self, test_client: TestClient): # noqa : F811
example_quiztivity["title"] = "New title"
resp = test_client.put(
f"/api/v1/quiztivity/{ValueStorage.quiztivity_id}", json=example_quiztivity, cookies=ValueStorage.cookies
)
assert resp.status_code == 200
resp = test_client.put(
"/api/v1/quiztivity/8bd77201-65ed-46fe-9160-cfe71dad501f",
json=example_quiztivity,
cookies=ValueStorage.cookies,
)
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_get_all_quiztivities(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/quiztivity/", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert type(data) is list
assert data[0]["id"] == ValueStorage.quiztivity_id
@pytest.mark.asyncio
async def test_create_share(self, test_client: TestClient): # noqa : F811
resp = test_client.post(
"/api/v1/quiztivity/shares/",
cookies=ValueStorage.cookies,
json={"quiztivity": ValueStorage.quiztivity_id, "expire_in": None},
)
assert resp.status_code == 200
data = resp.json()
ValueStorage.share_id = data["id"]
resp = test_client.post(
"/api/v1/quiztivity/shares/",
cookies=ValueStorage.cookies,
json={"quiztivity": "a090077f-9059-42bc-9783-f2cd01e069b8", "expire_in": None},
)
assert resp.status_code == 400
resp = test_client.post(
"/api/v1/quiztivity/shares/",
cookies=ValueStorage.cookies,
json={"quiztivity": ValueStorage.quiztivity_id, "expire_in": 0},
)
data = resp.json()
ValueStorage.expired_share_id = data["id"]
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_get_share(self, test_client: TestClient): # noqa : F811
resp = test_client.get(
"/api/v1/quiztivity/shares/8bd77201-65ed-46fe-9160-cfe71dad501f", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.get(f"/api/v1/quiztivity/shares/{ValueStorage.share_id}", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert data["id"] == ValueStorage.quiztivity_id
resp = test_client.get(
f"/api/v1/quiztivity/shares/{ValueStorage.expired_share_id}", cookies=ValueStorage.cookies
)
assert resp.status_code == 410
@pytest.mark.asyncio
async def test_update_share(self, test_client: TestClient): # noqa : F811
resp = test_client.put(
"/api/v1/quiztivity/shares/8bd77201-65ed-46fe-9160-cfe71dad501f",
cookies=ValueStorage.cookies,
json={"expire_in": 50},
)
assert resp.status_code == 404
resp = test_client.put(
f"/api/v1/quiztivity/shares/{ValueStorage.share_id}", cookies=ValueStorage.cookies, json={"expire_in": 50}
)
assert resp.status_code == 200
async def test_delete_share(self, test_client: TestClient): # noqa : F811
resp = test_client.delete(
"/api/v1/quiztivity/shares/8bd77201-65ed-46fe-9160-cfe71dad501f", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.delete(
f"/api/v1/quiztivity/shares/{ValueStorage.expired_share_id}", cookies=ValueStorage.cookies
)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_get_shares(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/quiztivity/shares/", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert type(data) is list
assert data[0]["id"] == ValueStorage.share_id
assert data[0]["expire_in"] == 49
@pytest.mark.asyncio
async def test_get_shares_by_quiztivity(self, test_client: TestClient): # noqa : F811
resp = test_client.get(f"/api/v1/quiztivity/{ValueStorage.quiztivity_id}/shares", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert type(data) is list
assert data[0]["id"] == ValueStorage.share_id
@pytest.mark.asyncio
async def test_get_shares(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/quiztivity/shares/", cookies=ValueStorage.cookies)
assert resp.status_code == 200
data = resp.json()
assert type(data) is list
assert data[0]["id"] == ValueStorage.share_id
@pytest.mark.asyncio
async def test_delete_quiztivity(self, test_client: TestClient): # noqa : F811
test_client.delete(f"/api/v1/quiztivity/shares/{ValueStorage.share_id}", cookies=ValueStorage.cookies)
resp = test_client.delete(
"/api/v1/quiztivity/8bd77201-65ed-46fe-9160-cfe71dad501f", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.delete(f"/api/v1/quiztivity/{ValueStorage.quiztivity_id}", cookies=ValueStorage.cookies)
assert resp.status_code == 200
class TestAvatar:
@pytest.mark.asyncio
async def test_get_customized_avatar(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/avatar/custom?skin_color=69", cookies=ValueStorage.cookies)
assert resp.status_code == 400
resp = test_client.get("/api/v1/avatar/custom", cookies=ValueStorage.cookies)
assert resp.status_code == 200
assert "image/svg+xml" in resp.headers.get("Content-Type")
@pytest.mark.asyncio
async def test_save_avatar(self, test_client: TestClient): # noqa : F811
resp = test_client.post("/api/v1/avatar/save?skin_color=69", cookies=ValueStorage.cookies)
assert resp.status_code == 400
resp = test_client.post("/api/v1/avatar/save", cookies=ValueStorage.cookies)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_get_own_avatar(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/users/avatar", cookies=ValueStorage.cookies)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_get_other_avatar(self, test_client: TestClient): # noqa : F811
resp = test_client.get(
"/api/v1/users/8bd77201-65ed-46fe-9160-cfe71dad501f/avatar", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
user_id = test_client.get("/api/v1/users/me", cookies=ValueStorage.cookies).json()["id"]
resp = test_client.get(f"/api/v1/users/avatar/{user_id}", cookies=ValueStorage.cookies)
assert resp.status_code == 200
class TestExImport:
@pytest.mark.asyncio
async def test_export_quiz(self, test_client): # noqa : F811
resp = test_client.get("/api/v1/eximport/jgfgufgfgfzftzi")
async def test_export_quiz(self, test_client: TestClient): # noqa : F811
resp = test_client.get("/api/v1/eximport/jgfgufgfgfzftzi", cookies=ValueStorage.cookies)
assert resp.status_code == 422
resp = test_client.get("/api/v1/eximport/8bd77201-65ed-46fe-9160-cfe71dad501f")
resp = test_client.get("/api/v1/eximport/8bd77201-65ed-46fe-9160-cfe71dad501f", cookies=ValueStorage.cookies)
assert resp.status_code == 404
resp = test_client.get(f"/api/v1/eximport/{ValueStorage.quiz_id}")
resp = test_client.get(f"/api/v1/eximport/{ValueStorage.quiz_id}", cookies=ValueStorage.cookies)
assert resp.status_code == 200
exported_data = resp.content
assert len(exported_data) > 3000
assert len(exported_data) < 3000
ValueStorage.exported_quiz_data = exported_data
@pytest.mark.asyncio
async def test_import_quiz(self, test_client): # noqa : F811
resp = test_client.post("/api/v1/eximport/", files={"file": ValueStorage.exported_quiz_data})
async def test_import_quiz(self, test_client: TestClient): # noqa : F811
resp = test_client.post(
"/api/v1/eximport/", files={"file": ValueStorage.exported_quiz_data}, cookies=ValueStorage.cookies
)
assert resp.status_code == 200
class TestDeleteStuff:
@pytest.mark.asyncio
async def test_delete_quiz(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
token = resp.cookies["access_token"]
async def test_delete_quiz(self, test_client: TestClient): # noqa : F811
resp = test_client.delete(
f"/api/v1/quiz/delete/{ValueStorage.imported_quizzes[0]}", cookies={"access_token": token}
f"/api/v1/quiz/delete/{ValueStorage.imported_quizzes[0]}", cookies=ValueStorage.cookies
)
assert resp.status_code == 200
resp = test_client.delete(
"/api/v1/quiz/delete/be582c77-da03-4271-929c-5d582056eb78", cookies={"access_token": token}
"/api/v1/quiz/delete/be582c77-da03-4271-929c-5d582056eb78", cookies=ValueStorage.cookies
)
assert resp.status_code == 404
resp = test_client.delete(
"/api/v1/quiz/delete/be582c77-da03-sdaasdadsasddas4271-929c-5d582056eb78", cookies={"access_token": token}
"/api/v1/quiz/delete/be582c77-da03-sdaasdadsasddas4271-929c-5d582056eb78", cookies=ValueStorage.cookies
)
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_delete_user(self, test_client): # noqa : F811
resp = test_client.post(
"/api/v1/users/token/cookie", data={"username": test_user_email, "password": test_user_password}
)
token = resp.cookies["access_token"]
data = {"password": test_user_password}
resp = test_client.delete("/api/v1/users/me", cookies={"access_token": token}, json=data)
assert resp.status_code == 200
# @pytest.mark.asyncio
# async def test_delete_user(self, test_client: TestClient): # noqa : F811
# data = {"password": test_user_password}
# resp = test_client.delete("/api/v1/users/me", cookies=ValueStorage.cookies, json=data)
# assert resp.status_code == 200
classquiz/tests/test_storage.py
+32 -22
View File
@@ -1,6 +1,7 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import io
import pytest
@@ -16,41 +17,50 @@ def test_storage_init():
with pytest.raises(NotImplementedError):
Storage(
backend="asdsad",
deta_key=settings.deta_project_key,
deta_id=settings.deta_project_id,
storage_path=settings.storage_path,
)
with pytest.raises(ValueError):
Storage(backend="deta", deta_key=None, deta_id=None, storage_path=None)
Storage(backend="s3", base_url=None, secret_key=None, access_key=None, storage_path=None)
with pytest.raises(ValueError):
Storage(backend="local", storage_path=None, deta_key=None, deta_id=None)
Storage(backend="local", storage_path=None)
async def storage_tester(storage: Storage):
res = await storage.upload(file_name="test.txt", file_data=file_contents)
res = await storage.upload(file_name="test.txt", file_data=io.BytesIO(initial_bytes=file_contents))
assert res is None
res = await storage.download(file_name="test.txt")
assert res.read() == file_contents
res = await storage.delete(file_names=["test.txt"])
assert res is None
res = await storage.download(file_name="test.txt")
res = storage.download(file_name="test.txt")
async for chunk in res:
assert bytes(chunk) == file_contents
res = await storage.get_file_size(file_name="test.txt")
assert res == len(file_contents)
res = await storage.get_file_size(file_name="asdsadasdasdadfdsf.txt")
assert res is None
res = await storage.delete(file_names=["test.txt"])
assert res is None
@pytest.mark.asyncio
async def test_deta():
storage: Storage = Storage(
backend="deta",
deta_key=settings.deta_project_key,
deta_id=settings.deta_project_id,
storage_path=settings.storage_path,
)
await storage_tester(storage)
res = storage.download(file_name="test.txt")
async for chunk in res:
assert chunk is None
res = await storage.delete(file_names=["test.txt"])
assert res is None
@pytest.mark.asyncio
async def test_local():
storage: Storage = Storage(backend="local", storage_path=settings.storage_path, deta_key=None, deta_id=None)
storage: Storage = Storage(backend="local", storage_path=settings.storage_path)
await storage_tester(storage)
@pytest.mark.asyncio
async def test_minio():
storage: Storage = Storage(
backend="s3",
access_key="Q3AM3UQ867SPQQA43P2F",
secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
bucket_name="classquiz",
base_url="https://play.min.io",
storage_path=None,
)
await storage_tester(storage)
await storage.upload(file_name="test.txt", file_data=io.BytesIO(initial_bytes=file_contents))
url = await storage.get_url(file_name="test.txt", expiry=20)
assert url is not None
classquiz/worker/__init__.py
+28
View File
@@ -0,0 +1,28 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
from arq import cron
from arq.connections import RedisSettings
from classquiz import settings
from classquiz.db import database
from classquiz.worker.storage import clean_editor_images_up, calculate_hash, quiz_update
async def startup(ctx):
ctx["db"] = database
if not ctx["db"].is_connected:
await ctx["db"].connect()
async def shutdown(ctx):
if ctx["db"].is_connected:
await ctx["db"].disconnect()
class WorkerSettings:
functions = [calculate_hash, quiz_update]
cron_jobs = [cron(clean_editor_images_up, hour={0, 6, 12, 18}, minute=0)]
on_startup = startup
on_shutdown = shutdown
redis_settings = RedisSettings.from_dsn(settings.redis)
classquiz/worker/storage.py
+108
View File
@@ -0,0 +1,108 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
import uuid
import ormar.exceptions
from arq.worker import Retry
import xxhash
from classquiz.config import redis, storage
from tempfile import SpooledTemporaryFile
from classquiz.db.models import StorageItem, Quiz, User
from classquiz.helpers import extract_image_ids_from_quiz
from classquiz.storage.errors import DeletionFailedError
from thumbhash import image_to_thumbhash
# skipcq: PYL-W0613
async def clean_editor_images_up(ctx):
print("Cleaning images up")
edit_sessions = await redis.smembers("edit_sessions")
for session_id in edit_sessions:
session = await redis.get(f"edit_session:{session_id}")
if session is None:
images = await redis.lrange(f"edit_session:{session_id}:images", 0, 3000)
if len(images) != 0:
try:
await storage.delete(images)
except DeletionFailedError:
print("Deletion Error", images)
await redis.srem("edit_sessions", session_id)
await redis.delete(f"edit_session:{session_id}:images")
async def calculate_hash(ctx, file_id_as_str: str):
file_id = uuid.UUID(file_id_as_str)
file_data: StorageItem = await StorageItem.objects.select_related(StorageItem.user).get(id=file_id)
file_path = file_id.hex
if file_data.storage_path is not None:
file_path = file_data.storage_path
file = SpooledTemporaryFile()
file_data.size = await storage.get_file_size(file_name=file_path)
if file_data.size is None:
file_data.size = 0
file_bytes = storage.download(file_path)
if file_bytes is None:
print("Retry raised!")
raise Retry(defer=ctx["job_try"] * 10)
async for chunk in file_bytes:
file.write(chunk)
try:
if 0 < file_data.size < 20_970_000: # greater than 0 but smaller than 20mbytes
file_data.thumbhash = image_to_thumbhash(file)
# skipcq: PYL-W0703
except Exception:
pass
hash_obj = xxhash.xxh3_128()
# skipcq: PY-W0069
# assert hash_obj.block_size == 64
while chunk := file.read(6400):
hash_obj.update(chunk)
file_data.hash = hash_obj.digest()
await file_data.update()
file.close()
user: User | None = await User.objects.get_or_none(id=file_data.user.id)
if user is None:
return
user.storage_used += file_data.size
await user.update()
# skipcq: PYL-W0613
async def quiz_update(ctx, old_quiz: Quiz, quiz_id: uuid.UUID):
new_quiz: Quiz = await Quiz.objects.get(id=quiz_id)
old_images = extract_image_ids_from_quiz(old_quiz)
new_images = extract_image_ids_from_quiz(new_quiz)
# If images are identical, then return
if sorted(old_images) == sorted(new_images):
print("Nothing's changed")
return
print("Change detected")
removed_images = list(set(old_images) - set(new_images))
added_images = list(set(new_images) - set(old_images))
change_made = False
for image in removed_images:
if "--" in image:
await storage.delete([image])
else:
item = await StorageItem.objects.get_or_none(id=uuid.UUID(image))
if item is None:
continue
try:
await new_quiz.storageitems.remove(item)
except ormar.exceptions.NoMatch:
continue
change_made = True
for image in added_images:
if "--" not in image:
item = await StorageItem.objects.get_or_none(id=uuid.UUID(image))
if item is None:
continue
await new_quiz.storageitems.add(item)
change_made = True
if change_made:
await new_quiz.update()