diff --git a/services_backend/models/database.py b/services_backend/models/database.py index 6568f41..db1c887 100644 --- a/services_backend/models/database.py +++ b/services_backend/models/database.py @@ -1,14 +1,20 @@ from __future__ import annotations +import logging from enum import Enum +from fastapi_sqlalchemy import db from sqlalchemy import Enum as DbEnum from sqlalchemy import ForeignKey, Integer, String +from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy.orm import Mapped, mapped_column, relationship from .base import Base +logger = logging.getLogger(__name__) + + class Category(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True) order: Mapped[int] = mapped_column(Integer, default=1) @@ -17,7 +23,28 @@ class Category(Base): buttons: Mapped[list[Button]] = relationship( "Button", back_populates="category", foreign_keys="Button.category_id", order_by='Button.order' ) - scopes: Mapped[list[Scope]] = relationship("Scope", back_populates="category") + + _scopes: Mapped[list[Scope]] = relationship("Scope", back_populates="category", lazy='joined', cascade='delete') + + @hybrid_property + def scopes(self) -> set[str]: + return set(s.name for s in self._scopes) + + @scopes.inplace.setter + def _scopes_setter(self, value: set[str]): + old_scopes = self.scopes + new_scopes = set(value) + + # Удаляем более ненужные скоупы + for s in self._scopes: + if s.name in (old_scopes - new_scopes): + db.session.delete(s) + + # Добавляем недостающие скоупы + for s in new_scopes - old_scopes: + new_scope = Scope(category=self, name=s) + db.session.add(new_scope) + self._scopes.append(new_scope) class Type(str, Enum): @@ -41,4 +68,4 @@ class Scope(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String, nullable=True) category_id: Mapped[int] = mapped_column(Integer, ForeignKey("category.id")) - category: Mapped[Category] = relationship("Category", back_populates="scopes", foreign_keys=[category_id]) + category: Mapped[Category] = relationship("Category", back_populates="_scopes", foreign_keys=[category_id]) diff --git a/services_backend/routes/base.py b/services_backend/routes/base.py index 67697cc..4435393 100644 --- a/services_backend/routes/base.py +++ b/services_backend/routes/base.py @@ -7,7 +7,9 @@ from .button import button from .category import category -from .scope import scope + + +# from .scope import scope settings = get_settings() @@ -38,4 +40,3 @@ app.include_router(button, prefix='/category/{category_id}/button', tags=["Button"]) app.include_router(category, prefix='/category', tags=["Category"]) -app.include_router(scope, prefix='/category/{category_id}/scope', tags=["Scope"]) diff --git a/services_backend/routes/button.py b/services_backend/routes/button.py index f7eb7e1..8faa3b1 100644 --- a/services_backend/routes/button.py +++ b/services_backend/routes/button.py @@ -3,17 +3,51 @@ from auth_lib.fastapi import UnionAuth from fastapi import APIRouter, Depends, HTTPException from fastapi_sqlalchemy import db +from pydantic import Field -from ..models.database import Button, Category -from .models.button import ButtonCreate, ButtonGet, ButtonUpdate -from .models.category import CategoryGet +from services_backend.models.database import Button, Category, Type +from services_backend.schemas import Base logger = logging.getLogger(__name__) button = APIRouter() +# region schemas -@button.post("/", response_model=ButtonGet) + +class ButtonCreate(Base): + icon: str = Field(description='Иконка кнопки') + name: str = Field(description='Название кнопки') + link: str = Field(description='Ссылка, на которую перенаправляет кнопка') + type: Type = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер') + + +class ButtonUpdate(Base): + category_id: int | None = Field(description='Айди категории') + icon: str | None = Field(description='Иконка кнопки') + name: str | None = Field(description='Название кнопки') + order: int | None = Field(description='Порядок, в котором отображаются кнопки') + link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка') + type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер') + + +class ButtonGet(Base): + id: int = Field(description='Айди кнопки') + icon: str | None = Field(description='Иконка кнопки') + name: str | None = Field(description='Название кнопки') + link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка') + order: int | None = Field(description='Порядок, в котором отображаются кнопки') + type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер') + + +class ButtonsGet(Base): + buttons: list[ButtonGet] | None + + +# endregion + + +@button.post("", response_model=ButtonGet) def create_button( button_inp: ButtonCreate, category_id: int, @@ -39,7 +73,7 @@ def create_button( return button -@button.get("/", response_model=CategoryGet) +@button.get("", response_model=ButtonsGet) def get_buttons( category_id: int, user=Depends(UnionAuth(allow_none=True, auto_error=False)), diff --git a/services_backend/routes/category.py b/services_backend/routes/category.py index 8a0744d..e8db5af 100644 --- a/services_backend/routes/category.py +++ b/services_backend/routes/category.py @@ -4,17 +4,56 @@ from auth_lib.fastapi import UnionAuth from fastapi import APIRouter, Depends, HTTPException, Query from fastapi_sqlalchemy import db -from sqlalchemy.orm import joinedload +from pydantic import Field -from ..models.database import Button, Category -from .models.category import CategoryCreate, CategoryGet, CategoryUpdate +from services_backend.models.database import Button, Category, Type +from services_backend.schemas import Base logger = logging.getLogger(__name__) category = APIRouter() -@category.post("/", response_model=CategoryGet) +# region schemas + + +class ButtonGet(Base): + id: int = Field(description='Айди кнопки') + icon: str | None = Field(description='Иконка кнопки') + name: str | None = Field(description='Название кнопки') + link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка') + order: int | None = Field(description='Порядок, в котором отображаются кнопки') + type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер') + + +class CategoryCreate(Base): + type: str = Field(description='Тип отображения категории') + name: str = Field(description='Имя категории') + scopes: set[str] | None = Field(description='Каким пользователям будет видна категория') + + +class CategoryUpdate(Base): + order: int | None = Field(description='На какую позицию перенести категорию') + type: str | None = Field(description='Тип отображения категории') + name: str | None = Field(description='Имя категории') + scopes: set[str] | None = Field(description='Каким пользователям будет видна категория') + + +class CategoryGet(Base): + id: int + order: int + type: str | None + name: str | None + buttons: list[ButtonGet] | None + scopes: list[str] | None + + +# endregion + +# region routes + + +@category.post("", response_model=CategoryGet) def create_category( category_inp: CategoryCreate, user=Depends(UnionAuth(['services.category.create'])), @@ -25,15 +64,19 @@ def create_category( """ logger.info(f"User {user.get('id')} triggered create_category") last_category = db.session.query(Category).order_by(Category.order.desc()).first() + scopes = category_inp.scopes + category_inp.scopes = None category = Category(**category_inp.dict(exclude_none=True)) + if scopes is not None: + category.scopes = scopes if last_category: category.order = last_category.order + 1 db.session.add(category) - db.session.flush() + db.session.commit() return category -@category.get("/", response_model=list[CategoryGet], response_model_exclude_none=True) +@category.get("", response_model=list[CategoryGet], response_model_exclude_none=True) def get_categories( info: list[Literal['buttons']] = Query([]), user=Depends(UnionAuth(allow_none=True, auto_error=False)), @@ -50,9 +93,9 @@ def get_categories( user_scopes = set([scope["name"] for scope in user["session_scopes"]] if user else []) filtered_categories = [] - for category in db.session.query(Category).order_by(Category.order).options(joinedload(Category.scopes)).all(): - category_scopes = set([scope.__dict__["name"] for scope in category.scopes]) - if (category_scopes == set()) or (user_scopes & category_scopes): + for category in db.session.query(Category).order_by(Category.order).all(): + category_scopes = set(category.scopes) + if (category_scopes == set()) or len(category_scopes - user_scopes) != 0: filtered_categories.append(category) return [ @@ -78,9 +121,7 @@ def get_category( user_scopes = set([scope["name"] for scope in user["session_scopes"]] if user else []) category = db.session.query(Category).filter(Category.id == category_id).one_or_none() - if not category or ( - category.scopes and not (user_scopes & set([scope.__dict__["name"] for scope in category.scopes])) - ): + if not category or (category.scopes and len(category.scopes - user_scopes) != 0): raise HTTPException(status_code=404, detail="Category does not exist") return { "id": category_id, @@ -123,14 +164,17 @@ def update_category( Необходимые scopes: `services.category.update` """ + logger.info(f"User {user.get('id')} triggered update_category") category = db.session.query(Category).filter(Category.id == category_id).one_or_none() last_category = db.session.query(Category).order_by(Category.order.desc()).first() if not category: raise HTTPException(status_code=404, detail="Category does not exist") - if not any(category_inp.dict().values()): - raise HTTPException(status_code=400, detail="Empty schema") + + if category_inp.scopes is not None: + category.scopes = category_inp.scopes + db.session.flush() if category_inp.order: if category_inp.order < 1: @@ -148,6 +192,11 @@ def update_category( db.session.query(Category).filter(Category.order > category.order).update({"order": Category.order - 1}) query = db.session.query(Category).filter(Category.id == category_id) - query.update(category_inp.dict(exclude_unset=True, exclude_none=True)) - db.session.flush() + update_values = category_inp.dict(exclude_unset=True, exclude_none=True, exclude={'scopes': True}) + if update_values: + query.update(update_values) + db.session.commit() return category + + +# endregion diff --git a/services_backend/routes/models/__init__.py b/services_backend/routes/models/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/services_backend/routes/models/button.py b/services_backend/routes/models/button.py deleted file mode 100644 index 33c8500..0000000 --- a/services_backend/routes/models/button.py +++ /dev/null @@ -1,30 +0,0 @@ -from pydantic import Field - -from services_backend.models.database import Type - -from .base import Base - - -class ButtonCreate(Base): - icon: str = Field(description='Иконка кнопки') - name: str = Field(description='Название кнопки') - link: str = Field(description='Ссылка, на которую перенаправляет кнопка') - type: Type = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер') - - -class ButtonUpdate(Base): - category_id: int | None = Field(description='Айди категории') - icon: str | None = Field(description='Иконка кнопки') - name: str | None = Field(description='Название кнопки') - order: int | None = Field(description='Порядок, в котором отображаются кнопки') - link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка') - type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер') - - -class ButtonGet(Base): - id: int = Field(description='Айди кнопки') - order: int = Field(description='Порядок, в котором отображаются кнопки') - icon: str | None = Field(description='Иконка кнопки') - name: str | None = Field(description='Название кнопки') - link: str | None = Field(description='Ссылка, на которую перенаправляет кнопка') - type: Type | None = Field(description='Тип открываемой ссылки (Ссылка приложения/Браузер в приложении/Браузер') diff --git a/services_backend/routes/models/category.py b/services_backend/routes/models/category.py deleted file mode 100644 index 6c8774f..0000000 --- a/services_backend/routes/models/category.py +++ /dev/null @@ -1,25 +0,0 @@ -from pydantic import Field - -from .base import Base -from .button import ButtonGet -from .scope import ScopeGet - - -class CategoryCreate(Base): - type: str = Field(description='Тип отображения категории') - name: str = Field(description='Имя категории') - - -class CategoryUpdate(Base): - order: int | None - type: str | None - name: str | None - - -class CategoryGet(Base): - id: int - order: int - type: str | None - name: str | None - buttons: list[ButtonGet] | None - scopes: list[ScopeGet] | None diff --git a/services_backend/routes/models/scope.py b/services_backend/routes/models/scope.py deleted file mode 100644 index 438698f..0000000 --- a/services_backend/routes/models/scope.py +++ /dev/null @@ -1,10 +0,0 @@ -from .base import Base - - -class ScopeCreate(Base): - name: str - - -class ScopeGet(Base): - id: int - name: str diff --git a/services_backend/routes/scope.py b/services_backend/routes/scope.py deleted file mode 100644 index f91026c..0000000 --- a/services_backend/routes/scope.py +++ /dev/null @@ -1,33 +0,0 @@ -from auth_lib.fastapi import UnionAuth -from fastapi import APIRouter, Depends, HTTPException -from fastapi_sqlalchemy import db - -from ..models.database import Scope -from .models.scope import ScopeCreate, ScopeGet - - -scope = APIRouter() - - -@scope.post("/", response_model=ScopeGet) -def create_scope( - scope_inp: ScopeCreate, category_id: int, user=Depends(UnionAuth(['services.category.permission.create'])) -): - scope = Scope(**{"name": scope_inp.name, "category_id": category_id}) - db.session.add(scope) - db.session.flush() - return scope - - -@scope.get("/", response_model=list[ScopeGet]) -def get_scopes(category_id: int, offset: int = 0, limit: int = 100): - return db.session.query(Scope).filter(category_id == Scope.category_id).offset(offset).limit(limit).all() - - -@scope.delete("/{scope_id}", response_model=None) -def delete_scope(category_id: int, scope_id: int, user=Depends(UnionAuth(['services.category.permission.delete']))): - scope = db.session.query(Scope).filter(category_id == Scope.category_id).filter(Scope.id == scope_id).one_or_none() - if not scope: - raise HTTPException(status_code=404, detail="Scope does not exist") - db.session.delete(scope) - db.session.flush() diff --git a/services_backend/routes/models/base.py b/services_backend/schemas.py similarity index 100% rename from services_backend/routes/models/base.py rename to services_backend/schemas.py diff --git a/tests/api/button.py b/tests/api/button.py index e2d667d..1655b09 100644 --- a/tests/api/button.py +++ b/tests/api/button.py @@ -6,160 +6,173 @@ from services_backend.settings import get_settings -class TestButton: - settings = get_settings() - - def test_get_success(self, client, db_button, db_category): - res = client.get(f"/category/{db_category.id}/button/{db_button.id}") - assert res.status_code == status.HTTP_200_OK - assert res.json()['id'] == db_button.id - - def test_post_success(self, client, db_category, dbsession): - body = { - "icon": "https://lh3.googleusercontent.com/yURn6ISxDySTdXZAW2PUcADMnU3y9YX0M1RyXOH8a3sa1Tr0pHhPLGw5BKuiLiXa3Eh0fyHm7Dfsd9FodK3fxJge6g=w640-h400-e365-rj-sc0x00ffffff", - "name": "string", - "link": "google.com", - "type": "inapp", - } - res = client.post(f"/category/{db_category.id}/button/", data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert res_body["icon"] == body["icon"] - assert res_body["order"] == 1 - assert res_body["name"] == body["name"] - assert res_body["link"] == body["link"] - assert res_body["type"] == body["type"] - db_button_created: Button = dbsession.query(Button).filter(Button.id == res_body["id"]).one_or_none() - assert db_button_created - assert db_button_created.icon == body["icon"] - assert db_button_created.name == body["name"] - assert db_button_created.category == db_category - assert db_button_created.link == body["link"] - assert db_button_created.type == body["type"] - assert db_button_created.order == 1 - - def test_get_by_id_success(self, client, db_button, db_category): - res = client.get(f"/category/{db_category.id}/button/{db_button.id}") - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert res_body['icon'] == db_button.icon - assert res_body['name'] == db_button.name - assert res_body['order'] == db_button.order - assert res_body['link'] == db_button.link - assert res_body['type'] == db_button.type - - def test_delete_by_id_success(self, client, dbsession, db_button, db_category): - res = client.delete(f"/category/{db_category.id}/button/{db_button.id}") - assert res.status_code == status.HTTP_200_OK - q = dbsession.query(Button).filter(Button.id == db_button.id) - assert not q.one_or_none() - get_res = client.get(f"/category/{db_category.id}/button/{db_button.id}") - assert get_res.status_code == status.HTTP_404_NOT_FOUND - - def test_patch_by_id_success(self, db_button, client, db_category): - body = {"icon": "cool icon", "name": "nice name", "order": 2, "link": "ya.ru", "type": "inapp"} - res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert res_body["icon"] == body["icon"] - assert res_body["order"] == body["order"] - assert res_body["name"] == body["name"] - assert res_body["link"] == body["link"] - assert res_body["type"] == body["type"] - - def test_patch_unset_params(self, client, db_button, db_category): - body = {} - res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body)) - assert res.status_code == status.HTTP_400_BAD_REQUEST - body["icon"] = "string" - body["order"] = 1 - res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - assert res.json()["icon"] == body["icon"] - body_name = {"name": "string", "order": 1} - res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body_name)) - assert res.status_code == status.HTTP_200_OK - assert res.json()["name"] == body_name["name"] - - def test_get_by_id_not_found(self, client, db_button, db_category): - res = client.get(f"/category/{db_category.id}/button/{db_button.id + 1}") - assert res.status_code == status.HTTP_404_NOT_FOUND - - def test_delete_by_id_not_found(self, client, db_button, db_category): - res = client.delete(f"/category/{db_category.id}/button/{db_button.id + 1}") - assert res.status_code == status.HTTP_404_NOT_FOUND - - def test_patch_by_id_not_found(self, client, db_button, db_category): - body = {"icon": "cool icon", "name": "nice name"} - res = client.patch(f"/category/{db_category.id}/button/{db_button.id + 1}", data=json.dumps(body)) - assert res.status_code == status.HTTP_404_NOT_FOUND - - def test_create_first(self, client, db_button, db_category): - body = { - "icon": "test", - "name": "test", - "link": "test", - "type": "inapp", - } - - res = client.post(f"/category/{db_category.id}/button/", data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - - res = client.patch(f"/category/{db_category.id}/button/{res.json()['id']}", data=json.dumps({"order": 1})) - assert res.json()["order"] == 1 - res_old = client.get(f"/category/{db_category.id}/button/{db_button.id}") - assert res_old.json()["order"] == 2 - - def test_patch_order_fail(self, client, db_button, db_category): - body = { - "icon": "test", - "name": "new", - "link": "test", - "type": "inapp", - } - res1 = client.post(f"/category/{db_category.id}/button/", data=json.dumps(body)) - assert res1.status_code == status.HTTP_200_OK - - body_patch = { - "name": db_button.name, - "order": 44, - } - res = client.patch(f"/category/{db_category.id}/button/{res1.json()['id']}", data=json.dumps(body_patch)) - assert res.status_code == status.HTTP_400_BAD_REQUEST - - def test_patch_negative_order_fail(self, db_button, client, db_category): - body = { - "icon": "test", - "name": "new", - "link": "test", - "type": "inapp", - } - res = client.post(f"/category/{db_category.id}/button/", data=json.dumps(body)) - res1 = client.patch(f"/category/{db_category.id}/button/{res.json()['id']}", data=json.dumps({"order": -10})) - assert res1.status_code == status.HTTP_400_BAD_REQUEST - - def test_delete_order(self, db_button, client, db_category): - body = { - "icon": "test", - "name": "new", - "link": "test", - "type": "inapp", - } - res1 = client.post(f"/category/{db_category.id}/button/", data=json.dumps(body)) - assert res1.status_code == status.HTTP_200_OK - - res = client.delete(f"/category/{db_category.id}/button/{res1.json()['id']}") - assert res.status_code == status.HTTP_200_OK - - res = client.get(f"/category/{db_category.id}/button/{db_button.id}") - assert res.json()['order'] == 1 - - def test_type_not_enum(self, client, dbsession, db_category): - body = { - "icon": "test", - "name": "new", - "link": "test", - "type": "lmao", - } - res = client.post(f"/category/{db_category.id}/button/", data=json.dumps(body)) - assert res.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY +settings = get_settings() + + +def test_get_success(client, db_button, db_category): + res = client.get(f"/category/{db_category.id}/button/{db_button.id}") + assert res.status_code == status.HTTP_200_OK + assert res.json()['id'] == db_button.id + + +def test_post_success(client, db_category, dbsession): + body = { + "icon": "https://lh3.googleusercontent.com/yURn6ISxDySTdXZAW2PUcADMnU3y9YX0M1RyXOH8a3sa1Tr0pHhPLGw5BKuiLiXa3Eh0fyHm7Dfsd9FodK3fxJge6g=w640-h400-e365-rj-sc0x00ffffff", + "name": "string", + "link": "google.com", + "type": "inapp", + } + res = client.post(f"/category/{db_category.id}/button", data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert res_body["icon"] == body["icon"] + assert res_body["order"] == 1 + assert res_body["name"] == body["name"] + assert res_body["link"] == body["link"] + assert res_body["type"] == body["type"] + db_button_created: Button = dbsession.query(Button).filter(Button.id == res_body["id"]).one_or_none() + assert db_button_created + assert db_button_created.icon == body["icon"] + assert db_button_created.name == body["name"] + assert db_button_created.category == db_category + assert db_button_created.link == body["link"] + assert db_button_created.type == body["type"] + assert db_button_created.order == 1 + + +def test_get_by_id_success(client, db_button, db_category): + res = client.get(f"/category/{db_category.id}/button/{db_button.id}") + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert res_body['icon'] == db_button.icon + assert res_body['name'] == db_button.name + assert res_body['order'] == db_button.order + assert res_body['link'] == db_button.link + assert res_body['type'] == db_button.type + + +def test_delete_by_id_success(client, dbsession, db_button, db_category): + res = client.delete(f"/category/{db_category.id}/button/{db_button.id}") + assert res.status_code == status.HTTP_200_OK + q = dbsession.query(Button).filter(Button.id == db_button.id) + assert not q.one_or_none() + get_res = client.get(f"/category/{db_category.id}/button/{db_button.id}") + assert get_res.status_code == status.HTTP_404_NOT_FOUND + + +def test_patch_by_id_success(db_button, client, db_category): + body = {"icon": "cool icon", "name": "nice name", "order": 2, "link": "ya.ru", "type": "inapp"} + res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert res_body["icon"] == body["icon"] + assert res_body["order"] == body["order"] + assert res_body["name"] == body["name"] + assert res_body["link"] == body["link"] + assert res_body["type"] == body["type"] + + +def test_patch_unset_params(client, db_button, db_category): + body = {} + res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body)) + assert res.status_code == status.HTTP_400_BAD_REQUEST + body["icon"] = "string" + body["order"] = 1 + res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + assert res.json()["icon"] == body["icon"] + body_name = {"name": "string", "order": 1} + res = client.patch(f"/category/{db_category.id}/button/{db_button.id}", data=json.dumps(body_name)) + assert res.status_code == status.HTTP_200_OK + assert res.json()["name"] == body_name["name"] + + +def test_get_by_id_not_found(client, db_button, db_category): + res = client.get(f"/category/{db_category.id}/button/{db_button.id + 1}") + assert res.status_code == status.HTTP_404_NOT_FOUND + + +def test_delete_by_id_not_found(client, db_button, db_category): + res = client.delete(f"/category/{db_category.id}/button/{db_button.id + 1}") + assert res.status_code == status.HTTP_404_NOT_FOUND + + +def test_patch_by_id_not_found(client, db_button, db_category): + body = {"icon": "cool icon", "name": "nice name"} + res = client.patch(f"/category/{db_category.id}/button/{db_button.id + 1}", data=json.dumps(body)) + assert res.status_code == status.HTTP_404_NOT_FOUND + + +def test_create_first(client, db_button, db_category): + body = { + "icon": "test", + "name": "test", + "link": "test", + "type": "inapp", + } + + res = client.post(f"/category/{db_category.id}/button", data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + + res = client.patch(f"/category/{db_category.id}/button/{res.json()['id']}", data=json.dumps({"order": 1})) + assert res.json()["order"] == 1 + res_old = client.get(f"/category/{db_category.id}/button/{db_button.id}") + assert res_old.json()["order"] == 2 + + +def test_patch_order_fail(client, db_button, db_category): + body = { + "icon": "test", + "name": "new", + "link": "test", + "type": "inapp", + } + res1 = client.post(f"/category/{db_category.id}/button", data=json.dumps(body)) + assert res1.status_code == status.HTTP_200_OK + + body_patch = { + "name": db_button.name, + "order": 44, + } + res = client.patch(f"/category/{db_category.id}/button/{res1.json()['id']}", data=json.dumps(body_patch)) + assert res.status_code == status.HTTP_400_BAD_REQUEST + + +def test_patch_negative_order_fail(db_button, client, db_category): + body = { + "icon": "test", + "name": "new", + "link": "test", + "type": "inapp", + } + res = client.post(f"/category/{db_category.id}/button", data=json.dumps(body)) + res1 = client.patch(f"/category/{db_category.id}/button/{res.json()['id']}", data=json.dumps({"order": -10})) + assert res1.status_code == status.HTTP_400_BAD_REQUEST + + +def test_delete_order(db_button, client, db_category): + body = { + "icon": "test", + "name": "new", + "link": "test", + "type": "inapp", + } + res1 = client.post(f"/category/{db_category.id}/button", data=json.dumps(body)) + assert res1.status_code == status.HTTP_200_OK + + res = client.delete(f"/category/{db_category.id}/button/{res1.json()['id']}") + assert res.status_code == status.HTTP_200_OK + + res = client.get(f"/category/{db_category.id}/button/{db_button.id}") + assert res.json()['order'] == 1 + + +def test_type_not_enum(client, dbsession, db_category): + body = { + "icon": "test", + "name": "new", + "link": "test", + "type": "lmao", + } + res = client.post(f"/category/{db_category.id}/button", data=json.dumps(body)) + assert res.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY diff --git a/tests/api/category.py b/tests/api/category.py index 3b5fa78..b6cc116 100644 --- a/tests/api/category.py +++ b/tests/api/category.py @@ -7,228 +7,234 @@ from services_backend.settings import get_settings -class TestCategory: - _url = '/category/' - settings = get_settings() - - def test_get_success(self, client, db_category): - res = client.get(self._url) - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert len(res_body) == 1 - assert res_body[0]['id'] == db_category.id - assert res_body[0]['type'] == db_category.type - assert res_body[0]['name'] == db_category.name - assert res_body[0]['order'] == db_category.order - - def test_post_success(self, client, dbsession): - body = {"type": "string", "name": "string"} - res = client.post(self._url, data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert res_body["type"] == body["type"] - assert res_body["name"] == body["name"] - assert res_body["order"] == 1 - db_category_created: Category = dbsession.query(Category).filter(Category.name == body["name"]).one_or_none() - assert db_category_created - assert db_category_created.name == body["name"] - assert db_category_created.type == body["type"] - assert db_category_created.order == 1 - assert not db_category_created.buttons - client.delete(f'{self._url}{db_category_created.id}') - - def test_get_by_id_success(self, client, db_category, mocker: MockerFixture): - res = client.get(f'{self._url}{db_category.id}') - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert res_body['id'] == db_category.id - assert res_body['type'] == db_category.type - assert res_body['name'] == db_category.name - assert res_body['order'] == db_category.order - - user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__') - user_mock.return_value = { - "session_scopes": [ - {"id": 0, "name": "string", "comment": "string"}, - {"id": 1, "name": "test", "comment": "string"}, - ], - "user_scopes": [{"id": 0, "name": "string", "comment": "string"}], - "indirect_groups": [{"id": 0, "name": "string", "parent_id": 0}], - "groups": [{"id": 0, "name": "string", "parent_id": 0}], - "id": 0, - "email": "string", - } - - res = client.get(f'{self._url}{db_category.id}') - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert res_body['id'] == db_category.id - assert res_body['type'] == db_category.type - assert res_body['name'] == db_category.name - assert res_body['order'] == db_category.order - - def test_delete_by_id_success(self, client, dbsession, db_category): - res = client.delete(f'{self._url}{db_category.id}') - assert res.status_code == status.HTTP_200_OK - q = dbsession.query(Category).filter(Category.id == db_category.id) - assert q.one_or_none() is None - get_res = client.get(f'{self._url}{db_category.id}') - assert get_res.status_code == status.HTTP_404_NOT_FOUND - - def test_patch_by_id_success(self, client, db_category): - body = {"type": "test", "name": "test", "order": 1} - res = client.patch(f"{self._url}{db_category.id}", data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - res_body = res.json() - assert res_body['type'] == body['type'] - assert res_body['name'] == body['name'] - assert res_body['order'] == body['order'] - - def test_patch_unset_params(self, client, db_category): - body = {} - res = client.patch(f"{self._url}{db_category.id}", data=json.dumps(body)) - assert res.status_code == status.HTTP_400_BAD_REQUEST - body["order"] = 1 - res = client.patch(f"{self._url}{db_category.id}", data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - assert res.json()["order"] == body["order"] - body_ord = {"order": 1} - res = client.patch(f"{self._url}{db_category.id}", data=json.dumps(body_ord)) - assert res.status_code == status.HTTP_200_OK - assert res.json()["order"] == body_ord["order"] - - def test_get_by_id_not_found(self, client, db_category): - res = client.get(f'{self._url}/{db_category.id + 1}') - assert res.status_code == status.HTTP_404_NOT_FOUND - - def test_delete_by_id_not_found(self, client, db_category): - res = client.delete(f'{self._url}{db_category.id + 1}') - assert res.status_code == status.HTTP_404_NOT_FOUND - - def test_patch_by_id_not_found(self, client, db_category): - body = {"type": "string", "name": "string", "order": 1} - res = client.patch(f"{self._url}{db_category.id + 1}", data=json.dumps(body)) - assert res.status_code == status.HTTP_404_NOT_FOUND - - def test_create_first(self, client, db_category): - body = { - "name": "test", - "type": "test", - } - - res = client.post(self._url, data=json.dumps(body)) - assert res.status_code == status.HTTP_200_OK - res1 = client.patch(f"{self._url}{res.json()['id']}", data=json.dumps({"order": 1})) - assert res1.status_code == status.HTTP_200_OK - assert res1.json()["order"] == 1 - - res_old = client.get(f"{self._url}{db_category.id}") - assert res_old.json()["order"] == 2 - client.delete(f"{self._url}{res.json()['id']}") - - def test_patch_order(self, client, db_category): - body = { - "name": "new", - "type": "test", - } - res1 = client.post(self._url, data=json.dumps(body)) - assert res1.status_code == status.HTTP_200_OK - res = client.patch(f"{self._url}{res1.json()['id']}", data=json.dumps({"order": 1})) - assert res.status_code == status.HTTP_200_OK - assert res.json()["order"] == 1 - - res = client.get(f"{self._url}{db_category.id}") - assert res.json()["order"] == 2 - client.delete(f"{self._url}{res1.json()['id']}") - - def test_create_third_fail(self, db_category, client): - body = { - "name": "new", - "type": "test", - } - res1 = client.post(self._url, data=json.dumps(body)) - assert res1.status_code == status.HTTP_200_OK - res = client.patch(f"{self._url}{res1.json()['id']}", data=json.dumps({"order": 33})) - assert res.status_code == status.HTTP_400_BAD_REQUEST - client.delete(f"{self._url}{res1.json()['id']}") - - def test_create_negative_order_fail(self, db_category, client): - body = { - "name": "new", - "type": "test", - } - res1 = client.post(self._url, data=json.dumps(body)) - assert res1.status_code == status.HTTP_200_OK - res = client.patch(f"{self._url}{res1.json()['id']}", data=json.dumps({"order": -1})) - assert res.status_code == status.HTTP_400_BAD_REQUEST - client.delete(f"{self._url}{res1.json()['id']}") - - def test_delete_order(self, db_category, client): - body = { - "name": "new", - "type": "test", - } - res1 = client.post(self._url, data=json.dumps(body)) - assert res1.status_code == status.HTTP_200_OK - assert res1.json()['order'] == 2 - - res = client.delete(f"{self._url}{db_category.id}") - assert res.status_code == status.HTTP_200_OK - - res = client.get(f"{self._url}{res1.json()['id']}") - assert res.json()['order'] == 1 - - def test_category_scopes(self, client, dbsession, mocker: MockerFixture): - category_body = {"name": "t", "type": "string"} - res1 = client.post(self._url, data=json.dumps(category_body)) - assert res1.status_code == status.HTTP_200_OK - res1_body = res1.json() - - scope_body = {"name": "test"} - res2 = client.post(f"{self._url}{res1_body['id']}/scope/", data=json.dumps(scope_body)) - assert res2.status_code == status.HTTP_200_OK - - res = client.get(f"{self._url}{res1_body['id']}") - assert res.status_code == status.HTTP_404_NOT_FOUND - - user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__') - user_mock.return_value = { - "session_scopes": [ - {"id": 0, "name": "string", "comment": "string"}, - {"id": 1, "name": "test", "comment": "string"}, - ], - "user_scopes": [{"id": 0, "name": "string", "comment": "string"}], - "indirect_groups": [{"id": 0, "name": "string", "parent_id": 0}], - "groups": [{"id": 0, "name": "string", "parent_id": 0}], - "id": 0, - "email": "string", - } - - res = client.get(f"{self._url}{res1_body['id']}") - assert res.status_code == status.HTTP_200_OK - - def test_category_invalid_scopes(self, client, dbsession, mocker: MockerFixture): - category_body = {"name": "t", "type": "string"} - res1 = client.post(self._url, data=json.dumps(category_body)) - assert res1.status_code == status.HTTP_200_OK - res1_body = res1.json() - - scope_body = {"name": "test"} - res2 = client.post(f"{self._url}{res1_body['id']}/scope/", data=json.dumps(scope_body)) - assert res2.status_code == status.HTTP_200_OK - - user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__') - user_mock.return_value = { - "session_scopes": [ - {"id": 0, "name": "string", "comment": "string"}, - {"id": 3, "name": "lmao", "comment": "string"}, - ], - "user_scopes": [{"id": 0, "name": "string", "comment": "string"}], - "indirect_groups": [{"id": 0, "name": "string", "parent_id": 0}], - "groups": [{"id": 0, "name": "string", "parent_id": 0}], - "id": 0, - "email": "string", - } - - res = client.get(f"{self._url}{res1_body['id']}") - assert res.status_code == status.HTTP_404_NOT_FOUND +settings = get_settings() + + +def test_get_success(client, db_category): + res = client.get('/category') + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert len(res_body) == 1 + assert res_body[0]['id'] == db_category.id + assert res_body[0]['type'] == db_category.type + assert res_body[0]['name'] == db_category.name + assert res_body[0]['order'] == db_category.order + + +def test_post_success(client, dbsession): + body = {"type": "string", "name": "string"} + res = client.post('/category', data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert res_body["type"] == body["type"] + assert res_body["name"] == body["name"] + assert res_body["order"] == 1 + db_category_created: Category = dbsession.query(Category).filter(Category.name == body["name"]).one_or_none() + assert db_category_created + assert db_category_created.name == body["name"] + assert db_category_created.type == body["type"] + assert db_category_created.order == 1 + assert not db_category_created.buttons + client.delete(f'/category/{db_category_created.id}') + + +def test_get_by_id_success(client, db_category, mocker: MockerFixture): + res = client.get(f'/category/{db_category.id}') + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert res_body['id'] == db_category.id + assert res_body['type'] == db_category.type + assert res_body['name'] == db_category.name + assert res_body['order'] == db_category.order + + user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__') + user_mock.return_value = { + "session_scopes": [ + {"id": 0, "name": "string", "comment": "string"}, + {"id": 1, "name": "test", "comment": "string"}, + ], + "user_scopes": [{"id": 0, "name": "string", "comment": "string"}], + "indirect_groups": [{"id": 0, "name": "string", "parent_id": 0}], + "groups": [{"id": 0, "name": "string", "parent_id": 0}], + "id": 0, + "email": "string", + } + + res = client.get(f'/category/{db_category.id}') + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert res_body['id'] == db_category.id + assert res_body['type'] == db_category.type + assert res_body['name'] == db_category.name + assert res_body['order'] == db_category.order + + +def test_delete_by_id_success(client, dbsession, db_category): + res = client.delete(f'/category/{db_category.id}') + assert res.status_code == status.HTTP_200_OK + q = dbsession.query(Category).filter(Category.id == db_category.id) + assert q.one_or_none() is None + get_res = client.get(f'/category/{db_category.id}') + assert get_res.status_code == status.HTTP_404_NOT_FOUND + + +def test_patch_by_id_success(client, db_category): + body = {"type": "test", "name": "test", "order": 1} + res = client.patch(f"/category/{db_category.id}", data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + res_body = res.json() + assert res_body['type'] == body['type'] + assert res_body['name'] == body['name'] + assert res_body['order'] == body['order'] + + +def test_patch_unset_params(client, db_category): + body = {} + body["order"] = 1 + res = client.patch(f"/category/{db_category.id}", data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + assert res.json()["order"] == body["order"] + body_ord = {"order": 1} + res = client.patch(f"/category/{db_category.id}", data=json.dumps(body_ord)) + assert res.status_code == status.HTTP_200_OK + assert res.json()["order"] == body_ord["order"] + + +def test_get_by_id_not_found(client, db_category): + res = client.get(f'/category/{db_category.id + 1}') + assert res.status_code == status.HTTP_404_NOT_FOUND + + +def test_delete_by_id_not_found(client, db_category): + res = client.delete(f'/category/{db_category.id + 1}') + assert res.status_code == status.HTTP_404_NOT_FOUND + + +def test_patch_by_id_not_found(client, db_category): + body = {"type": "string", "name": "string", "order": 1} + res = client.patch(f"/category/{db_category.id + 1}", data=json.dumps(body)) + assert res.status_code == status.HTTP_404_NOT_FOUND + + +def test_create_first(client, db_category): + body = { + "name": "test", + "type": "test", + } + + res = client.post('/category', data=json.dumps(body)) + assert res.status_code == status.HTTP_200_OK + res1 = client.patch(f"/category/{res.json()['id']}", data=json.dumps({"order": 1})) + assert res1.status_code == status.HTTP_200_OK + assert res1.json()["order"] == 1 + + res_old = client.get(f"/category/{db_category.id}") + assert res_old.json()["order"] == 2 + client.delete(f"/category/{res.json()['id']}") + + +def test_patch_order(client, db_category): + body = { + "name": "new", + "type": "test", + } + res1 = client.post('/category', data=json.dumps(body)) + assert res1.status_code == status.HTTP_200_OK + res = client.patch(f"/category/{res1.json()['id']}", data=json.dumps({"order": 1})) + assert res.status_code == status.HTTP_200_OK + assert res.json()["order"] == 1 + + res = client.get(f"/category/{db_category.id}") + assert res.json()["order"] == 2 + client.delete(f"/category/{res1.json()['id']}") + + +def test_create_third_fail(db_category, client): + body = { + "name": "new", + "type": "test", + } + res1 = client.post('/category', data=json.dumps(body)) + assert res1.status_code == status.HTTP_200_OK + res = client.patch(f"/category/{res1.json()['id']}", data=json.dumps({"order": 33})) + assert res.status_code == status.HTTP_400_BAD_REQUEST + client.delete(f"/category/{res1.json()['id']}") + + +def test_create_negative_order_fail(db_category, client): + body = { + "name": "new", + "type": "test", + } + res1 = client.post('/category', data=json.dumps(body)) + assert res1.status_code == status.HTTP_200_OK + res = client.patch(f"/category/{res1.json()['id']}", data=json.dumps({"order": -1})) + assert res.status_code == status.HTTP_400_BAD_REQUEST + client.delete(f"/category/{res1.json()['id']}") + + +def test_delete_order(db_category, client): + body = { + "name": "new", + "type": "test", + } + res1 = client.post('/category', data=json.dumps(body)) + assert res1.status_code == status.HTTP_200_OK + assert res1.json()['order'] == 2 + + res = client.delete(f"/category/{db_category.id}") + assert res.status_code == status.HTTP_200_OK + + res = client.get(f"/category/{res1.json()['id']}") + assert res.json()['order'] == 1 + + +def test_scopes(client, mocker: MockerFixture): + user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__') + user_mock.return_value = { + "session_scopes": [{"id": 0, "name": "string", "comment": "string"}], + "user_scopes": [{"id": 0, "name": "string", "comment": "string"}], + "indirect_groups": [{"id": 0, "name": "string", "parent_id": 0}], + "groups": [{"id": 0, "name": "string", "parent_id": 0}], + "id": 0, + "email": "string", + } + + # Create value + body = {"name": "new", "type": "grid3", "scopes": ["string"]} + res1 = client.post('/category', data=json.dumps(body)) + assert res1.status_code == status.HTTP_200_OK + id_ = res1.json()['id'] + assert res1.json()['scopes'] == ["string"] + + res2 = client.get(f'/category/{id_}') + assert res2.status_code == status.HTTP_200_OK + assert res2.json()['scopes'] == ["string"] + + # Patch value to value + body = {"scopes": ["string2"]} + res3 = client.patch(f'/category/{id_}', data=json.dumps(body)) + assert res3.status_code == status.HTTP_200_OK + assert res3.json()['scopes'] == ["string2"] + + res4 = client.get(f'/category/{id_}') + assert res4.status_code == status.HTTP_404_NOT_FOUND + + # Patch value to null + body = {"scopes": []} + res5 = client.patch(f'/category/{id_}', data=json.dumps(body)) + assert res5.status_code == status.HTTP_200_OK + assert res5.json()['scopes'] == [] + + res6 = client.get(f'/category/{id_}') + assert res6.status_code == status.HTTP_200_OK + assert res6.json()['scopes'] == [] + + # Patch null to value + body = {"scopes": ["string3"]} + res7 = client.patch(f'/category/{id_}', data=json.dumps(body)) + assert res7.status_code == status.HTTP_200_OK + assert res7.json()['scopes'] == ["string3"] + + res8 = client.get(f'/category/{id_}') + assert res8.status_code == status.HTTP_404_NOT_FOUND