phi/async_tests/test_async_ldap_new_model.py

197 lines
4.9 KiB
Python

# -*- encoding: utf-8 -*-
import asyncio
from async_generator import asynccontextmanager
from bonsai import LDAPEntry
import mock
import pytest
from phi.async_ldap.new_model import get_dn, User
from phi.async_ldap.mixins import Member
from phi.exceptions import PhiCannotExecute
BASE_DN = "dc=test,dc=domain,dc=tld"
class MockClient(object):
def __init__(self, *args, **kwargs):
self.return_value = kwargs.get("return_value")
self.connect_called = False
self.conn = mock.MagicMock()
self.search_event = asyncio.Event()
self.add_event = asyncio.Event()
self.delete_event = asyncio.Event()
async def connect_called_with_search(self):
return await self.search_event.wait()
async def connect_called_with_add(self):
return await self.add_event.wait()
async def connect_called_with_delete(self):
return await self.delete_event.wait()
@property
def base_dn(self):
return BASE_DN
@asynccontextmanager
async def connect(self, *args, **kwargs):
self.connect_called = True
async def _search(*a, **kw):
self.search_event.set()
return self.return_value
async def _add(*a, **kw):
self.add_event.set()
return self.return_value
async def _modify(*a, **kw):
return self.return_value
async def _delete(*a, **kw):
self.delete_event.set()
return self.return_value
self.conn.search = mock.MagicMock(side_effect=_search)
self.conn.add = mock.MagicMock(side_effect=_add)
self.conn.modify = mock.MagicMock(side_effect=_modify)
self.conn.delete = mock.MagicMock(side_effect=_delete)
yield self.conn
cl = mock.MagicMock()
cl.base_dn = BASE_DN
@pytest.mark.parametrize(
"input_obj, expected_result",
[
(User(cl, "test_user"), f"uid=test_user,ou=Hackers,{BASE_DN}"),
(
LDAPEntry(f"uid=test_user,ou=Hackers,{BASE_DN}"),
f"uid=test_user,ou=Hackers,{BASE_DN}",
),
(f"uid=test_user,ou=Hackers,{BASE_DN}", f"uid=test_user,ou=Hackers,{BASE_DN}"),
],
)
def test_get_dn(input_obj, expected_result):
assert get_dn(input_obj) == expected_result
def test_get_dn_raises():
with pytest.raises(ValueError) as e:
_ = get_dn(object)
assert "Unacceptable input:" in str(e.value)
def test_repr():
_cl = MockClient(return_value=None)
u = User(_cl, "test_user")
assert repr(u) == f"<User uid=test_user,ou=Hackers,{BASE_DN}>"
def test_str():
_cl = MockClient(return_value=None)
u = User(_cl, "test_user")
assert str(u) == f"<User uid=test_user,ou=Hackers,{BASE_DN}>"
@pytest.mark.parametrize(
"input_obj",
[
User(cl, "test_user"),
LDAPEntry(f"uid=test_user,ou=Hackers,{BASE_DN}"),
f"uid=test_user,ou=Hackers,{BASE_DN}",
],
)
def test_eq(input_obj):
u = User(cl, "test_user")
assert u == input_obj
@pytest.mark.asyncio
async def test_User_add():
_cl = MockClient(return_value=None)
u = User(_cl, "test_user")
assert u.dn == f"uid=test_user,ou=Hackers,{BASE_DN}"
_ = await u.save()
assert _cl.connect_called
assert await _cl.connect_called_with_add()
@pytest.mark.asyncio
async def test_User_modify():
"""
This test does not use the MockClient check facilities because
of implementation details of the Entry class.
"""
_cl = MockClient(
return_value=[
LDAPEntry(f"uid=test_user,ou=Hackers,{BASE_DN}"),
]
)
u = User(_cl, "test_user")
# This is the asyncio equivalent of a semaphore
modified = asyncio.Event()
async def _mock_modify():
modified.set()
u._entry = mock.MagicMock()
u._entry.modify = mock.MagicMock(side_effect=_mock_modify)
u["cn"] = "random_cn"
_ = await u.modify()
assert _cl.connect_called
assert await _cl.connect_called_with_search()
# The `wait()` call here is needed to wait for `_mock_modify`
# to end its async execution
assert await modified.wait()
@pytest.mark.asyncio
async def test_User_delete():
_cl = MockClient(return_value=None)
u = User(_cl, "test_user")
_ = await u.delete()
assert _cl.connect_called
assert await _cl.connect_called_with_delete()
@pytest.mark.asyncio
async def test_User_get_invalid_attr_raises():
_cl = MockClient(return_value=None)
u = User(_cl, "test_user")
with pytest.raises(PhiCannotExecute) as ex:
_ = u["iDoNotExist"]
assert "iDoNotExist" in str(ex.value)
assert "is not an allowed ldap attribute" in str(ex.value)
@pytest.mark.asyncio
async def test_User_set_invalid_attr_raises():
_cl = MockClient(return_value=None)
u = User(_cl, "test_user")
with pytest.raises(PhiCannotExecute) as ex:
u["iDoNotExist"] = "hello"
assert "iDoNotExist" in str(ex.value)
assert "is not an allowed ldap attribute" in str(ex.value)