From b611bb17b410473dca66aea12343d128cd1b0d94 Mon Sep 17 00:00:00 2001 From: Timur Enikeev Date: Sat, 16 May 2026 09:16:21 -0400 Subject: [PATCH 1/4] Fix Makefile SHELL --- Makefile | 2 ++ 1 file changed, 2 insertions(+) diff --git a/Makefile b/Makefile index 4a75ed1..053ca83 100644 --- a/Makefile +++ b/Makefile @@ -1,3 +1,5 @@ +SHELL := /bin/bash + run: source ./venv/bin/activate && uvicorn --reload --log-config logging_dev.conf userdata_api.routes.base:app From 0b54b1ef5886d806a4ab196ff21df85b03ffa52d Mon Sep 17 00:00:00 2001 From: Timur Enikeev Date: Sat, 16 May 2026 21:23:40 -0400 Subject: [PATCH 2/4] Add alias --- .../8b49f52e2c01_add_param_alias_table.py | 44 ++++++ userdata_api/models/db.py | 48 ++++++ userdata_api/routes/base.py | 2 + userdata_api/routes/param_alias.py | 141 ++++++++++++++++++ userdata_api/schemas/param.py | 2 + userdata_api/schemas/param_alias.py | 19 +++ userdata_api/utils/param_alias.py | 52 +++++++ userdata_api/utils/user.py | 16 +- worker/user.py | 18 +-- 9 files changed, 321 insertions(+), 21 deletions(-) create mode 100644 migrations/versions/8b49f52e2c01_add_param_alias_table.py create mode 100644 userdata_api/routes/param_alias.py create mode 100644 userdata_api/schemas/param_alias.py create mode 100644 userdata_api/utils/param_alias.py diff --git a/migrations/versions/8b49f52e2c01_add_param_alias_table.py b/migrations/versions/8b49f52e2c01_add_param_alias_table.py new file mode 100644 index 0000000..0353746 --- /dev/null +++ b/migrations/versions/8b49f52e2c01_add_param_alias_table.py @@ -0,0 +1,44 @@ +"""Add param_alias table + +Revision ID: 8b49f52e2c01 +Revises: fc911d58459b +Create Date: 2026-05-16 12:10:00.000000 + +""" + +import sqlalchemy as sa +from alembic import op + + +# revision identifiers, used by Alembic. +revision = '8b49f52e2c01' +down_revision = 'fc911d58459b' +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + 'param_alias', + sa.Column('name', sa.String(), nullable=False), + sa.Column('param_id', sa.Integer(), nullable=False), + sa.Column('source_id', sa.Integer(), nullable=True), + sa.Column('create_ts', sa.DateTime(), nullable=False), + sa.Column('modify_ts', sa.DateTime(), nullable=False), + sa.Column('is_deleted', sa.Boolean(), nullable=False), + sa.Column('id', sa.Integer(), nullable=False), + sa.ForeignKeyConstraint( + ['param_id'], + ['param.id'], + ), + sa.ForeignKeyConstraint( + ['source_id'], + ['source.id'], + ), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('name'), + ) + + +def downgrade(): + op.drop_table('param_alias') diff --git a/userdata_api/models/db.py b/userdata_api/models/db.py index 2ba3da5..aa5b5ad 100644 --- a/userdata_api/models/db.py +++ b/userdata_api/models/db.py @@ -84,6 +84,13 @@ class Param(BaseDbModel): primaryjoin="and_(Param.id==Info.param_id, not_(Info.is_deleted))", lazy="joined", ) + aliases: Mapped[list[ParamAlias]] = relationship( + "ParamAlias", + foreign_keys="ParamAlias.param_id", + back_populates="param", + primaryjoin="and_(Param.id==ParamAlias.param_id, not_(ParamAlias.is_deleted))", + lazy="joined", + ) @property def pytype(self) -> type[str | list[str]]: @@ -111,6 +118,47 @@ class Source(BaseDbModel): primaryjoin="and_(Source.id==Info.source_id, not_(Info.is_deleted))", lazy="joined", ) + aliases: Mapped[list[ParamAlias]] = relationship( + "ParamAlias", + foreign_keys="ParamAlias.source_id", + back_populates="source", + primaryjoin="and_(Source.id==ParamAlias.source_id, not_(ParamAlias.is_deleted))", + lazy="joined", + ) + + +class ParamAlias(BaseDbModel): + """ + Алиас параметра. + Может быть привязан к конкретному источнику или быть общим для всех источников. + """ + + name: Mapped[str] = mapped_column(String, unique=True) + param_id: Mapped[int] = mapped_column(Integer, ForeignKey(Param.id)) + source_id: Mapped[int | None] = mapped_column(Integer, ForeignKey(Source.id), nullable=True) + create_ts: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + modify_ts: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + is_deleted: Mapped[bool] = mapped_column(Boolean, default=False) + + param: Mapped[Param] = relationship( + "Param", + foreign_keys="ParamAlias.param_id", + back_populates="aliases", + primaryjoin="and_(ParamAlias.param_id==Param.id, not_(Param.is_deleted))", + lazy="joined", + ) + + source: Mapped[Source | None] = relationship( + "Source", + foreign_keys="ParamAlias.source_id", + back_populates="aliases", + primaryjoin="and_(ParamAlias.source_id==Source.id, not_(Source.is_deleted))", + lazy="joined", + ) + + @hybrid_property + def source_name(self) -> str | None: + return self.source.name if self.source else None class Info(BaseDbModel): diff --git a/userdata_api/routes/base.py b/userdata_api/routes/base.py index 84b566b..a48ffd9 100644 --- a/userdata_api/routes/base.py +++ b/userdata_api/routes/base.py @@ -8,6 +8,7 @@ from .admin import admin from .category import category from .param import param +from .param_alias import param_alias from .source import source from .user import user @@ -41,5 +42,6 @@ app.include_router(source) app.include_router(category) app.include_router(param) +app.include_router(param_alias) app.include_router(user) app.include_router(admin) diff --git a/userdata_api/routes/param_alias.py b/userdata_api/routes/param_alias.py new file mode 100644 index 0000000..4fbac21 --- /dev/null +++ b/userdata_api/routes/param_alias.py @@ -0,0 +1,141 @@ +from typing import Any + +from auth_lib.fastapi import UnionAuth +from fastapi import APIRouter, Depends, Request +from fastapi_sqlalchemy import db +from pydantic.type_adapter import TypeAdapter +from sqlalchemy import not_ + +from userdata_api.exceptions import AlreadyExists, ObjectNotFound +from userdata_api.models.db import Param, ParamAlias, Source +from userdata_api.schemas.param_alias import ParamAliasGet, ParamAliasPatch, ParamAliasPost +from userdata_api.schemas.response_model import StatusResponseModel + + +param_alias = APIRouter(prefix="/category/{category_id}/param/{param_id}/alias", tags=["Param Alias"]) + + +def _get_param(*, category_id: int, param_id: int) -> Param: + param = ( + Param.query(session=db.session) + .filter( + Param.id == param_id, + Param.category_id == category_id, + ) + .one_or_none() + ) + if not param: + raise ObjectNotFound(Param, param_id) + return param + + +def _get_param_alias(*, category_id: int, param_id: int, alias_id: int) -> ParamAlias: + alias = ( + ParamAlias.query(session=db.session) + .join(Param) + .filter( + ParamAlias.id == alias_id, + ParamAlias.param_id == param_id, + Param.category_id == category_id, + not_(Param.is_deleted), + ) + .one_or_none() + ) + if not alias: + raise ObjectNotFound(ParamAlias, alias_id) + return alias + + +def _validate_source(source_id: int | None) -> None: + if source_id is None: + return + source = Source.query(session=db.session).filter(Source.id == source_id).one_or_none() + if not source: + raise ObjectNotFound(Source, source_id) + + +def _check_alias_name_exists(name: str, *, alias_id: int | None = None) -> None: + query = db.session.query(ParamAlias).filter(ParamAlias.name == name) + if alias_id is not None: + query = query.filter(ParamAlias.id != alias_id) + if query.one_or_none(): + raise AlreadyExists(ParamAlias, name) + + +@param_alias.post("", response_model=ParamAliasGet) +async def create_param_alias( + request: Request, + category_id: int, + param_id: int, + alias_inp: ParamAliasPost, + _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.create"], allow_none=False, auto_error=True)), +) -> ParamAliasGet: + """ + Создать алиас параметра. + """ + _ = _get_param(category_id=category_id, param_id=param_id) + _validate_source(alias_inp.source_id) + _check_alias_name_exists(alias_inp.name) + alias = ParamAlias.create(session=db.session, param_id=param_id, **alias_inp.model_dump()) + return ParamAliasGet.model_validate(alias) + + +@param_alias.get("/{alias_id}", response_model=ParamAliasGet) +async def get_param_alias( + category_id: int, + param_id: int, + alias_id: int, +) -> ParamAliasGet: + """ + Получить алиас параметра по айди. + """ + alias = _get_param_alias(category_id=category_id, param_id=param_id, alias_id=alias_id) + return ParamAliasGet.model_validate(alias) + + +@param_alias.get("", response_model=list[ParamAliasGet]) +async def get_param_aliases(category_id: int, param_id: int) -> list[ParamAliasGet]: + """ + Получить все алиасы параметра. + """ + _ = _get_param(category_id=category_id, param_id=param_id) + aliases = ParamAlias.query(session=db.session).filter(ParamAlias.param_id == param_id).all() + type_adapter = TypeAdapter(list[ParamAliasGet]) + return type_adapter.validate_python(aliases) + + +@param_alias.patch("/{alias_id}", response_model=ParamAliasGet) +async def patch_param_alias( + category_id: int, + param_id: int, + alias_id: int, + alias_inp: ParamAliasPatch, + _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.update"], allow_none=False, auto_error=True)), +) -> ParamAliasGet: + """ + Обновить алиас параметра. + """ + alias = _get_param_alias(category_id=category_id, param_id=param_id, alias_id=alias_id) + patch_data = alias_inp.model_dump(exclude_unset=True) + if "name" in patch_data: + _check_alias_name_exists(patch_data["name"], alias_id=alias.id) + if "source_id" in patch_data: + _validate_source(patch_data["source_id"]) + alias = ParamAlias.update(alias.id, session=db.session, **patch_data) + return ParamAliasGet.model_validate(alias) + + +@param_alias.delete("/{alias_id}", response_model=StatusResponseModel) +async def delete_param_alias( + request: Request, + category_id: int, + param_id: int, + alias_id: int, + _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.delete"], allow_none=False, auto_error=True)), +) -> StatusResponseModel: + """ + Удалить алиас параметра. + """ + _ = _get_param_alias(category_id=category_id, param_id=param_id, alias_id=alias_id) + ParamAlias.delete(alias_id, session=db.session) + return StatusResponseModel(status="Success", message="Param alias deleted", ru="Алиас параметра удален") diff --git a/userdata_api/schemas/param.py b/userdata_api/schemas/param.py index 7a4bc01..3c0e3ce 100644 --- a/userdata_api/schemas/param.py +++ b/userdata_api/schemas/param.py @@ -3,6 +3,7 @@ from userdata_api.models.db import ViewType from .base import Base +from .param_alias import ParamAliasGet class ParamPost(Base): @@ -28,3 +29,4 @@ class ParamPatch(Base): class ParamGet(ParamPost): id: int category_id: int + aliases: list[ParamAliasGet] | None = None diff --git a/userdata_api/schemas/param_alias.py b/userdata_api/schemas/param_alias.py new file mode 100644 index 0000000..b580e5e --- /dev/null +++ b/userdata_api/schemas/param_alias.py @@ -0,0 +1,19 @@ +from pydantic import constr + +from .base import Base + + +class ParamAliasPost(Base): + name: constr(min_length=1) + source_id: int | None = None + + +class ParamAliasPatch(Base): + name: constr(min_length=1) | None = None + source_id: int | None = None + + +class ParamAliasGet(ParamAliasPost): + id: int + param_id: int + source_name: str | None = None diff --git a/userdata_api/utils/param_alias.py b/userdata_api/utils/param_alias.py new file mode 100644 index 0000000..65b2f7a --- /dev/null +++ b/userdata_api/utils/param_alias.py @@ -0,0 +1,52 @@ +from sqlalchemy import not_, or_ +from sqlalchemy.orm import Session + +from userdata_api.models.db import Category, Param, ParamAlias, Source + + +def get_param_by_name_or_alias( + *, + session: Session, + category_name: str, + param_name: str, + source_name: str, +) -> Param | None: + """ + Находит параметр сначала по каноническому имени, затем по алиасу. + + Если у алиаса source_id = NULL, алиас считается общим для всех источников. + """ + param = ( + session.query(Param) + .join(Category) + .filter( + Param.name == param_name, + Category.name == category_name, + not_(Param.is_deleted), + not_(Category.is_deleted), + ) + .one_or_none() + ) + if param: + return param + + source = Source.query(session=session).filter(Source.name == source_name).one_or_none() + source_id = source.id if source else None + + query = ( + session.query(Param) + .join(Category) + .join(ParamAlias, ParamAlias.param_id == Param.id) + .filter( + ParamAlias.name == param_name, + Category.name == category_name, + not_(ParamAlias.is_deleted), + not_(Param.is_deleted), + not_(Category.is_deleted), + ) + ) + if source_id is None: + query = query.filter(ParamAlias.source_id.is_(None)) + else: + query = query.filter(or_(ParamAlias.source_id == source_id, ParamAlias.source_id.is_(None))) + return query.one_or_none() diff --git a/userdata_api/utils/user.py b/userdata_api/utils/user.py index a854e00..882aef2 100644 --- a/userdata_api/utils/user.py +++ b/userdata_api/utils/user.py @@ -8,6 +8,7 @@ from userdata_api.exceptions import Forbidden, InvalidValidation, ObjectNotFound from userdata_api.models.db import Category, Info, Param, Source, ViewType from userdata_api.schemas.user import UserInfoGet, UserInfoUpdate, UsersInfoGet +from userdata_api.utils.param_alias import get_param_by_name_or_alias async def patch_user_info(new: UserInfoUpdate, user_id: int, user: dict[str, int | list[dict[str, str | int]]]) -> None: @@ -50,16 +51,11 @@ async def patch_user_info(new: UserInfoUpdate, user_id: int, user: dict[str, int if new.source == "user" and user["id"] != user_id: raise Forbidden("'user' source requires information own", "Требуется владение информацией") for item in new.items: - param = ( - db.session.query(Param) - .join(Category) - .filter( - Param.name == item.param, - Category.name == item.category, - not_(Param.is_deleted), - not_(Category.is_deleted), - ) - .one_or_none() + param = get_param_by_name_or_alias( + session=db.session, + category_name=item.category, + param_name=item.param, + source_name=new.source, ) if not param: raise ObjectNotFound(Param, item.param) diff --git a/worker/user.py b/worker/user.py index bba4661..8bff05e 100644 --- a/worker/user.py +++ b/worker/user.py @@ -4,7 +4,8 @@ from event_schema.auth import UserLogin from sqlalchemy import not_ -from userdata_api.models.db import Category, Info, Param, Source +from userdata_api.models.db import Info, Source +from userdata_api.utils.param_alias import get_param_by_name_or_alias log = logging.getLogger(__name__) @@ -12,16 +13,11 @@ def patch_user_info(new: UserLogin, user_id: int, *, session: sqlalchemy.orm.Session) -> None: for item in new.items: - param = ( - session.query(Param) - .join(Category) - .filter( - Param.name == item.param, - Category.name == item.category, - not_(Param.is_deleted), - not_(Category.is_deleted), - ) - .one_or_none() + param = get_param_by_name_or_alias( + session=session, + category_name=item.category, + param_name=item.param, + source_name=new.source, ) if not param: session.rollback() From 27c95f2effff5c92d6ceb3592406fc81fd6f1bca Mon Sep 17 00:00:00 2001 From: Timur Enikeev Date: Sun, 23 Aug 2026 12:28:08 -0400 Subject: [PATCH 3/4] Remove category from Alias --- userdata_api/routes/param_alias.py | 32 ++++++++++-------------------- 1 file changed, 10 insertions(+), 22 deletions(-) diff --git a/userdata_api/routes/param_alias.py b/userdata_api/routes/param_alias.py index 4fbac21..d44bf84 100644 --- a/userdata_api/routes/param_alias.py +++ b/userdata_api/routes/param_alias.py @@ -12,31 +12,23 @@ from userdata_api.schemas.response_model import StatusResponseModel -param_alias = APIRouter(prefix="/category/{category_id}/param/{param_id}/alias", tags=["Param Alias"]) +param_alias = APIRouter(prefix="/param/{param_id}/alias", tags=["Param Alias"]) -def _get_param(*, category_id: int, param_id: int) -> Param: - param = ( - Param.query(session=db.session) - .filter( - Param.id == param_id, - Param.category_id == category_id, - ) - .one_or_none() - ) +def _get_param(*, param_id: int) -> Param: + param = Param.query(session=db.session).filter(Param.id == param_id).one_or_none() if not param: raise ObjectNotFound(Param, param_id) return param -def _get_param_alias(*, category_id: int, param_id: int, alias_id: int) -> ParamAlias: +def _get_param_alias(*, param_id: int, alias_id: int) -> ParamAlias: alias = ( ParamAlias.query(session=db.session) .join(Param) .filter( ParamAlias.id == alias_id, ParamAlias.param_id == param_id, - Param.category_id == category_id, not_(Param.is_deleted), ) .one_or_none() @@ -65,7 +57,6 @@ def _check_alias_name_exists(name: str, *, alias_id: int | None = None) -> None: @param_alias.post("", response_model=ParamAliasGet) async def create_param_alias( request: Request, - category_id: int, param_id: int, alias_inp: ParamAliasPost, _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.create"], allow_none=False, auto_error=True)), @@ -73,7 +64,7 @@ async def create_param_alias( """ Создать алиас параметра. """ - _ = _get_param(category_id=category_id, param_id=param_id) + _ = _get_param(param_id=param_id) _validate_source(alias_inp.source_id) _check_alias_name_exists(alias_inp.name) alias = ParamAlias.create(session=db.session, param_id=param_id, **alias_inp.model_dump()) @@ -82,23 +73,22 @@ async def create_param_alias( @param_alias.get("/{alias_id}", response_model=ParamAliasGet) async def get_param_alias( - category_id: int, param_id: int, alias_id: int, ) -> ParamAliasGet: """ Получить алиас параметра по айди. """ - alias = _get_param_alias(category_id=category_id, param_id=param_id, alias_id=alias_id) + alias = _get_param_alias(param_id=param_id, alias_id=alias_id) return ParamAliasGet.model_validate(alias) @param_alias.get("", response_model=list[ParamAliasGet]) -async def get_param_aliases(category_id: int, param_id: int) -> list[ParamAliasGet]: +async def get_param_aliases(param_id: int) -> list[ParamAliasGet]: """ Получить все алиасы параметра. """ - _ = _get_param(category_id=category_id, param_id=param_id) + _ = _get_param(param_id=param_id) aliases = ParamAlias.query(session=db.session).filter(ParamAlias.param_id == param_id).all() type_adapter = TypeAdapter(list[ParamAliasGet]) return type_adapter.validate_python(aliases) @@ -106,7 +96,6 @@ async def get_param_aliases(category_id: int, param_id: int) -> list[ParamAliasG @param_alias.patch("/{alias_id}", response_model=ParamAliasGet) async def patch_param_alias( - category_id: int, param_id: int, alias_id: int, alias_inp: ParamAliasPatch, @@ -115,7 +104,7 @@ async def patch_param_alias( """ Обновить алиас параметра. """ - alias = _get_param_alias(category_id=category_id, param_id=param_id, alias_id=alias_id) + alias = _get_param_alias(param_id=param_id, alias_id=alias_id) patch_data = alias_inp.model_dump(exclude_unset=True) if "name" in patch_data: _check_alias_name_exists(patch_data["name"], alias_id=alias.id) @@ -128,7 +117,6 @@ async def patch_param_alias( @param_alias.delete("/{alias_id}", response_model=StatusResponseModel) async def delete_param_alias( request: Request, - category_id: int, param_id: int, alias_id: int, _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.delete"], allow_none=False, auto_error=True)), @@ -136,6 +124,6 @@ async def delete_param_alias( """ Удалить алиас параметра. """ - _ = _get_param_alias(category_id=category_id, param_id=param_id, alias_id=alias_id) + _ = _get_param_alias(param_id=param_id, alias_id=alias_id) ParamAlias.delete(alias_id, session=db.session) return StatusResponseModel(status="Success", message="Param alias deleted", ru="Алиас параметра удален") From 193ee0824af1c9eaf0d362b764be0de250552bac Mon Sep 17 00:00:00 2001 From: Timur Enikeev Date: Sun, 23 Aug 2026 13:05:32 -0400 Subject: [PATCH 4/4] Alias tests --- tests/test_routes/test_param_alias.py | 134 ++++++++++++++++++++++++++ tests/test_routes/test_user_update.py | 89 +++++++++++++++++ tests/test_worker/test_worker.py | 87 ++++++++++++++++- 3 files changed, 309 insertions(+), 1 deletion(-) create mode 100644 tests/test_routes/test_param_alias.py diff --git a/tests/test_routes/test_param_alias.py b/tests/test_routes/test_param_alias.py new file mode 100644 index 0000000..b5fee6e --- /dev/null +++ b/tests/test_routes/test_param_alias.py @@ -0,0 +1,134 @@ +import pytest + +from userdata_api.models.db import ParamAlias +from userdata_api.utils.utils import random_string + + +@pytest.mark.authenticated("userdata.param.create") +def test_create_param_alias(client, dbsession, param, source): + _param = param() + _source = source() + alias_name = f"alias_{random_string()}" + response = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name, "source_id": _source.id}, + ) + assert response.status_code == 200 + assert response.json()["name"] == alias_name + assert response.json()["param_id"] == _param.id + assert response.json()["source_id"] == _source.id + assert response.json()["source_name"] == _source.name + alias = ParamAlias.get(response.json()["id"], session=dbsession) + assert alias + assert alias.name == alias_name + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create") +def test_create_param_alias_global(client, dbsession, param): + _param = param() + alias_name = f"alias_{random_string()}" + response = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name}, + ) + assert response.status_code == 200 + assert response.json()["name"] == alias_name + assert response.json()["param_id"] == _param.id + assert response.json()["source_id"] is None + assert response.json()["source_name"] is None + alias = ParamAlias.get(response.json()["id"], session=dbsession) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create") +def test_create_param_alias_unique_whole_table(client, dbsession, param): + _param1 = param() + _param2 = param() + alias_name = f"alias_{random_string()}" + response_1 = client.post( + f"/param/{_param1.id}/alias", + json={"name": alias_name}, + ) + assert response_1.status_code == 200 + response_2 = client.post( + f"/param/{_param2.id}/alias", + json={"name": alias_name}, + ) + assert response_2.status_code == 409 + alias = ParamAlias.get(response_1.json()["id"], session=dbsession) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create") +def test_get_param_aliases(client, dbsession, param, source): + _param = param() + _source = source() + first_name = f"alias_{random_string()}" + second_name = f"alias_{random_string()}" + create_1 = client.post( + f"/param/{_param.id}/alias", + json={"name": first_name, "source_id": _source.id}, + ) + create_2 = client.post( + f"/param/{_param.id}/alias", + json={"name": second_name}, + ) + assert create_1.status_code == 200 + assert create_2.status_code == 200 + response = client.get(f"/param/{_param.id}/alias") + assert response.status_code == 200 + assert any(item["name"] == first_name and item["source_id"] == _source.id for item in response.json()) + assert any(item["name"] == second_name and item["source_id"] is None for item in response.json()) + dbsession.delete(ParamAlias.get(create_1.json()["id"], session=dbsession)) + dbsession.delete(ParamAlias.get(create_2.json()["id"], session=dbsession)) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create", "userdata.param.update") +def test_patch_param_alias(client, dbsession, param, source): + _param = param() + _source = source() + _new_source = source() + alias_name = f"alias_{random_string()}" + alias_new_name = f"alias_{random_string()}" + create = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name, "source_id": _source.id}, + ) + assert create.status_code == 200 + alias_id = create.json()["id"] + response = client.patch( + f"/param/{_param.id}/alias/{alias_id}", + json={"name": alias_new_name, "source_id": _new_source.id}, + ) + assert response.status_code == 200 + assert response.json()["name"] == alias_new_name + assert response.json()["source_id"] == _new_source.id + dbsession.expire_all() + alias = ParamAlias.get(alias_id, session=dbsession) + assert alias.name == alias_new_name + assert alias.source_id == _new_source.id + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create", "userdata.param.delete") +def test_delete_param_alias(client, dbsession, param): + _param = param() + alias_name = f"alias_{random_string()}" + create = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name}, + ) + assert create.status_code == 200 + alias_id = create.json()["id"] + response = client.delete(f"/param/{_param.id}/alias/{alias_id}") + assert response.status_code == 200 + alias = ParamAlias.get(alias_id, session=dbsession, with_deleted=True) + assert alias.is_deleted is True + dbsession.delete(alias) + dbsession.commit() diff --git a/tests/test_routes/test_user_update.py b/tests/test_routes/test_user_update.py index 8b1e8e2..cbf29f1 100644 --- a/tests/test_routes/test_user_update.py +++ b/tests/test_routes/test_user_update.py @@ -440,3 +440,92 @@ def test_delete_forbidden_by_category_scope(dbsession, client, param, admin_sour assert response.status_code == 403 assert not info1.is_deleted dbsession.delete(info1) + + +@pytest.mark.authenticated("test.cat_update.first", "userdata.info.admin", user_id=1) +def test_update_by_source_alias(dbsession, client, param, admin_source): + _param = param() + _param.category.update_scope = "test.cat_update.first" + alias_name = f"alias_{random_string()}" + dbsession.add(ParamAlias(name=alias_name, param_id=_param.id, source_id=admin_source.id)) + dbsession.commit() + response = client.post( + f"/user/0", + json={ + "source": "admin", + "items": [{"category": _param.category.name, "param": alias_name, "value": "first_updated"}], + }, + ) + dbsession.expire_all() + assert response.status_code == 200 + info = ( + dbsession.query(Info) + .filter( + Info.param_id == _param.id, Info.owner_id == 0, Info.source_id == admin_source.id, Info.is_deleted == False + ) + .one() + ) + assert info.value == "first_updated" + alias = dbsession.query(ParamAlias).filter(ParamAlias.name == alias_name).one() + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("test.cat_update.first", "userdata.info.admin", user_id=1) +def test_update_by_global_alias(dbsession, client, param, admin_source): + _param = param() + _param.category.update_scope = "test.cat_update.first" + alias_name = f"alias_{random_string()}" + dbsession.add(ParamAlias(name=alias_name, param_id=_param.id, source_id=None)) + dbsession.commit() + response = client.post( + f"/user/0", + json={ + "source": "admin", + "items": [{"category": _param.category.name, "param": alias_name, "value": "global_updated"}], + }, + ) + dbsession.expire_all() + assert response.status_code == 200 + info = ( + dbsession.query(Info) + .filter( + Info.param_id == _param.id, Info.owner_id == 0, Info.source_id == admin_source.id, Info.is_deleted == False + ) + .one() + ) + assert info.value == "global_updated" + alias = dbsession.query(ParamAlias).filter(ParamAlias.name == alias_name).one() + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("test.cat_update.first", "userdata.info.admin", user_id=1) +def test_update_by_foreign_source_alias_not_found(dbsession, client, param, admin_source, source): + _param = param() + _param.category.update_scope = "test.cat_update.first" + _source = source() + alias_name = f"alias_{random_string()}" + dbsession.add(ParamAlias(name=alias_name, param_id=_param.id, source_id=_source.id)) + dbsession.commit() + response = client.post( + f"/user/0", + json={ + "source": "admin", + "items": [{"category": _param.category.name, "param": alias_name, "value": "should_not_work"}], + }, + ) + assert response.status_code == 404 + info = ( + dbsession.query(Info) + .filter( + Info.param_id == _param.id, Info.owner_id == 0, Info.source_id == admin_source.id, Info.is_deleted == False + ) + .one_or_none() + ) + assert info is None + alias = dbsession.query(ParamAlias).filter(ParamAlias.name == alias_name).one() + dbsession.delete(alias) + dbsession.commit() diff --git a/tests/test_worker/test_worker.py b/tests/test_worker/test_worker.py index e1dd4a5..fb705a3 100644 --- a/tests/test_worker/test_worker.py +++ b/tests/test_worker/test_worker.py @@ -2,7 +2,7 @@ import sqlalchemy.exc from event_schema.auth import UserLogin -from userdata_api.models.db import Category, Info, Param, Source +from userdata_api.models.db import Category, Info, Param, ParamAlias, Source from userdata_api.utils.utils import random_string from worker.user import patch_user_info @@ -108,3 +108,88 @@ def test_delete(info, dbsession): dbsession.expire(info) assert info.is_deleted is True + + +def test_create_by_alias(param, source, dbsession): + alias_name = f"alias_{random_string()}" + alias = ParamAlias(name=alias_name, param_id=param.id, source_id=source.id) + dbsession.add(alias) + dbsession.commit() + patch_user_info( + UserLogin.model_validate( + { + "items": [{"category": param.category.name, "param": alias_name, "value": "test_by_alias"}], + "source": source.name, + } + ), + 1, + session=dbsession, + ) + info = ( + dbsession.query(Info) + .filter( + Info.param_id == param.id, Info.source_id == source.id, Info.owner_id == 1, Info.value == "test_by_alias" + ) + .one() + ) + assert info + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +def test_create_by_global_alias(param, source, dbsession): + alias_name = f"alias_{random_string()}" + alias = ParamAlias(name=alias_name, param_id=param.id, source_id=None) + dbsession.add(alias) + dbsession.commit() + patch_user_info( + UserLogin.model_validate( + { + "items": [{"category": param.category.name, "param": alias_name, "value": "test_by_global_alias"}], + "source": source.name, + } + ), + 1, + session=dbsession, + ) + info = ( + dbsession.query(Info) + .filter( + Info.param_id == param.id, + Info.source_id == source.id, + Info.owner_id == 1, + Info.value == "test_by_global_alias", + ) + .one() + ) + assert info + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +def test_create_by_foreign_source_alias_not_found(param, source, dbsession): + first_source = source + second_source = Source(name=f"test{random_string()}", trust_level=8) + dbsession.add(second_source) + dbsession.commit() + alias_name = f"alias_{random_string()}" + alias = ParamAlias(name=alias_name, param_id=param.id, source_id=second_source.id) + dbsession.add(alias) + dbsession.commit() + patch_user_info( + UserLogin.model_validate( + { + "items": [{"category": param.category.name, "param": alias_name, "value": "should_not_work"}], + "source": first_source.name, + } + ), + 1, + session=dbsession, + ) + with pytest.raises(sqlalchemy.exc.NoResultFound): + dbsession.query(Info).filter(Info.param_id == param.id, Info.value == "should_not_work").one() + dbsession.delete(alias) + dbsession.delete(second_source) + dbsession.commit()