conftest.py

"""Create test user.

Copyright (c) 2024 MultiFactor
License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
"""

import pytest_asyncio
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from entities import CatalogueSetting
from ldap_protocol.dns import (
    DNS_MANAGER_IP_ADDRESS_NAME,
    DNS_MANAGER_STATE_NAME,
    DNS_MANAGER_ZONE_NAME,
    DNSManagerState,
)
from ldap_protocol.ldap_codes import LDAPCodes
from ldap_protocol.ldap_requests.modify import Operation


@pytest_asyncio.fixture(scope="function")
async def adding_test_user(
    app: FastAPI,
    http_client: AsyncClient,
    _force_override_tls: None,
) -> None:
    """Test add user like keycloak."""
    test_user_dn = "cn=test,dc=md,dc=test"
    user_password = '"\x00P\x00@\x00s\x00s\x00w\x000\x00r\x00d\x00"\x00'  # noqa
    response = await http_client.post(
        "/entry/add",
        json={
            "entry": test_user_dn,
            "password": None,
            "attributes": [
                {
                    "type": "name",
                    "vals": ["test"],
                },
                {
                    "type": "cn",
                    "vals": ["test"],
                },
                {
                    "type": "testing_attr",
                    "vals": ["test"],
                },
                {
                    "type": "sAMAccountName",
                    "vals": ["test"],
                },
                {
                    "type": "objectClass",
                    "vals": [
                        "top",
                        "user",
                        "person",
                        "organizationalPerson",
                        "posixAccount",
                        "shadowAccount",
                        "inetOrgPerson",
                    ],
                },
            ],
        },
    )
    data = response.json()
    assert data["resultCode"] == LDAPCodes.SUCCESS

    response = await http_client.patch(
        "/entry/update",
        json={
            "object": test_user_dn,
            "changes": [
                {
                    "operation": Operation.ADD,
                    "modification": {
                        "type": "mail",
                        "vals": ["new_user@md.test"],
                    },
                },
                {
                    "operation": Operation.ADD,
                    "modification": {
                        "type": "userPrincipalName",
                        "vals": ["new_user@md.test"],
                    },
                },
                {
                    "operation": Operation.ADD,
                    "modification": {
                        "type": "displayName",
                        "vals": ["Test User"],
                    },
                },
                {
                    "operation": Operation.ADD,
                    "modification": {
                        "type": "unicodePwd",
                        "vals": [user_password],
                    },
                },
                {
                    "operation": Operation.ADD,
                    "modification": {
                        "type": "memberOf",
                        "vals": ["cn=domain admins,cn=Groups,dc=md,dc=test"],
                    },
                },
                {
                    "operation": Operation.ADD,
                    "modification": {
                        "type": "userAccountControl",
                        "vals": ["0"],
                    },
                },
            ],
        },
    )
    data = response.json()
    assert data["resultCode"] == LDAPCodes.SUCCESS

    async with AsyncClient(
        transport=ASGITransport(app=app, root_path="/api"),
        timeout=3,
        base_url="http://test",
    ) as client:
        auth = await client.post(
            "auth/",
            data={
                "username": "new_user@md.test",
                "password": "P@ssw0rd",
            },
        )

        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,
) -> None:
    """Add DNS manager settings to DB."""
    dns_ip_address = "127.0.0.1"
    domain = "example.com"
    dns_state = DNSManagerState.HOSTED

    session.add_all(
        [
            CatalogueSetting(
                name=DNS_MANAGER_IP_ADDRESS_NAME,
                value=dns_ip_address,
            ),
            CatalogueSetting(
                name=DNS_MANAGER_ZONE_NAME,
                value=domain,
            ),
            CatalogueSetting(
                name=DNS_MANAGER_STATE_NAME,
                value=dns_state,
            ),
        ],
    )
    await session.commit()