lumi2/tests/test_ldap.py

"""Unit tests for the lumi2.ldap module."""

import pytest

import lumi2.ldap as ll
import lumi2.usermodel as lu
import lumi2.exceptions as le


def test_assert_is_valid_base_dn():
    for invalid_type in [None, 0, True, ("x",), ["x"]]:
        with pytest.raises(TypeError):
            ll._assert_is_valid_base_dn(invalid_type)

    for invalid_string in [
        "", " ", "ou=foo", "abc", "\u1337",
        "a,b", "dc=,dc=", "dc=a,dc=b", "ou=example,dc=com", "dc=example,dc="
        "dc=,dc=com", "dc=EXAMPLE,dc=com", "dc=example,dc=COM",
        "dc=example$,dc=com", "dc=foo,dc=com$", "dc=exa%,dc=com%",
        "dc=ex-ample,dc=com", "dc=f_oo,dc=com",
    ]:
        with pytest.raises(ll.InvalidStringFormatException):
            ll._assert_is_valid_base_dn(invalid_string)

    for valid_string in ["dc=example,dc=com", "dc=a,dc=de"]:
        assert ll._assert_is_valid_base_dn(valid_string) == None


def test_assert_is_valid_ou_dn():
    for invalid_type in [None, 0, True, ("x",), ["x"]]:
        with pytest.raises(TypeError):
            ll._assert_is_valid_ou_dn(invalid_type)

    for invalid_string in [
        "", " ", "ou=foo", "abc", "\u1337",
        "ou=,dc=example,dc=com", "ou=foo,dc=example,dc=", "ou=foo,dc=example,",
        "ou=foo,dc=example", "ou=foo,ou=bar,dc=example",
        "ou=foo,ou=bar,dc=example,dc=com", "cn=foo,dc=example,dc=com",
        "ou=abc%,dc=example,dc=com",
    ]:
        with pytest.raises(ll.InvalidStringFormatException):
            print(invalid_string)
            ll._assert_is_valid_ou_dn(invalid_string)

    for valid_string in [
        "ou=users,dc=example,dc=com", "ou=users,dc=a,dc=de",
        "ou=Users,dc=example,dc=com", "ou=abcdABCD,dc=example,dc=com",
    ]:
        assert ll._assert_is_valid_ou_dn(valid_string) == None


def test_assert_is_valid_user_dn():
    for invalid_type in [None, 0, True, ("x",), ["x"]]:
        with pytest.raises(TypeError):
            ll._assert_is_valid_user_dn(invalid_type)

    for invalid_string in [
        "", " ", "uid=foo", "abc", "\u1337",
        "uid=,ou=foo,dc=bar,dc=baz", "uid=foo,ou=,dc=bar,dc=baz",
        "uid=foo,ou=bar,dc=,dc=baz", "uid=foo,ou=bar,dc=baz,dc=",    
        "uid=foo,dc=example,dc=com", "uid=1foo,ou=bar,dc=example,dc=com",
        "uid=.bob,ou=foo,dc=bar,dc=baz", "uid=_foo,ou=foo,dc=bar,dc=baz",
        "uid=foo,uid=bar,ou=bar,dc=com", "cn=foo,ou=bar,dc=example,dc=com",
        "uid=abc%,ou=foo,dc=example,dc=com",
    ]:
        with pytest.raises(ll.InvalidStringFormatException):
            print(invalid_string)
            ll._assert_is_valid_user_dn(invalid_string)

    for valid_string in [
        "uid=foo,ou=users,dc=example,dc=com", "uid=Foo-bar.baz_,ou=users,dc=a,dc=de",
        "uid=bOb_.-,ou=Users,dc=example,dc=com", "uid=foo,ou=foo,dc=bar,dc=baz",
    ]:
        assert ll._assert_is_valid_user_dn(valid_string) == None


def test_assert_is_valid_group_dn():
    for invalid_type in [None, 0, True, ("x",), ["x"]]:
        with pytest.raises(TypeError):
            ll._assert_is_valid_group_dn(invalid_type)

    for invalid_string in [
        "", " ", "cn=foo", "abc", "\u1337",
        "cn=,ou=foo,dc=bar,dc=baz", "cn=foo,ou=,dc=bar,dc=baz",
        "cn=foo,ou=bar,dc=,dc=baz", "cn=foo,ou=bar,dc=baz,dc=",    
        "cn=foo,dc=example,dc=com", "cn=1foo,ou=bar,dc=example,dc=com",
        "cn=.bob,ou=foo,dc=bar,dc=baz", "cn=_foo,ou=foo,dc=bar,dc=baz",
        "cn=foo,cn=bar,ou=bar,dc=com",
        "cn=abc%,ou=foo,dc=bar,dc=baz", "cn=abc1,ou=foo,dc=bar,dc=baz",
        "cn=ab-c,ou=foo,dc=bar,dc=baz",
    ]:
        with pytest.raises(ll.InvalidStringFormatException):
            print(invalid_string)
            ll._assert_is_valid_group_dn(invalid_string)

    for valid_string in [
        "cn=foo,ou=groups,dc=example,dc=com", "cn=foo,ou=bar,dc=a,dc=de",
        "cn=Abc,ou=foo,dc=bar,dc=baz",
    ]:
        assert ll._assert_is_valid_group_dn(valid_string) == None


def test_assert_is_valid_user_object_class():
    for invalid_type in [None, 0, True, ("x",), ["x"]]:
        with pytest.raises(TypeError):
            ll._assert_is_valid_user_object_class(invalid_type)

    for invalid_string in [
        "", " ", "\u1337",
        "organizationalUnit", "INETORGPERSON", "inetorgperson",
    ]:
        with pytest.raises(ll.InvalidStringFormatException):
            ll._assert_is_valid_user_object_class(invalid_string)

    assert ll._assert_is_valid_user_object_class("inetOrgPerson") == None


def test_assert_is_valid_group_object_class():
    for invalid_type in [None, 0, True, ("x",), ["x"]]:
        with pytest.raises(TypeError):
            ll._assert_is_valid_group_object_class(invalid_type)

    for invalid_string in [
        "", " ", "\u1337",
        "organizationalUnit", "inetOrgPerson",
        "groupofuniquenames", "GROUPOFUNIQUENAMES",
    ]:
        with pytest.raises(ll.InvalidStringFormatException):
            ll._assert_is_valid_group_object_class(invalid_string)

    assert ll._assert_is_valid_group_object_class("groupOfUniqueNames") == None


def test_assert_is_valid_connection(connection):
    for invalid_type in [1, "", None, (connection,)]:
        with pytest.raises(TypeError):
            ll._assert_is_valid_connection(invalid_type)

    assert ll._assert_is_valid_connection(connection) == None

    with pytest.raises(ll.InvalidConnectionException):
        connection.unbind()
        ll._assert_is_valid_connection(connection)


def test_ou_exists(app, connection):
    with app.app_context():
        for invalid_type in [1, True, None]:
            with pytest.raises(TypeError):
                ll.ou_exists(connection, invalid_type)

        with pytest.raises(ll.InvalidStringFormatException):
            ll.ou_exists(connection, "cn=foo,dc=example,dc=com")

        assert ll.ou_exists(connection, "ou=users,dc=example,dc=com")
        assert ll.ou_exists(connection, "ou=groups,dc=example,dc=com")
        assert not ll.ou_exists(connection, "ou=nonexsistent,dc=example,dc=com")
        

def test_ou_exists_unpopulated(app, connection_empty_dit):
    with app.app_context():
        assert not ll.ou_exists(connection_empty_dit, "ou=users,dc=example,dc=com")
        assert not ll.ou_exists(connection_empty_dit, "ou=groups,dc=example,dc=com")


def test_create_ou(app, connection):
    with app.app_context():
        with pytest.raises(ll.EntryExistsException):
            ll.create_ou(connection, "ou=users,dc=example,dc=com")

        assert not ll.ou_exists(connection, "ou=foo,dc=example,dc=com")
        ll.create_ou(connection, "ou=foo,dc=example,dc=com")
        assert ll.ou_exists(connection, "ou=foo,dc=example,dc=com")


def test_user_exists(app, connection):
    with app.app_context():
        assert ll.user_exists(connection, "uid=alice,ou=users,dc=example,dc=com")
        assert not ll.user_exists(connection, "uid=nonexistent,ou=users,dc=example,dc=com")
    

def test_get_user(app, connection):
    with app.app_context():
        for invalid_type in [1, True, None]:
            with pytest.raises(TypeError):
                ll.get_user(connection, invalid_type)

        with pytest.raises(ll.EntryNotFoundException):
            ll.get_user(connection, "nonexistent")

        assert ll.get_user(connection, "alice").username == "alice"


def test_create_user(app, connection):
    with app.app_context():
        user0 = lu.User(
            username="user0",
            password_hash=lu.User.generate_password_hash("password"),
            email="valid@example.com",
            first_name="Test",
            last_name="User",
        )
        alice_duplicate = lu.User(
            username="alice",
            password_hash=lu.User.generate_password_hash("password"),
            email="valid@example.com",
            first_name="Test",
            last_name="User",
        )

        with pytest.raises(ll.EntryExistsException):
            ll.create_user(connection, alice_duplicate)

        assert ll.create_user(connection, user0) == None


def test_update_user(app, connection):
    with app.app_context():
        user0 = lu.User(
            username="user0",
            password_hash=lu.User.generate_password_hash("password"),
            email="valid@example.com",
            first_name="Test",
            last_name="User",
        )

        with pytest.raises(TypeError):
            ll.update_user(connection, None)
        with pytest.raises(ll.EntryNotFoundException):
            ll.update_user(connection, user0)

        ll.create_user(connection, user0)
        assert ll.get_user(connection, "user0").email == "valid@example.com"
        user0.email = "other@example.com"
        ll.update_user(connection, user0)
        assert ll.get_user(connection, "user0").email == "other@example.com"


def test_delete_user(app, connection):
    with app.app_context():
        user0 = lu.User(
            username="user0",
            password_hash=lu.User.generate_password_hash("password"),
            email="valid@example.com",
            first_name="Test",
            last_name="User",
        )

        with pytest.raises(ll.EntryNotFoundException):
            ll.delete_user(connection, "user0")

        ll.create_user(connection, user0)
        assert ll.user_exists(connection, user0.get_dn())
        ll.delete_user(connection, "user0")
        assert not ll.user_exists(connection, user0.get_dn())

        # TODO check that group deletion works


def test_get_users(app, connection):
    with app.app_context():
        assert len(ll.get_users(connection)) == 2
        for user in ll.get_users(connection):
            assert user.username in ["alice", "bobbuilder"]

        ll.create_user(
            connection,
            lu.User(
                username="user0",
                password_hash=lu.User.generate_password_hash("password"),
                email="valid@example.com",
                first_name="Test",
                last_name="User",
            ),
        )

        assert len(ll.get_users(connection)) == 3
        for user in ll.get_users(connection):
            assert user.username in ["alice", "bobbuilder", "user0"]


def test_group_exists(app, connection):
    with app.app_context():
        assert ll.group_exists(
            connection, "cn=employees,ou=groups,dc=example,dc=com"
        )
        assert not ll.group_exists(
            connection, "cn=nonexistent,ou=groups,dc=example,dc=com"
        )


def test_create_group(app, connection):
    with app.app_context():
        alice = ll.get_user(connection, "alice")
        testgroup = lu.Group("testgroup", {alice})

        ll.create_group(connection, testgroup)
        assert ll.group_exists(connection, "cn=testgroup,ou=groups,dc=example,dc=com")

        with pytest.raises(ll.EntryExistsException):
            ll.create_group(connection, testgroup)

        user0 = lu.User(
            username="user0",
            password_hash=lu.User.generate_password_hash("password"),
            email="valid@example.com",
            first_name="Test",
            last_name="User",
        )
        with pytest.raises(ll.EntryNotFoundException):
            ll.create_group(connection, lu.Group("invalidgroup", {user0}))



def test_delete_group(app, connection):
    with app.app_context():
        assert ll.group_exists(connection, "cn=employees,ou=groups,dc=example,dc=com")
        ll.delete_group(connection, "employees")
        assert not ll.group_exists(connection, "cn=employees,ou=groups,dc=example,dc=com")

        with pytest.raises(ll.EntryNotFoundException):
            ll.delete_group(connection, "employees")


def test_update_group(app, connection):
    with app.app_context():
        employees = ll.get_group(connection, "employees")
        user0 = lu.User(
            username="user0",
            password_hash=lu.User.generate_password_hash("password"),
            email="valid@example.com",
            first_name="Test",
            last_name="User",
        )
        employees.members.add(user0)
        with pytest.raises(ll.EntryNotFoundException):
            ll.update_group(connection, employees)
        ll.create_user(connection, user0)
        ll.update_group(connection, employees)
        assert user0 in ll.get_group(connection, "employees").members


def test_get_group(app, connection):
    with app.app_context():
        employees = ll.get_group(connection, "employees")
        assert len(employees.members) == 2

        with pytest.raises(ll.EntryNotFoundException):
            ll.get_group(connection, "nonexistent")


def test_get_groups(app, connection):
    with app.app_context():
        groups = ll.get_groups(connection)
        assert len(groups) == 2

        for group in groups:
            ll.delete_group(connection, group.groupname)
        assert len(ll.get_groups(connection)) == 0


def test_get_groups_of_user(app, connection):
    with app.app_context():
        alice = ll.get_user(connection, "alice")
        bobbuilder = ll.get_user(connection, "bobbuilder")

        assert len(ll.get_groups_of_user(connection, alice)) == 2
        assert len(ll.get_groups_of_user(connection, bobbuilder)) == 1