Feature/add extra fields to user info

pybotx/bot/handler.py

@@ -1,14 +1,6 @@
 from dataclasses import dataclass
 from functools import partial
-from typing import (
-    TYPE_CHECKING,
-    Awaitable,
-    Callable,
-    List,
-    Literal,
-    TypeVar,
-    Union,
-)
+from typing import TYPE_CHECKING, Awaitable, Callable, List, Literal, TypeVar, Union
 
 from pybotx.models.commands import BotCommand
 from pybotx.models.message.incoming_message import IncomingMessage

pybotx/client/users_api/user_from_csv.py

@@ -30,6 +30,8 @@ class BotXAPIUserFromCSVResult(VerifiedPayloadBaseModel):
     office: Optional[str] = Field(alias="Office")
     manager: Optional[str] = Field(alias="Manager")
     manager_huid: Optional[UUID] = Field(alias="Manager HUID")
+    manager_dn: Optional[str] = Field(alias="Manager DN")
+    user_dn: str = Field(alias="User DN")
     description: Optional[str] = Field(alias="Description")
     phone: Optional[str] = Field(alias="Phone")
     other_phone: Optional[str] = Field(alias="Other phone")
@@ -47,6 +49,7 @@ class BotXAPIUserFromCSVResult(VerifiedPayloadBaseModel):
         "office",
         "manager",
         "manager_huid",
+        "manager_dn",
         "description",
         "phone",
         "other_phone",
@@ -80,6 +83,8 @@ def to_domain(self) -> UserFromCSV:
             office=self.office,
             manager=self.manager,
             manager_huid=self.manager_huid,
+            manager_dn=self.manager_dn,
+            user_dn=self.user_dn,
             description=self.description,
             phone=self.phone,
             other_phone=self.other_phone,

pybotx/models/users.py

@@ -79,6 +79,8 @@ class UserFromCSV:
         office: Office info.
         manager: User's manager full name.
         manager_huid: User's manager huid.
+        manager_dn: User's manager DN.
+        user_dn: User DN.
         description: Description.
         phone: Phone number.
         other_phone: Extra phone number.
@@ -109,3 +111,5 @@ class UserFromCSV:
     ip_phone: Optional[str] = None
     other_ip_phone: Optional[str] = None
     personnel_number: Optional[str] = None
+    user_dn: Optional[str] = None
+    manager_dn: Optional[str] = None

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pybotx"
-version = "0.75.1"
+version = "0.75.2"
 description = "A python library for interacting with eXpress BotX API"
 authors = [
     "Sidnev Nikolay <[email protected]>",

tests/client/users_api/conftest.py

@@ -1,10 +1,15 @@
-from typing import Any, Dict
+import csv
+from http import HTTPStatus
+from io import StringIO
+from typing import Any, Dict, List
 from uuid import UUID
 
+import httpx
 import pytest
 
 from pybotx import UserFromSearch, UserKinds
 from tests.client.users_api.convert_to_datetime import convert_to_datetime
+from tests.client.users_api.factories import CsvUserResponseValues
 
 
 @pytest.fixture()
@@ -115,3 +120,78 @@ def user_from_search_without_data() -> UserFromSearch:
         rts_id=None,
         updated_at=None,
     )
+
+
+@pytest.fixture
+def csv_users_from_api() -> list[dict[str, str]]:
+    """Generate a list of user dictionaries for CSV testing.
+
+    This fixture creates a list of dictionaries representing user data as it would
+    appear in a CSV response from the BotX API.
+
+    :return: A list of dictionaries where each dictionary represents
+        a user with CSV column names as keys and corresponding values as strings.
+    """
+    result_list = CsvUserResponseValues.create_batch(2)
+
+    # add extra value for check APISyncSourceTypes convertion
+    result_list.append(CsvUserResponseValues(Sync_source="unsupported"))  # type: ignore
+
+    return result_list
+
+
+@pytest.fixture
+def users_csv_response(csv_users_from_api: List[Dict[str, str]]) -> httpx.Response:
+    """
+    Create a mock HTTP response with CSV user data.
+
+    This fixture takes the user data from ``csv_users_from_api`` and converts it into
+    CSV format. It then returns an ``httpx.Response`` object containing the CSV content.
+
+    :param: A list of dictionaries containing user data for CSV conversion.
+
+    :return: An HTTP response object with status code 200 (OK) and CSV-formatted
+        user data as content.
+    """
+
+    csv_columns = [
+        "HUID",
+        "AD Login",
+        "Domain",
+        "AD E-mail",
+        "Name",
+        "Sync source",
+        "Active",
+        "Kind",
+        "User DN",
+        "Company",
+        "Department",
+        "Position",
+        "Manager",
+        "Manager HUID",
+        "Manager DN",
+        "Personnel number",
+        "Description",
+        "IP phone",
+        "Other IP phone",
+        "Phone",
+        "Other phone",
+        "Avatar",
+        "Office",
+        "Avatar preview",
+    ]
+
+    output = StringIO()
+    writer = csv.DictWriter(output, fieldnames=csv_columns)
+    writer.writeheader()
+
+    for user in csv_users_from_api:
+        ordered_row_values = {
+            column_name: user[column_name] for column_name in csv_columns
+        }
+        writer.writerow(ordered_row_values)
+
+    return httpx.Response(
+        status_code=HTTPStatus.OK,
+        content=output.getvalue().encode("utf-8"),
+    )

tests/client/users_api/factories.py

@@ -0,0 +1,50 @@
+from factory import DictFactory, Faker  # type: ignore
+
+
+class CsvUserResponseValues(DictFactory):
+    """Factory for generating CSV user response data.
+
+    This factory creates dictionaries that simulate user data as it would appear
+    in a CSV response from the BotX API.
+
+    """
+
+    HUID = Faker("uuid4")  # type: ignore
+    AD_Login = Faker("user_name")  # type: ignore
+    Domain = "cts.example.com"
+    AD_E_mail = Faker("email")  # type: ignore
+    Name = Faker("name")  # type: ignore
+    Sync_source = "ad"
+    Active = "true"
+    Kind = "cts_user"
+    User_DN = Faker("name")  # type: ignore
+    Company = Faker("company")  # type: ignore
+    Department = Faker("catch_phrase")  # type: ignore
+    Position = Faker("job")  # type: ignore
+    Manager = Faker("name")  # type: ignore
+    Manager_HUID = ""
+    Manager_DN = ""
+    Personnel_number = ""
+    Description = Faker("sentence")  # type: ignore
+    IP_phone = Faker("phone_number")  # type: ignore
+    Other_IP_phone = Faker("phone_number")  # type: ignore
+    Phone = Faker("phone_number")  # type: ignore
+    Other_phone = Faker("phone_number")  # type: ignore
+    Avatar = Faker("file_name", category="image")  # type: ignore
+    Office = Faker("city")  # type: ignore
+    Avatar_preview = Faker("file_name", category="image")  # type: ignore
+
+    class Meta:
+        rename = {
+            "AD_Login": "AD Login",
+            "AD_E_mail": "AD E-mail",
+            "Sync_source": "Sync source",
+            "IP_phone": "IP phone",
+            "Other_IP_phone": "Other IP phone",
+            "Other_phone": "Other phone",
+            "Avatar_preview": "Avatar preview",
+            "Manager_HUID": "Manager HUID",
+            "Manager_DN": "Manager DN",
+            "Personnel_number": "Personnel number",
+            "User_DN": "User DN",
+        }

tests/client/users_api/test_users_as_csv.py

@@ -1,15 +1,14 @@
 from http import HTTPStatus
+from typing import Any, Dict, List
 from uuid import UUID
 
 import httpx
 import pytest
 from respx.router import MockRouter
 
-from pybotx import Bot, HandlerCollector, lifespan_wrapper
+from pybotx import Bot, HandlerCollector, SyncSourceTypes, UserFromCSV, lifespan_wrapper
 from pybotx.client.exceptions.users import NoUserKindSelectedError
 from pybotx.models.bot_account import BotAccountWithSecret
-from pybotx.models.enums import SyncSourceTypes, UserKinds
-from pybotx.models.users import UserFromCSV
 
 pytestmark = [
     pytest.mark.asyncio,
@@ -18,6 +17,66 @@
 ]
 
 
+def assert_csv_user_consist_csv_data(  # noqa:  WPS218
+    csv_user: UserFromCSV,
+    original_row: Dict[str, str],
+) -> None:
+    """Verify that a UserFromCSV object contains the correct data from a CSV row.
+
+    This function performs assertions to check that all fields in the UserFromCSV object
+    match the corresponding values in the original CSV row dictionary.
+
+    Args:
+        csv_user: A UserFromCSV object created from CSV data
+        original_row: A dictionary containing the original CSV row data
+    """
+    assert csv_user.username == original_row["Name"]
+    assert csv_user.ad_domain == original_row["Domain"]
+    assert csv_user.email == original_row["AD E-mail"]
+    assert str(csv_user.active).lower() == original_row["Active"]
+    assert str(csv_user.huid) == original_row["HUID"]
+    assert csv_user.user_kind.value.lower() == original_row["Kind"]
+    if isinstance(csv_user.sync_source, SyncSourceTypes):
+        assert str(csv_user.sync_source.value).lower() == original_row["Sync source"]
+    else:
+        assert str(csv_user.sync_source).lower() == original_row["Sync source"]
+
+    if original_row["Manager HUID"]:
+        assert str(csv_user.manager_huid) == original_row["Manager HUID"]
+    else:
+        assert csv_user.manager_huid is None
+
+    string_fields = [
+        "AD Login",
+        "User DN",
+        "Company",
+        "Department",
+        "Position",
+        "Manager",
+        "Manager HUID",
+        "Manager DN",
+        "Personnel number",
+        "Description",
+        "IP phone",
+        "Other IP phone",
+        "Phone",
+        "Other phone",
+        "Avatar",
+        "Office",
+        "Avatar preview",
+    ]
+
+    for string_field_name in string_fields:
+        object_field_name = string_field_name.lower().replace(" ", "_")
+        if original_row[string_field_name]:
+            assert original_row[string_field_name] == getattr(
+                csv_user,
+                object_field_name,
+            )
+        else:
+            assert getattr(csv_user, object_field_name) is None
+
+
 async def test__users_as_csv__no_user_kind_selected_error(
     respx_mock: MockRouter,
     host: str,
@@ -63,21 +122,19 @@ async def test__users_as_csv__succeed(
     host: str,
     bot_id: UUID,
     bot_account: BotAccountWithSecret,
+    csv_users_from_api: List[Dict[str, Any]],
+    users_csv_response: httpx.Response,
 ) -> None:
+    """Test successful retrieval of users as CSV.
+
+    This test verifies that the users_as_csv method correctly retrieves users
+    in CSV format from the BotX API and converts them to UserFromCSV objects.
+    """
     endpoint = respx_mock.get(
         url=f"https://{host}/api/v3/botx/users/users_as_csv",
         headers={"Authorization": "Bearer token"},
         params={"cts_user": True, "unregistered": True, "botx": False},
-    ).mock(
-        return_value=httpx.Response(
-            status_code=HTTPStatus.OK,
-            content=(
-                b"HUID,AD Login,Domain,AD E-mail,Name,Sync source,Active,Kind,Company,Department,Position,Manager,Manager HUID,Personnel number,Description,IP phone,Other IP phone,Phone,Other phone,Avatar,Office,Avatar preview\n"
-                b"dbc8934f-d0d7-4a9e-89df-d45c137a851c,test_user_17,cts.example.com,,test_user_17,ad,false,cts_user,Company,Department,Position,Manager John,13a6909c-bce1-4dbf-8359-efb7ef8e5b34,Some number,Description,IP phone,Other IP phone,Phone,Other_phone,Avatar,Office,Avatar_preview\n"
-                b"13a6909c-bce1-4dbf-8359-efb7ef8e5b34,test_user_18,cts.example.com,,test_user_18,unsupported,true,cts_user,,,,,,,,,,,,,,"
-            ),
-        ),
-    )
+    ).mock(return_value=users_csv_response)
 
     built_bot = Bot(collectors=[HandlerCollector()], bot_accounts=[bot_account])
     users_from_csv = []
@@ -90,53 +147,6 @@ async def test__users_as_csv__succeed(
 
     # - Assert -
     assert endpoint.called
-    assert users_from_csv == [
-        UserFromCSV(
-            huid=UUID("dbc8934f-d0d7-4a9e-89df-d45c137a851c"),
-            ad_login="test_user_17",
-            ad_domain="cts.example.com",
-            username="test_user_17",
-            sync_source=SyncSourceTypes.AD,
-            active=False,
-            user_kind=UserKinds.CTS_USER,
-            email=None,
-            company="Company",
-            department="Department",
-            position="Position",
-            personnel_number="Some number",
-            manager="Manager John",
-            manager_huid=UUID("13a6909c-bce1-4dbf-8359-efb7ef8e5b34"),
-            description="Description",
-            ip_phone="IP phone",
-            other_ip_phone="Other IP phone",
-            phone="Phone",
-            other_phone="Other_phone",
-            avatar="Avatar",
-            office="Office",
-            avatar_preview="Avatar_preview",
-        ),
-        UserFromCSV(
-            huid=UUID("13a6909c-bce1-4dbf-8359-efb7ef8e5b34"),
-            ad_login="test_user_18",
-            ad_domain="cts.example.com",
-            username="test_user_18",
-            sync_source="UNSUPPORTED",
-            active=True,
-            user_kind=UserKinds.CTS_USER,
-            email=None,
-            company=None,
-            department=None,
-            position=None,
-            manager=None,
-            manager_huid=None,
-            personnel_number=None,
-            description=None,
-            ip_phone=None,
-            other_ip_phone=None,
-            phone=None,
-            other_phone=None,
-            avatar=None,
-            office=None,
-            avatar_preview=None,
-        ),
-    ]
+
+    for generated_user, original_user in zip(users_from_csv, csv_users_from_api):
+        assert_csv_user_consist_csv_data(generated_user, original_user)