Add files via upload

app/store/admin/accessor.py

@@ -2,8 +2,9 @@
 import typing
 from hashlib import sha256
 from typing import Optional
-
+from sqlalchemy import select
 from app.base.base_accessor import BaseAccessor
+from app.admin.models import Admin, AdminModel
 from app.admin.models import Admin
 import bcrypt
 
@@ -12,21 +13,27 @@
 
 
 class AdminAccessor(BaseAccessor):
-    async def connect(self, app: "Application"):
-        first_admin = await self.create_admin(app.config.admin.email, app.config.admin.password)
-        self.app.database.admins.append(first_admin)
-
-    async def get_by_email(self, email: str) -> Optional[Admin]:
-        for admin in self.app.database.admins:
-            if admin.email == email:
-                return admin
-        return None
+    async def get_by_email(self, email: str) -> Admin | None:
+        async with self.app.database.session() as session:
+            admin = (await session.execute(
+                select(AdminModel)
+                .where(AdminModel.email == email)
+            )).scalar()
+        if not admin:
+            return
+        return admin.to_dc()
 
     async def create_admin(self, email: str, password: str) -> Admin:
-        return Admin(self.app.database.next_admins_id, email, self._password_hasher(password))
-
+        async with self.app.database.session.begin() as session:
+            new_admin = AdminModel(email=email, password=self._password_hasher(password))
+            session.add(new_admin)
+            return new_admin.to_dc()
+
     @staticmethod
     def _password_hasher(raw_password: str) -> str:
         hash_binary = bcrypt.hashpw(raw_password.encode('utf-8'), bcrypt.gensalt())
         encoded = base64.b64encode(hash_binary)
         return encoded.decode('utf-8')
+
+
+

app/web/config.py

@@ -24,27 +24,38 @@ class BotConfig:
     group_id: int
 
 
+@dataclass
+class DatabaseConfig:
+    host: str = "localhost"
+    port: int = 5432
+    user: str = "postgres"
+    password: str = "postgres"
+    database: str = "project"
+
+
 @dataclass
 class Config:
     admin: AdminConfig
-    session: SessionConfig
+    session: SessionConfig = None
     bot: BotConfig = None
+    database: DatabaseConfig = None
 
 
 def setup_config(app: "Application", config_path: str):
     with open(config_path, "r") as f:
         raw_config = yaml.safe_load(f)
 
     app.config = Config(
+        session=SessionConfig(
+            key=raw_config["session"]["key"],
+        ),
         admin=AdminConfig(
             email=raw_config["admin"]["email"],
             password=raw_config["admin"]["password"],
         ),
-        session=SessionConfig(
-            key=raw_config['session']['key']
-        ),
         bot=BotConfig(
-            token=raw_config['bot']['token'],
-            group_id=raw_config['bot']['group_id']
-        )
+            token=raw_config["bot"]["token"],
+            group_id=raw_config["bot"]["group_id"],
+        ),
+        database=DatabaseConfig(**raw_config["database"]),
     )

app/web/mixins.py

@@ -1,2 +1,9 @@
+from aiohttp.abc import StreamResponse
+from aiohttp.web_exceptions import HTTPUnauthorized
+
+
 class AuthRequiredMixin:
-    raise NotImplementedError
+    async def _iter(self) -> StreamResponse:
+        if not getattr(self.request, "admin", None):
+            raise HTTPUnauthorized
+        return await super()._iter()

tests/bot/test_bot_manager.py

@@ -1,4 +1,4 @@
-from app.store.vk_api.dataclasses import Update, UpdateObject, Message, UpdateMessage
+from app.store.vk_api.dataclasses import Message, Update, UpdateObject
 
 
 class TestHandleUpdates:
@@ -12,17 +12,14 @@ async def test_new_message(self, store):
                 Update(
                     type="message_new",
                     object=UpdateObject(
-                        message=UpdateMessage(
-                            id=1,
-                            from_id=1,
-                            text="kek",
-                        ),
-
+                        id=1,
+                        user_id=1,
+                        body="kek",
                     ),
                 )
             ]
         )
         assert store.vk_api.send_message.call_count == 1
         message: Message = store.vk_api.send_message.mock_calls[0].args[0]
         assert message.user_id == 1
-        assert message.text
+        assert message.text

tests/fixtures/common.py

@@ -1,15 +1,22 @@
+import logging
 import os
+from hashlib import sha256
 from unittest.mock import AsyncMock
+
 import pytest
 from aiohttp.test_utils import TestClient, loop_context
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.admin.models import Admin, AdminModel
+from app.store import Database
 from app.store import Store
 from app.web.app import setup_app
 from app.web.config import Config
 
 
 @pytest.fixture(scope="session")
-def loop():
+def event_loop():
     with loop_context() as _loop:
         yield _loop
 
@@ -25,8 +32,11 @@ def server():
     app.on_shutdown.clear()
     app.store.vk_api = AsyncMock()
     app.store.vk_api.send_message = AsyncMock()
-    app.on_startup.append(app.store.admins.connect)
-    app.on_shutdown.append(app.store.admins.connect)
+
+    app.database = Database(app)
+    app.on_startup.append(app.database.connect)
+    app.on_shutdown.append(app.database.disconnect)
+
     return app
 
 
@@ -35,9 +45,26 @@ def store(server) -> Store:
     return server.store
 
 
+@pytest.fixture
+def db_session(server):
+    return server.database.session
+
+
 @pytest.fixture(autouse=True, scope="function")
-def clear_db(server):
-    server.database.clear()
+async def clear_db(server):
+    yield
+    try:
+        session = AsyncSession(server.database._engine)
+        connection = session.connection()
+        for table in server.database._db.metadata.tables:
+            await session.execute(text(f"TRUNCATE {table} CASCADE"))
+            await session.execute(text(f"ALTER SEQUENCE {table}_id_seq RESTART WITH 1"))
+
+        await session.commit()
+        connection.close()
+
+    except Exception as err:
+        logging.warning(err)
 
 
 @pytest.fixture
@@ -46,8 +73,8 @@ def config(server) -> Config:
 
 
 @pytest.fixture(autouse=True)
-def cli(aiohttp_client, loop, server) -> TestClient:
-    return loop.run_until_complete(aiohttp_client(server))
+def cli(aiohttp_client, event_loop, server) -> TestClient:
+    return event_loop.run_until_complete(aiohttp_client(server))
 
 
 @pytest.fixture
@@ -60,3 +87,15 @@ async def authed_cli(cli, config) -> TestClient:
         },
     )
     yield cli
+
+
+@pytest.fixture(autouse=True)
+async def admin(cli, db_session, config: Config) -> Admin:
+    new_admin = AdminModel(
+        email=config.admin.email,
+        password=sha256(config.admin.password.encode()).hexdigest(),
+    )
+    async with db_session.begin() as session:
+        session.add(new_admin)
+
+    return Admin(id=new_admin.id, email=new_admin.email)

tests/fixtures/quiz.py

@@ -1,53 +1,111 @@
 import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.quiz.models import Theme, Question, Answer
+from app.quiz.models import (
+    Answer,
+    AnswerModel,
+    Question,
+    QuestionModel,
+    Theme,
+    ThemeModel,
+)
 
 
 @pytest.fixture
-async def theme_1(store) -> Theme:
-    theme = await store.quizzes.create_theme(title="web-development")
-    yield theme
+def answers(store) -> list[Answer]:
+    return [
+        Answer(title="1", is_correct=True),
+        Answer(title="2", is_correct=False),
+        Answer(title="3", is_correct=False),
+        Answer(title="4", is_correct=False),
+    ]
 
 
 @pytest.fixture
-async def theme_2(store) -> Theme:
-    theme = await store.quizzes.create_theme(title="backend")
-    yield theme
+async def theme_1(db_session: AsyncSession) -> Theme:
+    title = "web-development"
+    new_theme = ThemeModel(title=title)
+    async with db_session.begin() as session:
+        session.add(new_theme)
+
+    return Theme(id=new_theme.id, title=title)
+
+
+@pytest.fixture
+async def theme_2(db_session: AsyncSession) -> Theme:
+    title = "backend"
+    new_theme = ThemeModel(title=title)
+    async with db_session.begin() as session:
+        session.add(new_theme)
+
+    return Theme(id=new_theme.id, title=title)
 
 
 @pytest.fixture
-async def question_1(store, theme_1) -> Question:
-    question = await store.quizzes.create_question(
-        title="how are you?",
+async def question_1(db_session, theme_1: Theme) -> Question:
+    title = "how are you?"
+    async with db_session.begin() as session:
+        question = QuestionModel(
+            title=title,
+            theme_id=theme_1.id,
+            answers=[
+                AnswerModel(
+                    title="well",
+                    is_correct=True,
+                ),
+                AnswerModel(
+                    title="bad",
+                    is_correct=False,
+                ),
+            ],
+        )
+
+        session.add(question)
+
+    return Question(
+        id=question.id,
+        title=title,
         theme_id=theme_1.id,
         answers=[
             Answer(
-                title="well",
-                is_correct=True,
-            ),
-            Answer(
-                title="bad",
-                is_correct=False,
-            ),
+                title=a.title,
+                is_correct=a.is_correct,
+            )
+            for a in question.answers
         ],
     )
-    yield question
 
 
 @pytest.fixture
-async def question_2(store, theme_1) -> Question:
-    question = await store.quizzes.create_question(
-        title="are you doing fine?",
+async def question_2(db_session, theme_1: Theme) -> Question:
+    title = "are you doing fine?"
+    async with db_session.begin() as session:
+        question = QuestionModel(
+            title=title,
+            theme_id=theme_1.id,
+            answers=[
+                AnswerModel(
+                    title="yep",
+                    is_correct=True,
+                ),
+                AnswerModel(
+                    title="nop",
+                    is_correct=False,
+                ),
+            ],
+        )
+
+        session.add(question)
+
+    return Question(
+        id=question.id,
+        title=question.title,
         theme_id=theme_1.id,
         answers=[
             Answer(
-                title="yep",
-                is_correct=True,
-            ),
-            Answer(
-                title="nop",
-                is_correct=False,
-            ),
+                title=a.title,
+                is_correct=a.is_correct,
+            )
+            for a in question.answers
         ],
     )
-    yield question

tests/managers/test_login.py

@@ -1,8 +1,9 @@
+from app.store import Store
 from tests.utils import ok_response
 
 
 class TestAdminLoginView:
-    async def test_create_on_startup(self, store, config):
+    async def test_create_on_startup(self, store: Store, config):
         admin = await store.admins.get_by_email(config.admin.email)
         assert admin is not None
         assert admin.email == config.admin.email