Improve tests and added tests for sync, modify and remove.
This commit is contained in:
parent
3454c194b1
commit
fed48022af
|
@ -4,6 +4,8 @@ from argparse import Namespace
|
|||
import asyncio
|
||||
|
||||
from async_generator import asynccontextmanager
|
||||
from bonsai import NoSuchObjectError
|
||||
import mock
|
||||
import pytest
|
||||
|
||||
from phi.async_ldap.model import (
|
||||
|
@ -20,6 +22,9 @@ from phi.async_ldap.model import (
|
|||
Service,
|
||||
create_new_,
|
||||
User,
|
||||
PhiUserExists,
|
||||
PhiUserDoesNotExist,
|
||||
PhiAttributeMissing,
|
||||
Group,
|
||||
)
|
||||
|
||||
|
@ -30,7 +35,7 @@ SERVICE_LIST = [{"uid": ["phi"]}, {"uid": ["irc"]}]
|
|||
GROUP_LIST = [{"cn": ["amici_miei"]}, {"cn": ["antani"]}]
|
||||
EXISTING_USER = [
|
||||
{
|
||||
"dn": f"uid=existing_user,{BASE_DN}>",
|
||||
"dn": f"uid=existing_user,ou=Hackers,{BASE_DN}>",
|
||||
"objectClass": ["inetOrgPerson", "organizationalPerson", "person", "top"],
|
||||
"cn": ["exists"],
|
||||
"sn": ["Existing User"],
|
||||
|
@ -45,6 +50,7 @@ class MockClient:
|
|||
def __init__(self, base_dn, *args):
|
||||
self._base_dn = base_dn
|
||||
self.args = args
|
||||
self.called_with_args = {}
|
||||
|
||||
@property
|
||||
def base_dn(self):
|
||||
|
@ -55,13 +61,14 @@ class MockClient:
|
|||
conn = Namespace()
|
||||
|
||||
async def _search(*args, **kwargs):
|
||||
self.called_with_args["search"] = {"args": args, "kwargs": kwargs}
|
||||
if "Services" in self.args:
|
||||
return SERVICE_LIST
|
||||
elif "Groups" in self.args:
|
||||
return GROUP_LIST
|
||||
elif "Users" in self.args or f"None=Entry,{BASE_DN}" in args:
|
||||
if "Users" in self.args or f"None=Entry,{BASE_DN}" in args:
|
||||
return USER_LIST
|
||||
if f"uid=existing_user,ou=Hackers,{BASE_DN}" in args:
|
||||
elif f"uid=existing_user,ou=Hackers,{BASE_DN}" in args:
|
||||
return EXISTING_USER
|
||||
elif f"uid=not_existing,ou=Hackers,{BASE_DN}" in args:
|
||||
return []
|
||||
|
@ -69,10 +76,17 @@ class MockClient:
|
|||
conn.search = _search
|
||||
|
||||
async def _add(*args, **kwargs):
|
||||
self.called_with_args["add"] = {"args": args, "kwargs": kwargs}
|
||||
return
|
||||
|
||||
conn.add = _add
|
||||
|
||||
async def _modify(*args, **kwargs):
|
||||
self.called_with_args["modify"] = {"args": args, "kwargs": kwargs}
|
||||
return
|
||||
|
||||
conn.modify = _modify
|
||||
|
||||
yield conn
|
||||
|
||||
|
||||
|
@ -286,6 +300,10 @@ async def test_Hackers_get_by_attr(client_fixture):
|
|||
|
||||
assert len(res) == 1
|
||||
assert res[0] is User("existing_user", client_fixture)
|
||||
assert client_fixture.called_with_args["search"]["args"] == (
|
||||
f"uid=existing_user,ou=Hackers,{BASE_DN}",
|
||||
0,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
@ -424,6 +442,133 @@ def test_User_singleton(client_fixture):
|
|||
assert c4 is not c1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_create_existing(client_fixture):
|
||||
c = User("existing_user", client_fixture)
|
||||
|
||||
with pytest.raises(PhiUserExists):
|
||||
await c.create("existing@mail.org", sn="exists", cn="Existing User")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_create_not_existing(client_fixture):
|
||||
c = User("not_existing", client_fixture)
|
||||
|
||||
await c.create("not@existing.org")
|
||||
|
||||
assert client_fixture.called_with_args["search"]["args"] == (c.dn, 0)
|
||||
assert c._entry["mail"][0] == "not@existing.org"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_sync_existing(client_fixture, caplog):
|
||||
c = User("existing_user", client_fixture)
|
||||
|
||||
await c.sync()
|
||||
|
||||
assert "[existing_user] sync -> res:" in caplog.text
|
||||
assert client_fixture.called_with_args["search"]["args"] == (c.dn, 0)
|
||||
for k, v in EXISTING_USER[0].items():
|
||||
if k != "dn":
|
||||
assert c._entry[k] == v
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_sync_not_existing(client_fixture, caplog):
|
||||
c = User("not_existing", client_fixture)
|
||||
|
||||
with pytest.raises(PhiUserDoesNotExist):
|
||||
await c.sync()
|
||||
|
||||
assert client_fixture.called_with_args["search"]["args"] == (c.dn, 0)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_modify_existing(client_fixture):
|
||||
c = User("existing_user", client_fixture)
|
||||
c._entry = mock.MagicMock()
|
||||
|
||||
async def _modify():
|
||||
return
|
||||
|
||||
c._entry.modify = _modify
|
||||
|
||||
await c.sync()
|
||||
await c.modify("mail", "other@existing.org")
|
||||
|
||||
c._entry.__setitem__.assert_called_with("mail", "other@existing.org")
|
||||
c._entry.__delitem__.assert_called_with("mail")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_modify_existing_append(client_fixture):
|
||||
c = User("existing_user", client_fixture)
|
||||
c._entry = mock.MagicMock()
|
||||
|
||||
async def _modify():
|
||||
return
|
||||
|
||||
c._entry.modify = _modify
|
||||
|
||||
await c.sync()
|
||||
await c.modify("mail", "other@existing.org", append=True)
|
||||
|
||||
c._entry.__setitem__.assert_called_with("mail", "other@existing.org")
|
||||
c._entry.__delitem__.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_modify_not_existing(client_fixture):
|
||||
c = User("existing_user", client_fixture)
|
||||
c._entry = mock.MagicMock()
|
||||
attr = {'__delitem__.side_effect': KeyError}
|
||||
c._entry.configure_mock(**attr)
|
||||
|
||||
async def _modify():
|
||||
return
|
||||
|
||||
c._entry.modify = _modify
|
||||
|
||||
await c.sync()
|
||||
with pytest.raises(PhiAttributeMissing):
|
||||
await c.modify("snafu", "modified")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_remove_existing(client_fixture):
|
||||
c = User("existing_user", client_fixture)
|
||||
c._entry = mock.MagicMock()
|
||||
|
||||
def _delete():
|
||||
coro = mock.Mock(name="coroutine")
|
||||
fn = mock.MagicMock(side_effect=asyncio.coroutine(coro))
|
||||
return fn
|
||||
|
||||
delete = _delete()
|
||||
c._entry.delete = delete
|
||||
|
||||
await c.sync()
|
||||
await c.remove()
|
||||
|
||||
delete.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_User_remove_not_existing(client_fixture):
|
||||
c = User("not_existing", client_fixture)
|
||||
c._entry = mock.MagicMock()
|
||||
|
||||
def _delete():
|
||||
fn = mock.MagicMock(side_effect=NoSuchObjectError)
|
||||
return fn
|
||||
|
||||
delete = _delete()
|
||||
c._entry.delete = delete
|
||||
|
||||
with pytest.raises(PhiUserDoesNotExist):
|
||||
await c.remove()
|
||||
|
||||
|
||||
def test_Service(client_fixture):
|
||||
c = Service("phi", client_fixture)
|
||||
|
||||
|
|
Loading…
Reference in New Issue
Block a user