Add validation field to param (#42)

* Add validation field to param

* Add tests for info validation

* User data validation fix

migrations/versions/0932ab8ca14a_19_added_validtation_column_to_param.py

@@ -0,0 +1,29 @@
+"""19 Added validtation column to Param
+
+Revision ID: 0932ab8ca14a
+Revises: f8c57101c0f6
+Create Date: 2024-07-24 01:07:25.199873
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = '0932ab8ca14a'
+down_revision = 'f8c57101c0f6'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('param', sa.Column('validation', sa.String(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('param', 'validation')
+    # ### end Alembic commands ###

tests/test_routes/test_param.py

@@ -33,6 +33,62 @@ def test_create_with_scopes(client, dbsession, category):
     dbsession.flush()
 
 
+@pytest.mark.authenticated("userdata.param.create")
+def test_create_with_validation(client, dbsession, category):
+    _category = category()
+    name = f"test{random_string()}"
+    validation = "^test_[0-9]{3}$"
+    response = client.post(
+        f"/category/{_category.id}/param",
+        json={
+            "name": name,
+            "category_id": _category.id,
+            "type": "last",
+            "changeable": "true",
+            "is_required": "true",
+            "validation": validation,
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["id"]
+    assert response.json()["name"] == name
+    assert response.json()["category_id"] == _category.id
+    assert response.json()["type"] == "last"
+    assert response.json()["changeable"] == True
+    assert response.json()["is_required"] == True
+    assert response.json()["validation"] == validation
+    param = Param.get(response.json()["id"], session=dbsession)
+    assert param
+    assert param.name == name
+    assert param.id == response.json()["id"]
+    assert param.type == "last"
+    assert param.changeable == True
+    assert param.category_id == _category.id
+    assert param.category == _category
+    assert param.validation == validation
+    dbsession.delete(param)
+    dbsession.flush()
+
+
+@pytest.mark.authenticated("userdata.param.create")
+def test_create_with_uncompilable_validation(client, category):
+    _category = category()
+    name = f"test{random_string()}"
+    validation = '[]['
+    response = client.post(
+        f"/category/{_category.id}/param",
+        json={
+            "name": name,
+            "category_id": _category.id,
+            "type": "last",
+            "changeable": "true",
+            "is_required": "true",
+            "validation": validation,
+        },
+    )
+    assert response.status_code == 422
+
+
 @pytest.mark.authenticated()
 def test_get(client, dbsession, param):
     _param = param()

tests/test_routes/test_source.py

@@ -18,6 +18,8 @@ def test_create(client, dbsession):
     assert response.json()["name"] == q.name == name
     assert response.json()["trust_level"] == q.trust_level == 8
     assert response.json()["id"] == q.id
+    dbsession.delete(q)
+    dbsession.flush()
 
 
 @pytest.mark.authenticated()

tests/test_routes/test_user_post_then_get.py

@@ -290,3 +290,118 @@ def test_update_from_user_source_not_changeable(dbsession, client, param, source
     assert info1.value == "user_info"
     dbsession.delete(info1)
     dbsession.commit()
+
+
+@pytest.mark.authenticated(user_id=0)
+def test_create_new_with_validation(dbsession, client, param, source):
+    param = param()
+    source = source()
+    source.name = "user"
+    param.type = "all"
+    param.validation = "^validation_[1-3]{3}$"
+    dbsession.commit()
+    response_upd = client.post(
+        f"/user/0",
+        json={
+            "source": source.name,
+            "items": [{"category": param.category.name, "param": param.name, "value": "validation_123"}],
+        },
+    )
+    dbsession.expire_all()
+    assert response_upd.status_code == 200
+    response_get = client.get("/user/0")
+    assert response_upd.json() == {'status': 'Success', 'message': 'User patch succeeded', 'ru': 'Изменение успешно'}
+    assert {"category": param.category.name, "param": param.name, "value": "validation_123"} in list(
+        response_get.json()["items"]
+    )
+    assert len(response_get.json()["items"]) == 1
+    info_new = (
+        dbsession.query(Info)
+        .filter(Info.param_id == param.id, Info.owner_id == 0, Info.source_id == source.id, Info.is_deleted == False)
+        .one()
+    )
+    dbsession.delete(info_new)
+    dbsession.commit()
+
+
+@pytest.mark.authenticated(user_id=0)
+def test_update_with_validation(dbsession, client, param, source):
+    param = param()
+    source = source()
+    source.name = "user"
+    param.type = "all"
+    param.validation = "^validation_[1-3]{3}$"
+    info1 = Info(value="validation_111", source_id=source.id, param_id=param.id, owner_id=0)
+    dbsession.add(info1)
+    dbsession.commit()
+    response_upd = client.post(
+        f"/user/0",
+        json={
+            "source": source.name,
+            "items": [{"category": param.category.name, "param": param.name, "value": "validation_222"}],
+        },
+    )
+    dbsession.expire_all()
+    assert response_upd.status_code == 200
+    response_get = client.get("/user/0")
+    assert response_upd.json() == {'status': 'Success', 'message': 'User patch succeeded', 'ru': 'Изменение успешно'}
+    assert {"category": param.category.name, "param": param.name, "value": "validation_222"} in list(
+        response_get.json()["items"]
+    )
+    assert len(response_get.json()["items"]) == 1
+    info_new = (
+        dbsession.query(Info)
+        .filter(Info.param_id == param.id, Info.owner_id == 0, Info.source_id == source.id, Info.is_deleted == False)
+        .one()
+    )
+    dbsession.delete(info_new)
+    dbsession.commit()
+
+
+@pytest.mark.authenticated(user_id=0)
+def test_create_new_with_failing_validation(dbsession, client, param, source):
+    param = param()
+    source = source()
+    source.name = "user"
+    param.type = "all"
+    param.validation = "^validation_[1-3]{3}$"
+    dbsession.commit()
+    response_upd = client.post(
+        f"/user/0",
+        json={
+            "source": source.name,
+            "items": [{"category": param.category.name, "param": param.name, "value": "validation_000"}],
+        },
+    )
+    dbsession.expire_all()
+    assert response_upd.status_code == 422
+    response_get = client.get("/user/0")
+    assert response_get.status_code == 404
+
+
+@pytest.mark.authenticated(user_id=0)
+def test_update_with_failing_validation(dbsession, client, param, source):
+    param = param()
+    source = source()
+    source.name = "user"
+    param.type = "all"
+    param.validation = "^validation_[1-3]{3}$"
+    info = Info(value="validation_111", source_id=source.id, param_id=param.id, owner_id=0)
+    dbsession.add(info)
+    dbsession.commit()
+    response_upd = client.post(
+        f"/user/0",
+        json={
+            "source": source.name,
+            "items": [{"category": param.category.name, "param": param.name, "value": "validation_000"}],
+        },
+    )
+    dbsession.expire_all()
+    assert response_upd.status_code == 422
+    response_get = client.get("/user/0")
+    assert {"category": param.category.name, "param": param.name, "value": "validation_111"} in list(
+        response_get.json()["items"]
+    )
+    assert len(response_get.json()["items"]) == 1
+    dbsession.delete(info)
+    dbsession.commit()

userdata_api/exceptions.py

@@ -21,5 +21,21 @@ def __init__(self, obj: type, obj_id_or_name: int | str):
         )
 
 
+class InvalidValidation(UserDataApiError):
+    def __init__(self, obj: type, field_name: str):
+        super().__init__(
+            f"Invalid validation for field {field_name} in object {obj.__name__}",
+            f"Некорректная валидация для поля {field_name} в объекте {obj.__name__} ",
+        )
+
+
+class InvalidRegex(UserDataApiError):
+    def __init__(self, obj: type, field_name: str):
+        super().__init__(
+            f"Invalid regex for field {field_name} in object {obj.__name__}",
+            f"Некорректное регулярное выражение для поля {field_name} в объекте {obj.__name__} ",
+        )
+
+
 class Forbidden(UserDataApiError):
     pass

userdata_api/models/db.py

@@ -62,6 +62,7 @@ class Param(BaseDbModel):
     is_required: Mapped[bool] = mapped_column(Boolean, default=False)
     changeable: Mapped[bool] = mapped_column(Boolean, default=True)
     type: Mapped[ViewType] = mapped_column(DbEnum(ViewType, native_enum=False))
+    validation: Mapped[str] = mapped_column(String, 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)

userdata_api/routes/exc_handlers.py

@@ -1,7 +1,7 @@
 import starlette
 from starlette.responses import JSONResponse
 
-from ..exceptions import AlreadyExists, Forbidden, ObjectNotFound
+from ..exceptions import AlreadyExists, Forbidden, InvalidRegex, InvalidValidation, ObjectNotFound
 from ..schemas.response_model import StatusResponseModel
 from .base import app
 
@@ -25,3 +25,18 @@ async def already_exists_handler(req: starlette.requests.Request, exc: AlreadyEx
     return JSONResponse(
         content=StatusResponseModel(status="Already exists", message=exc.en, ru=exc.ru).model_dump(), status_code=409
     )
+
+
+@app.exception_handler(InvalidValidation)
+async def invalid_validation_handler(req: starlette.requests.Request, exc: InvalidValidation):
+    return JSONResponse(
+        content=StatusResponseModel(status="Invalid validation", message=exc.en, ru=exc.ru).model_dump(),
+        status_code=422,
+    )
+
+
+@app.exception_handler(InvalidRegex)
+async def invalid_regex_handler(req: starlette.requests.Request, exc: InvalidRegex):
+    return JSONResponse(
+        content=StatusResponseModel(status="Invalid regex", message=exc.en, ru=exc.ru).model_dump(), status_code=422
+    )

userdata_api/routes/param.py

@@ -1,11 +1,13 @@
+from re import compile
+from re import error as ReError
 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 userdata_api.exceptions import AlreadyExists, ObjectNotFound
+from userdata_api.exceptions import AlreadyExists, InvalidRegex, ObjectNotFound
 from userdata_api.models.db import Category, Param
 from userdata_api.schemas.param import ParamGet, ParamPatch, ParamPost
 from userdata_api.schemas.response_model import StatusResponseModel
@@ -35,6 +37,11 @@ async def create_param(
     Category.get(category_id, session=db.session)
     if Param.query(session=db.session).filter(Param.category_id == category_id, Param.name == param_inp.name).all():
         raise AlreadyExists(Param, param_inp.name)
+    if param_inp.validation:
+        try:
+            compile(param_inp.validation)
+        except ReError:
+            raise InvalidRegex(Param, "validation")
     return ParamGet.model_validate(Param.create(session=db.session, **param_inp.dict(), category_id=category_id))
 
 
@@ -83,10 +90,15 @@ async def patch_param(
     :param category_id: Адйи категории в которой находится параметр
     :param param_inp: Модель для создания параметра
     :param _: Аутентификация
-    :return: ParamGet- Обновленный параметр
+    :return: ParamGet - Обновленный параметр
     """
     if category_id:
         Category.get(category_id, session=db.session)
+    if param_inp.validation:
+        try:
+            compile(param_inp.validation)
+        except ReError:
+            raise InvalidRegex(Param, "validation")
     if category_id:
         return ParamGet.from_orm(
             Param.update(id, session=db.session, **param_inp.dict(exclude_unset=True), category_id=category_id)

userdata_api/schemas/param.py

@@ -10,13 +10,15 @@ class ParamPost(Base):
     is_required: bool
     changeable: bool
     type: ViewType
+    validation: constr(min_length=1) | None = None
 
 
 class ParamPatch(Base):
     name: constr(min_length=1) | None = None
     is_required: bool | None = None
     changeable: bool | None = None
     type: ViewType | None = None
+    validation: constr(min_length=1) | None = None
 
 
 class ParamGet(ParamPost):

userdata_api/utils/user.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
+from re import search
+
 from fastapi_sqlalchemy import db
 from sqlalchemy import not_
 
-from userdata_api.exceptions import Forbidden, ObjectNotFound
+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
 
@@ -81,6 +83,8 @@ async def patch_user_info(new: UserInfoUpdate, user_id: int, user: dict[str, int
             source = Source.query(session=db.session).filter(Source.name == new.source).one_or_none()
             if not source:
                 raise ObjectNotFound(Source, new.source)
+            if param.validation is not None and search(param.validation, item.value) is None:
+                raise InvalidValidation(Info, "value")
             Info.create(
                 session=db.session,
                 owner_id=user_id,
@@ -89,21 +93,20 @@ async def patch_user_info(new: UserInfoUpdate, user_id: int, user: dict[str, int
                 value=item.value,
             )
             continue
-        if item.value is not None:
-            if not param.changeable and "userdata.info.update" not in scope_names:
-                db.session.rollback()
-                raise Forbidden(
-                    f"Param {param.name=} change requires 'userdata.info.update' scope",
-                    f"Изменение {param.name=} параметра требует 'userdata.info.update' права",
-                )
-            info.value = item.value
-            db.session.flush()
-            continue
-
         if item.value is None:
             info.is_deleted = True
             db.session.flush()
             continue
+        if not param.changeable and "userdata.info.update" not in scope_names:
+            db.session.rollback()
+            raise Forbidden(
+                f"Param {param.name=} change requires 'userdata.info.update' scope",
+                f"Изменение {param.name=} параметра требует 'userdata.info.update' права",
+            )
+        if param.validation is not None and search(param.validation, item.value) is None:
+            raise InvalidValidation(Info, "value")
+        info.value = item.value
+        db.session.flush()
 
 
 async def get_users_info(