Add: RenameRequest for entry (LDAP object)

app/api/main/router.py

@@ -29,6 +29,7 @@
 
 from .schema import (
     PrimaryGroupRequest,
+    RenameRequest,
     SearchRequest,
     SearchResponse,
     SearchResultDone,
@@ -123,6 +124,15 @@ async def modify_dn_many(
     return results
 
 
+@entry_router.put("/rename", error_map=error_map)
+async def rename(
+    request: RenameRequest,
+    req: Request,
+) -> LDAPResult:
+    """LDAP rename entry request."""
+    return await request.handle_api(req.state.dishka_container)
+
+
 @entry_router.delete("/delete", error_map=error_map)
 async def delete(
     request: DeleteRequest,

app/api/main/schema.py

@@ -4,11 +4,13 @@
 License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
 """
 
+from functools import cached_property
 from ipaddress import IPv4Address, IPv6Address
 from typing import final
 
 from dishka import AsyncContainer
 from pydantic import BaseModel, Field, PrivateAttr, SecretStr
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql.elements import ColumnElement, UnaryExpression
 
 from entities import Directory
@@ -18,8 +20,17 @@
     FilterInterpreterProtocol,
     StringFilterInterpreter,
 )
-from ldap_protocol.ldap_requests import SearchRequest as LDAPSearchRequest
-from ldap_protocol.ldap_responses import SearchResultDone, SearchResultEntry
+from ldap_protocol.ldap_requests import (
+    ModifyDNRequest as LDAPModifyDNRequest,
+    ModifyRequest as LDAPModifyRequest,
+    SearchRequest as LDAPSearchRequest,
+)
+from ldap_protocol.ldap_responses import (
+    LDAPResult,
+    SearchResultDone,
+    SearchResultEntry,
+)
+from ldap_protocol.objects import Changes
 from ldap_protocol.utils.const import GRANT_DN_STRING
 
 
@@ -153,3 +164,72 @@ class PrimaryGroupRequest(BaseModel):
 
     directory_dn: GRANT_DN_STRING
     group_dn: GRANT_DN_STRING
+
+
+class RenameRequest(BaseModel):
+    """Rename request schema.
+
+    Combines ModifyDN and Modify operations.
+    """
+
+    object: str
+    newrdn: str
+    changes: list[Changes]
+
+    @cached_property
+    def _new_object(self) -> str:
+        return f"{self.newrdn},{','.join(self.object.split(',')[1:])}"
+
+    @cached_property
+    def _oldrdn(self) -> str:
+        return self.object.split(",")[0]
+
+    async def _modify_dn_request(
+        self,
+        container: AsyncContainer,
+        entry: str,
+        newrdn: str,
+    ) -> LDAPResult:
+        modify_dn_request = LDAPModifyDNRequest(
+            entry=entry,
+            newrdn=newrdn,
+            deleteoldrdn=True,
+            new_superior=None,
+        )
+        return await modify_dn_request.handle_api(container)
+
+    async def _clear_session_cache(self, container: AsyncContainer) -> None:
+        session = await container.get(AsyncSession)
+        session.expire_all()
+
+    async def _modify_request(self, container: AsyncContainer) -> LDAPResult:
+        modify_request = LDAPModifyRequest(
+            object=self._new_object,
+            changes=self.changes,
+        )
+        return await modify_request.handle_api(container)
+
+    async def handle_api(self, container: AsyncContainer) -> LDAPResult:
+        """Handle RenameRequest by executing ModifyDN then Modify.
+
+        If ModifyRequest fails, rollback the ModifyDnRequest and return error.
+        """
+        modify_dn_response = await self._modify_dn_request(
+            container,
+            self.object,
+            self.newrdn,
+        )
+        if not modify_dn_response or modify_dn_response.result_code != 0:
+            return modify_dn_response
+
+        await self._clear_session_cache(container)
+
+        modify_response = await self._modify_request(container)
+        if not modify_response or modify_response.result_code != 0:
+            await self._modify_dn_request(
+                container,
+                self._new_object,
+                self._oldrdn,
+            )
+
+        return modify_response

tests/test_api/test_main/conftest.py

@@ -138,6 +138,30 @@ async def adding_test_user(
         assert auth.cookies.get("id")
 
 
+@pytest_asyncio.fixture(scope="function")
+async def adding_test_computer(
+    http_client: AsyncClient,
+) -> None:
+    """Test api correct (name) add."""
+    response = await http_client.post(
+        "/entry/add",
+        json={
+            "entry": "cn=mycomputer,dc=md,dc=test",
+            "password": None,
+            "attributes": [
+                {"type": "name", "vals": ["mycomputer name"]},
+                {"type": "cn", "vals": ["mycomputer"]},
+                {"type": "objectClass", "vals": ["computer", "top"]},
+            ],
+        },
+    )
+
+    data = response.json()
+
+    assert isinstance(data, dict)
+    assert data.get("resultCode") == LDAPCodes.SUCCESS
+
+
 @pytest_asyncio.fixture(scope="function")
 async def add_dns_settings(
     session: AsyncSession,

tests/test_api/test_main/test_router/test_rename.py

@@ -0,0 +1,138 @@
+"""Test API Rename.
+
+Copyright (c) 2026 MultiFactor
+License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
+"""
+
+import pytest
+from httpx import AsyncClient
+
+from ldap_protocol.ldap_codes import LDAPCodes
+from ldap_protocol.ldap_requests.modify import Operation
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("adding_test_user")
+@pytest.mark.usefixtures("setup_session")
+@pytest.mark.usefixtures("session")
+async def test_api_correct_rename(http_client: AsyncClient) -> None:
+    response = await http_client.put(
+        "/entry/rename",
+        json={
+            "object": "cn=test,dc=md,dc=test",
+            "newrdn": "cn=admin2",
+            "changes": [
+                {
+                    "operation": Operation.REPLACE,
+                    "modification": {
+                        "type": "sAMAccountName",
+                        "vals": ["admin2"],
+                    },
+                },
+                {
+                    "operation": Operation.REPLACE,
+                    "modification": {
+                        "type": "displayName",
+                        "vals": ["Administrator"],
+                    },
+                },
+            ],
+        },
+    )
+
+    data = response.json()
+    assert isinstance(data, dict)
+    assert data.get("resultCode") == LDAPCodes.SUCCESS
+
+    response = await http_client.post(
+        "entry/search",
+        json={
+            "base_object": "cn=admin2,dc=md,dc=test",
+            "scope": 0,
+            "deref_aliases": 0,
+            "size_limit": 1000,
+            "time_limit": 10,
+            "types_only": True,
+            "filter": "(objectClass=*)",
+            "attributes": ["*"],
+            "page_number": 1,
+        },
+    )
+
+    data = response.json()
+    assert data["resultCode"] == LDAPCodes.SUCCESS
+    assert data["search_result"][0]["object_name"] == "cn=admin2,dc=md,dc=test"
+
+    for attr in data["search_result"][0]["partial_attributes"]:
+        if attr["type"] == "sAMAccountName":
+            assert attr["vals"][0] == "admin2"
+            break
+    else:
+        raise Exception("User without sAMAccountName")
+
+    for attr in data["search_result"][0]["partial_attributes"]:
+        if attr["type"] == "displayName":
+            assert attr["vals"][0] == "Administrator"
+            break
+    else:
+        raise Exception("User without displayName")
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("adding_test_computer")
+@pytest.mark.usefixtures("setup_session")
+@pytest.mark.usefixtures("session")
+async def test_api_correct_rename_computer(http_client: AsyncClient) -> None:
+    response = await http_client.put(
+        "/entry/rename",
+        json={
+            "object": "cn=mycomputer,dc=md,dc=test",
+            "newrdn": "cn=maincomputer",
+            "changes": [
+                {
+                    "operation": Operation.REPLACE,
+                    "modification": {
+                        "type": "sAMAccountName",
+                        "vals": ["main computer"],
+                    },
+                },
+                {
+                    "operation": Operation.REPLACE,
+                    "modification": {
+                        "type": "displayName",
+                        "vals": ["main computer"],
+                    },
+                },
+            ],
+        },
+    )
+
+    data = response.json()
+    assert isinstance(data, dict)
+    assert data.get("resultCode") == LDAPCodes.UNDEFINED_ATTRIBUTE_TYPE
+
+    response = await http_client.post(
+        "entry/search",
+        json={
+            "base_object": "cn=mycomputer,dc=md,dc=test",
+            "scope": 0,
+            "deref_aliases": 0,
+            "size_limit": 1000,
+            "time_limit": 10,
+            "types_only": True,
+            "filter": "(objectClass=*)",
+            "attributes": ["*"],
+            "page_number": 1,
+        },
+    )
+
+    data = response.json()
+    assert data["resultCode"] == LDAPCodes.SUCCESS
+    assert data["search_result"][0]["object_name"] == "cn=mycomputer,dc=md,dc=test"  # noqa: E501  # fmt: skip
+
+    for attr in data["search_result"][0]["partial_attributes"]:
+        if attr["type"] == "name":
+            assert attr["vals"][0] == "mycomputer name"
+            break
+    else:
+        raise Exception("Computer without name")