refactor: ldap requests

app/entities.py

@@ -259,10 +259,6 @@ def get_dn(self, dn: str = "cn") -> str:
     def is_domain(self) -> bool:
         return not self.parent_id and self.object_class == "domain"
 
-    @property
-    def host_principal(self) -> str:
-        return f"host/{self.name}"
-
     @property
     def path_dn(self) -> str:
         return ",".join(reversed(self.path))

app/ldap_protocol/ldap_requests/add.py

@@ -290,6 +290,7 @@ async def handle(  # noqa: C901
             or "userPrincipalName" in user_attributes
         )
         is_computer = "computer" in self.attrs_dict.get("objectClass", [])
+        computer_sam_account_name = None
 
         if is_user:
             if not any(
@@ -368,33 +369,44 @@ async def handle(  # noqa: C901
             items_to_add.append(group)
             group.parent_groups.extend(parent_groups)
 
-        elif is_computer and "useraccountcontrol" not in self.l_attrs_dict:
-            if not any(
-                group.directory.name.lower() == DOMAIN_COMPUTERS_GROUP_NAME
-                for group in parent_groups
-            ):
-                parent_groups.append(
-                    await get_group(
-                        DOMAIN_COMPUTERS_GROUP_NAME,
-                        ctx.session,
-                    ),
-                )
-            await ctx.session.refresh(
-                instance=new_dir,
-                attribute_names=["groups"],
-                with_for_update=None,
-            )
-            new_dir.groups.extend(parent_groups)
+        elif is_computer:
+            computer_sam_account_name = new_dir.name
+
             attributes.append(
                 Attribute(
-                    name="userAccountControl",
-                    value=str(
-                        UserAccountControlFlag.WORKSTATION_TRUST_ACCOUNT,
-                    ),
+                    name="sAMAccountName",
+                    value=computer_sam_account_name,
                     directory_id=new_dir.id,
                 ),
             )
 
+            if "useraccountcontrol" not in self.l_attrs_dict:
+                if not any(
+                    group.directory.name.lower() == DOMAIN_COMPUTERS_GROUP_NAME
+                    for group in parent_groups
+                ):
+                    parent_groups.append(
+                        await get_group(
+                            DOMAIN_COMPUTERS_GROUP_NAME,
+                            ctx.session,
+                        ),
+                    )
+                await ctx.session.refresh(
+                    instance=new_dir,
+                    attribute_names=["groups"],
+                    with_for_update=None,
+                )
+                new_dir.groups.extend(parent_groups)
+                attributes.append(
+                    Attribute(
+                        name="userAccountControl",
+                        value=str(
+                            UserAccountControlFlag.WORKSTATION_TRUST_ACCOUNT,
+                        ),
+                        directory_id=new_dir.id,
+                    ),
+                )
+
         if (is_user or is_group) and "gidnumber" not in self.l_attrs_dict:
             reverse_d_name = new_dir.name[::-1]
             value = (
@@ -417,15 +429,6 @@ async def handle(  # noqa: C901
                 ),
             )
 
-        if is_computer:
-            attributes.append(
-                Attribute(
-                    name="sAMAccountName",
-                    value=f"{new_dir.name}",
-                    directory_id=new_dir.id,
-                ),
-            )
-
         if "samaccounttype" not in self.l_attrs_dict:
             if is_user:
                 attributes.append(
@@ -503,11 +506,11 @@ async def handle(  # noqa: C901
 
                 elif is_computer:
                     await ctx.kadmin.add_principal(
-                        f"{new_dir.host_principal}.{base_dn.name}",
+                        f"host/{computer_sam_account_name}.{base_dn.name}",
                         None,
                     )
                     await ctx.kadmin.add_principal(
-                        new_dir.host_principal,
+                        f"host/{computer_sam_account_name}",
                         None,
                     )
             except (KRBAPIAddPrincipalError, KRBAPIConnectionError):

app/ldap_protocol/ldap_requests/delete.py

@@ -157,10 +157,13 @@ async def handle(  # noqa: C901
                 await ctx.kadmin.del_principal(directory.user.sam_account_name)
 
             if await is_computer(directory.id, ctx.session):
-                await ctx.kadmin.del_principal(directory.host_principal)
-                await ctx.kadmin.del_principal(
-                    f"{directory.host_principal}.{base_dn.name}",
-                )
+                computer_sam_account_names = directory.attributes_dict.get("sAMAccountName")  # noqa: E501  # fmt: skip
+                if computer_sam_account_names:
+                    computer_sam_account_name = computer_sam_account_names[0]
+                    await ctx.kadmin.del_principal(f"host/{computer_sam_account_name}")  # noqa: E501  # fmt: skip
+                    await ctx.kadmin.del_principal(f"host/{computer_sam_account_name}.{base_dn.name}")  # noqa: E501  # fmt: skip
+                else:
+                    raise KRBAPIDeletePrincipalError
         except KRBAPIPrincipalNotFoundError:
             pass
         except (KRBAPIDeletePrincipalError, KRBAPIConnectionError):

app/ldap_protocol/ldap_requests/modify.py

@@ -8,7 +8,7 @@
 from typing import AsyncGenerator, ClassVar
 
 from loguru import logger
-from pydantic import Field
+from pydantic import PrivateAttr
 from sqlalchemy import Select, and_, delete, func, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -111,7 +111,7 @@ class ModifyRequest(BaseRequest):
     # NOTE: If the old value was changed (for example, in _delete)
     # in one method, then you need to have access to the old value
     # from other methods (for example, from _add)
-    old_vals: dict[str, str | None] = Field(default_factory=dict)
+    _old_vals: dict[str, str | None] = PrivateAttr(default_factory=dict)
 
     @classmethod
     def from_data(cls, data: list[ASN1Row]) -> "ModifyRequest":
@@ -637,7 +637,7 @@ def _need_to_cache_samaccountname_old_value(
             directory.entity_type
             and directory.entity_type.name == EntityTypeNames.COMPUTER
             and change.l_type == "samaccountname"
-            and not self.old_vals.get(change.modification.type),
+            and not self._old_vals.get(change.modification.type),
         )
 
     async def _delete(
@@ -693,7 +693,7 @@ async def _delete(
         if self._need_to_cache_samaccountname_old_value(change, directory):
             vals = directory.attributes_dict.get(change.modification.type)
             if vals:
-                self.old_vals[change.modification.type] = vals[0]
+                self._old_vals[change.modification.type] = vals[0]
 
         if attrs:
             del_query = (
@@ -1035,7 +1035,7 @@ async def _modify_computer_samaccountname(
         base_dir: Directory,
         new_sam_account_name: bytes | str,
     ) -> None:
-        old_sam_account_name = self.old_vals.get(change.modification.type)
+        old_sam_account_name = self._old_vals.get(change.modification.type)
         new_sam_account_name = str(new_sam_account_name)
 
         if not old_sam_account_name:
@@ -1066,8 +1066,6 @@ async def _get_base_dir(
                 base_dir = base_directory
                 break
         else:
-            raise ModifyForbiddenError(
-                "Base directory for computer not found.",
-            )
+            raise ModifyForbiddenError("Base directory not found.")
 
         return base_dir

app/ldap_protocol/utils/pagination.py

@@ -95,6 +95,12 @@ class PaginationResult[S, P]:
     metadata: PaginationMetadata
     items: Sequence[P]
 
+    @classmethod
+    def _validate_query(cls, query: Select[tuple[S]]) -> bool:
+        return not (
+            query._order_by_clause is None or len(query._order_by_clause) == 0  # noqa: SLF001
+        )
+
     @classmethod
     async def get(
         cls,
@@ -104,7 +110,7 @@ async def get(
         session: AsyncSession,
     ) -> Self:
         """Get paginator."""
-        if query._order_by_clause is None or len(query._order_by_clause) == 0:  # noqa: SLF001
+        if not cls._validate_query(query):
             raise ValueError("Select query must have an order_by clause.")
 
         metadata = PaginationMetadata(