Param alias

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
 

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')

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()

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()

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()