forked from Github-Mirrors/canaille
Refactored tests so ldap connection is not a mandatory argument anymore for most LDAPObject methods
This commit is contained in:
Éloi Rivard
parent
9a22352958
commit
11a750d238
22 changed files with 331 additions and 393 deletions
|
|
@ -80,8 +80,9 @@ def setup_backend(app):
|
|||
|
||||
|
||||
def teardown_backend(app):
|
||||
if "ldap" in g:
|
||||
if g.get("ldap"):
|
||||
g.ldap.unbind_s()
|
||||
g.ldap = None
|
||||
|
||||
|
||||
def init_backend(app):
|
||||
|
|
@ -89,10 +90,12 @@ def init_backend(app):
|
|||
|
||||
@app.before_request
|
||||
def before_request():
|
||||
if not app.config["TESTING"]:
|
||||
return setup_backend(app)
|
||||
|
||||
@app.after_request
|
||||
def after_request(response):
|
||||
if not app.config["TESTING"]:
|
||||
teardown_backend(app)
|
||||
return response
|
||||
|
||||
|
|
|
|||
|
|
@ -122,17 +122,15 @@ class User(LDAPObject):
|
|||
self.load_groups()
|
||||
return self._groups
|
||||
|
||||
def set_groups(self, values, conn=None):
|
||||
def set_groups(self, values):
|
||||
before = self._groups
|
||||
after = [
|
||||
v if isinstance(v, Group) else Group.get(dn=v, conn=conn) for v in values
|
||||
]
|
||||
after = [v if isinstance(v, Group) else Group.get(dn=v) for v in values]
|
||||
to_add = set(after) - set(before)
|
||||
to_del = set(before) - set(after)
|
||||
for group in to_add:
|
||||
group.add_member(self, conn=conn)
|
||||
group.add_member(self)
|
||||
for group in to_del:
|
||||
group.remove_member(self, conn=conn)
|
||||
group.remove_member(self)
|
||||
self._groups = after
|
||||
|
||||
def load_permissions(self, conn=None):
|
||||
|
|
@ -202,10 +200,10 @@ class Group(LDAPObject):
|
|||
if User.get(dn=user_dn, conn=conn)
|
||||
]
|
||||
|
||||
def add_member(self, user, conn=None):
|
||||
def add_member(self, user):
|
||||
self.member = self.member + [user.dn]
|
||||
self.save(conn=conn)
|
||||
self.save()
|
||||
|
||||
def remove_member(self, user, conn=None):
|
||||
def remove_member(self, user):
|
||||
self.member = [m for m in self.member if m != user.dn]
|
||||
self.save(conn=conn)
|
||||
self.save()
|
||||
|
|
|
|||
|
|
@ -5,12 +5,14 @@ import pytest
|
|||
import slapd
|
||||
from canaille import create_app
|
||||
from canaille.installation import setup_ldap_tree
|
||||
from canaille.ldap_backend.backend import setup_ldap_models
|
||||
from canaille.ldap_backend.ldapobject import LDAPObject
|
||||
from canaille.models import Group
|
||||
from canaille.models import User
|
||||
from cryptography.hazmat.backends import default_backend as crypto_default_backend
|
||||
from cryptography.hazmat.primitives import serialization as crypto_serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from flask import g
|
||||
from flask_webtest import TestApp
|
||||
from werkzeug.security import gen_salt
|
||||
|
||||
|
|
@ -119,12 +121,13 @@ def slapd_server():
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def slapd_connection(slapd_server):
|
||||
conn = ldap.ldapobject.SimpleLDAPObject(slapd_server.ldap_uri)
|
||||
conn.protocol_version = 3
|
||||
conn.simple_bind_s(slapd_server.root_dn, slapd_server.root_pw)
|
||||
yield conn
|
||||
conn.unbind_s()
|
||||
def slapd_connection(slapd_server, testclient):
|
||||
g.ldap = ldap.ldapobject.SimpleLDAPObject(slapd_server.ldap_uri)
|
||||
g.ldap.protocol_version = 3
|
||||
g.ldap.simple_bind_s(slapd_server.root_dn, slapd_server.root_pw)
|
||||
yield g.ldap
|
||||
if g.ldap:
|
||||
g.ldap.unbind_s()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -212,6 +215,7 @@ def configuration(slapd_server, smtpd, keypair_path):
|
|||
@pytest.fixture
|
||||
def app(configuration):
|
||||
os.environ["AUTHLIB_INSECURE_TRANSPORT"] = "true"
|
||||
setup_ldap_models(configuration)
|
||||
setup_ldap_tree(configuration)
|
||||
app = create_app(configuration)
|
||||
return app
|
||||
|
|
@ -220,7 +224,8 @@ def app(configuration):
|
|||
@pytest.fixture
|
||||
def testclient(app):
|
||||
app.config["TESTING"] = True
|
||||
return TestApp(app)
|
||||
with app.app_context():
|
||||
yield TestApp(app)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -235,9 +240,9 @@ def user(app, slapd_connection):
|
|||
mail="john@doe.com",
|
||||
userPassword="{SSHA}fw9DYeF/gHTHuVMepsQzVYAkffGcU8Fz",
|
||||
)
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
yield u
|
||||
u.delete(slapd_connection)
|
||||
u.delete()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -252,9 +257,9 @@ def admin(app, slapd_connection):
|
|||
mail="jane@doe.com",
|
||||
userPassword="{SSHA}Vmgh2jkD0idX3eZHf8RzGos31oerjGiU",
|
||||
)
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
yield u
|
||||
u.delete(slapd_connection)
|
||||
u.delete()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -269,9 +274,9 @@ def moderator(app, slapd_connection):
|
|||
mail="jack@doe.com",
|
||||
userPassword="{SSHA}+eHyxWqajMHsOWnhONC2vbtfNZzKTkag",
|
||||
)
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
yield u
|
||||
u.delete(slapd_connection)
|
||||
u.delete()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -298,33 +303,31 @@ def logged_moderator(moderator, testclient):
|
|||
@pytest.fixture
|
||||
def foo_group(app, user, slapd_connection):
|
||||
Group.ldap_object_classes(slapd_connection)
|
||||
g = Group(
|
||||
group = Group(
|
||||
objectClass=["groupOfNames"],
|
||||
member=[user.dn],
|
||||
cn="foo",
|
||||
)
|
||||
g.save(slapd_connection)
|
||||
with app.app_context():
|
||||
user.load_groups(conn=slapd_connection)
|
||||
yield g
|
||||
group.save()
|
||||
user.load_groups()
|
||||
yield group
|
||||
user._groups = []
|
||||
g.delete(conn=slapd_connection)
|
||||
group.delete()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def bar_group(app, admin, slapd_connection):
|
||||
Group.ldap_object_classes(slapd_connection)
|
||||
g = Group(
|
||||
group = Group(
|
||||
objectClass=["groupOfNames"],
|
||||
member=[admin.dn],
|
||||
cn="bar",
|
||||
)
|
||||
g.save(slapd_connection)
|
||||
with app.app_context():
|
||||
admin.load_groups(conn=slapd_connection)
|
||||
yield g
|
||||
group.save()
|
||||
admin.load_groups()
|
||||
yield group
|
||||
admin._groups = []
|
||||
g.delete(conn=slapd_connection)
|
||||
group.delete()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
|
|||
|
|
@ -4,9 +4,9 @@ from canaille.models import User
|
|||
|
||||
|
||||
def test_equality(slapd_connection, foo_group, bar_group):
|
||||
Group.ldap_object_attributes(conn=slapd_connection)
|
||||
Group.ldap_object_attributes()
|
||||
assert foo_group != bar_group
|
||||
foo_group2 = Group.get(dn=foo_group.dn, conn=slapd_connection)
|
||||
foo_group2 = Group.get(dn=foo_group.dn)
|
||||
assert foo_group == foo_group2
|
||||
|
||||
|
||||
|
|
@ -19,13 +19,13 @@ def test_dn_when_leading_space_in_id_attribute(slapd_connection):
|
|||
uid="user",
|
||||
mail="john@doe.com",
|
||||
)
|
||||
user.save(slapd_connection)
|
||||
user.save()
|
||||
|
||||
assert ldap.dn.is_dn(user.dn)
|
||||
assert ldap.dn.dn2str(ldap.dn.str2dn(user.dn)) == user.dn
|
||||
assert user.dn == "cn=Doe,ou=users,dc=slapd-test,dc=python-ldap,dc=org"
|
||||
|
||||
user.delete(slapd_connection)
|
||||
user.delete()
|
||||
|
||||
|
||||
def test_dn_when_ldap_special_char_in_id_attribute(slapd_connection):
|
||||
|
|
@ -37,10 +37,10 @@ def test_dn_when_ldap_special_char_in_id_attribute(slapd_connection):
|
|||
uid="user",
|
||||
mail="john@doe.com",
|
||||
)
|
||||
user.save(slapd_connection)
|
||||
user.save()
|
||||
|
||||
assert ldap.dn.is_dn(user.dn)
|
||||
assert ldap.dn.dn2str(ldap.dn.str2dn(user.dn)) == user.dn
|
||||
assert user.dn == "cn=\\#Doe,ou=users,dc=slapd-test,dc=python-ldap,dc=org"
|
||||
|
||||
user.delete(slapd_connection)
|
||||
user.delete()
|
||||
|
|
|
|||
|
|
@ -37,7 +37,7 @@ def client(app, slapd_connection, other_client):
|
|||
token_endpoint_auth_method="client_secret_basic",
|
||||
)
|
||||
c.audience = [c.dn, other_client.dn]
|
||||
c.save(slapd_connection)
|
||||
c.save()
|
||||
|
||||
return c
|
||||
|
||||
|
|
@ -71,7 +71,7 @@ def other_client(app, slapd_connection):
|
|||
token_endpoint_auth_method="client_secret_basic",
|
||||
)
|
||||
c.audience = [c.dn]
|
||||
c.save(slapd_connection)
|
||||
c.save()
|
||||
|
||||
return c
|
||||
|
||||
|
|
@ -93,7 +93,7 @@ def authorization(app, slapd_connection, user, client):
|
|||
challenge_method="method",
|
||||
revokation="",
|
||||
)
|
||||
a.save(slapd_connection)
|
||||
a.save()
|
||||
return a
|
||||
|
||||
|
||||
|
|
@ -111,7 +111,7 @@ def token(slapd_connection, client, user):
|
|||
issue_date=datetime.datetime.now(),
|
||||
lifetime=str(3600),
|
||||
)
|
||||
t.save(slapd_connection)
|
||||
t.save()
|
||||
return t
|
||||
|
||||
|
||||
|
|
@ -123,12 +123,12 @@ def consent(slapd_connection, client, user):
|
|||
scope=["openid", "profile"],
|
||||
issue_date=datetime.datetime.now(),
|
||||
)
|
||||
t.save(slapd_connection)
|
||||
t.save()
|
||||
return t
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def cleanup_consents(slapd_connection):
|
||||
yield
|
||||
for consent in Consent.all(conn=slapd_connection):
|
||||
consent.delete(conn=slapd_connection)
|
||||
for consent in Consent.all():
|
||||
consent.delete()
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ def test_authorization_code_flow(
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
res = testclient.post(
|
||||
|
|
@ -47,7 +47,7 @@ def test_authorization_code_flow(
|
|||
)
|
||||
|
||||
access_token = res.json["access_token"]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token.client == client.dn
|
||||
assert token.subject == logged_user.dn
|
||||
|
||||
|
|
@ -74,7 +74,7 @@ def test_authorization_code_flow_preconsented(
|
|||
testclient, slapd_connection, logged_user, client, keypair, other_client
|
||||
):
|
||||
client.preconsent = True
|
||||
client.save(conn=slapd_connection)
|
||||
client.save()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
@ -90,7 +90,7 @@ def test_authorization_code_flow_preconsented(
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
res = testclient.post(
|
||||
|
|
@ -106,7 +106,7 @@ def test_authorization_code_flow_preconsented(
|
|||
)
|
||||
|
||||
access_token = res.json["access_token"]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token.client == client.dn
|
||||
assert token.subject == logged_user.dn
|
||||
|
||||
|
|
@ -159,7 +159,7 @@ def test_logout_login(testclient, slapd_connection, logged_user, client):
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
res = testclient.post(
|
||||
|
|
@ -175,7 +175,7 @@ def test_logout_login(testclient, slapd_connection, logged_user, client):
|
|||
)
|
||||
|
||||
access_token = res.json["access_token"]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token.client == client.dn
|
||||
assert token.subject == logged_user.dn
|
||||
|
||||
|
|
@ -214,7 +214,7 @@ def test_refresh_token(testclient, slapd_connection, user, client):
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
with freezegun.freeze_time("2020-01-01 00:01:00"):
|
||||
|
|
@ -230,7 +230,7 @@ def test_refresh_token(testclient, slapd_connection, user, client):
|
|||
status=200,
|
||||
)
|
||||
access_token = res.json["access_token"]
|
||||
old_token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
old_token = Token.get(access_token=access_token)
|
||||
assert old_token is not None
|
||||
assert not old_token.revokation_date
|
||||
|
||||
|
|
@ -245,11 +245,11 @@ def test_refresh_token(testclient, slapd_connection, user, client):
|
|||
status=200,
|
||||
)
|
||||
access_token = res.json["access_token"]
|
||||
new_token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
new_token = Token.get(access_token=access_token)
|
||||
assert new_token is not None
|
||||
assert old_token.access_token != new_token.access_token
|
||||
|
||||
old_token.reload(slapd_connection)
|
||||
old_token.reload()
|
||||
assert old_token.revokation_date
|
||||
|
||||
with freezegun.freeze_time("2020-01-01 00:03:00"):
|
||||
|
|
@ -268,7 +268,7 @@ def test_refresh_token(testclient, slapd_connection, user, client):
|
|||
|
||||
def test_code_challenge(testclient, slapd_connection, logged_user, client):
|
||||
client.token_endpoint_auth_method = "none"
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
||||
code_verifier = gen_salt(48)
|
||||
code_challenge = create_s256_code_challenge(code_verifier)
|
||||
|
|
@ -291,7 +291,7 @@ def test_code_challenge(testclient, slapd_connection, logged_user, client):
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
res = testclient.post(
|
||||
|
|
@ -308,7 +308,7 @@ def test_code_challenge(testclient, slapd_connection, logged_user, client):
|
|||
)
|
||||
access_token = res.json["access_token"]
|
||||
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token.client == client.dn
|
||||
assert token.subject == logged_user.dn
|
||||
|
||||
|
|
@ -325,13 +325,13 @@ def test_code_challenge(testclient, slapd_connection, logged_user, client):
|
|||
} == res.json
|
||||
|
||||
client.token_endpoint_auth_method = "client_secret_basic"
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
||||
|
||||
def test_authorization_code_flow_when_consent_already_given(
|
||||
testclient, slapd_connection, logged_user, client
|
||||
):
|
||||
assert not Consent.all(conn=slapd_connection)
|
||||
assert not Consent.all()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
@ -349,12 +349,10 @@ def test_authorization_code_flow_when_consent_already_given(
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
consents = Consent.filter(
|
||||
client=client.dn, subject=logged_user.dn, conn=slapd_connection
|
||||
)
|
||||
consents = Consent.filter(client=client.dn, subject=logged_user.dn)
|
||||
assert "profile" in consents[0].scope
|
||||
|
||||
res = testclient.post(
|
||||
|
|
@ -388,7 +386,7 @@ def test_authorization_code_flow_when_consent_already_given(
|
|||
def test_authorization_code_flow_when_consent_already_given_but_for_a_smaller_scope(
|
||||
testclient, slapd_connection, logged_user, client
|
||||
):
|
||||
assert not Consent.all(conn=slapd_connection)
|
||||
assert not Consent.all()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
@ -406,12 +404,10 @@ def test_authorization_code_flow_when_consent_already_given_but_for_a_smaller_sc
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
consents = Consent.filter(
|
||||
client=client.dn, subject=logged_user.dn, conn=slapd_connection
|
||||
)
|
||||
consents = Consent.filter(client=client.dn, subject=logged_user.dn)
|
||||
assert "profile" in consents[0].scope
|
||||
assert "groups" not in consents[0].scope
|
||||
|
||||
|
|
@ -444,12 +440,10 @@ def test_authorization_code_flow_when_consent_already_given_but_for_a_smaller_sc
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
consents = Consent.filter(
|
||||
client=client.dn, subject=logged_user.dn, conn=slapd_connection
|
||||
)
|
||||
consents = Consent.filter(client=client.dn, subject=logged_user.dn)
|
||||
assert "profile" in consents[0].scope
|
||||
assert "groups" in consents[0].scope
|
||||
|
||||
|
|
@ -484,7 +478,7 @@ def test_prompt_none(testclient, slapd_connection, logged_user, client):
|
|||
subject=logged_user.dn,
|
||||
scope=["openid", "profile"],
|
||||
)
|
||||
consent.save(conn=slapd_connection)
|
||||
consent.save()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
@ -508,7 +502,7 @@ def test_prompt_not_logged(testclient, slapd_connection, user, client):
|
|||
subject=user.dn,
|
||||
scope=["openid", "profile"],
|
||||
)
|
||||
consent.save(conn=slapd_connection)
|
||||
consent.save()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ def test_clean_command(testclient, slapd_connection, client, user):
|
|||
challenge_method="method",
|
||||
revokation="",
|
||||
)
|
||||
code.save(slapd_connection)
|
||||
code.save()
|
||||
|
||||
Token.ldap_object_classes(slapd_connection)
|
||||
token = Token(
|
||||
|
|
@ -37,15 +37,15 @@ def test_clean_command(testclient, slapd_connection, client, user):
|
|||
issue_date=(datetime.datetime.now() - datetime.timedelta(days=1)),
|
||||
lifetime=str(3600),
|
||||
)
|
||||
token.save(slapd_connection)
|
||||
token.save()
|
||||
|
||||
assert AuthorizationCode.get(code="my-code", conn=slapd_connection)
|
||||
assert Token.get(access_token="my-token", conn=slapd_connection)
|
||||
assert AuthorizationCode.get(code="my-code")
|
||||
assert Token.get(access_token="my-token")
|
||||
assert code.is_expired()
|
||||
assert token.is_expired()
|
||||
|
||||
runner = testclient.app.test_cli_runner()
|
||||
runner.invoke(cli, ["clean"])
|
||||
|
||||
assert not AuthorizationCode.get(code="my-code", conn=slapd_connection)
|
||||
assert not Token.get(access_token="my-token", conn=slapd_connection)
|
||||
assert not AuthorizationCode.get(code="my-code")
|
||||
assert not Token.get(access_token="my-token")
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ def test_client_add(testclient, logged_admin, slapd_connection):
|
|||
res = res.follow(status=200)
|
||||
|
||||
client_id = res.forms["readonly"]["client_id"].value
|
||||
client = Client.get(client_id, conn=slapd_connection)
|
||||
client = Client.get(client_id)
|
||||
data["audience"] = [client.dn]
|
||||
for k, v in data.items():
|
||||
client_value = getattr(client, k)
|
||||
|
|
@ -87,7 +87,7 @@ def test_client_edit(testclient, client, logged_admin, slapd_connection, other_c
|
|||
"The client has not been edited. Please check your information." not in res.text
|
||||
)
|
||||
|
||||
client = Client.get(client.dn, conn=slapd_connection)
|
||||
client = Client.get(client.dn)
|
||||
for k, v in data.items():
|
||||
client_value = getattr(client, k)
|
||||
if k == "scope":
|
||||
|
|
@ -100,4 +100,4 @@ def test_client_edit(testclient, client, logged_admin, slapd_connection, other_c
|
|||
res.forms["clientadd"].submit(status=302, name="action", value="delete").follow(
|
||||
status=200
|
||||
)
|
||||
assert Client.get(client.client_id, conn=slapd_connection) is None
|
||||
assert Client.get(client.client_id) is None
|
||||
|
|
|
|||
|
|
@ -14,5 +14,5 @@ def test_consent_list(
|
|||
res = res.follow(status=200)
|
||||
assert client.name not in res.text
|
||||
|
||||
token.reload(conn=slapd_connection)
|
||||
token.reload()
|
||||
assert token.revoked
|
||||
|
|
|
|||
|
|
@ -34,11 +34,11 @@ def test_oauth_hybrid(testclient, slapd_connection, user, client):
|
|||
params = parse_qs(urlsplit(res.location).fragment)
|
||||
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
access_token = params["access_token"][0]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token is not None
|
||||
|
||||
res = testclient.get(
|
||||
|
|
@ -74,11 +74,11 @@ def test_oidc_hybrid(
|
|||
params = parse_qs(urlsplit(res.location).fragment)
|
||||
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
access_token = params["access_token"][0]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token is not None
|
||||
|
||||
id_token = params["id_token"][0]
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ def test_oauth_implicit(testclient, slapd_connection, user, client):
|
|||
client.grant_type = ["token"]
|
||||
client.token_endpoint_auth_method = "none"
|
||||
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
@ -35,7 +35,7 @@ def test_oauth_implicit(testclient, slapd_connection, user, client):
|
|||
params = parse_qs(urlsplit(res.location).fragment)
|
||||
|
||||
access_token = params["access_token"][0]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token is not None
|
||||
|
||||
res = testclient.get(
|
||||
|
|
@ -51,7 +51,7 @@ def test_oauth_implicit(testclient, slapd_connection, user, client):
|
|||
|
||||
client.grant_type = ["code"]
|
||||
client.token_endpoint_auth_method = "client_secret_basic"
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
||||
|
||||
def test_oidc_implicit(
|
||||
|
|
@ -60,7 +60,7 @@ def test_oidc_implicit(
|
|||
client.grant_type = ["token id_token"]
|
||||
client.token_endpoint_auth_method = "none"
|
||||
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
@ -86,7 +86,7 @@ def test_oidc_implicit(
|
|||
params = parse_qs(urlsplit(res.location).fragment)
|
||||
|
||||
access_token = params["access_token"][0]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token is not None
|
||||
|
||||
id_token = params["id_token"][0]
|
||||
|
|
@ -110,7 +110,7 @@ def test_oidc_implicit(
|
|||
|
||||
client.grant_type = ["code"]
|
||||
client.token_endpoint_auth_method = "client_secret_basic"
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
||||
|
||||
def test_oidc_implicit_with_group(
|
||||
|
|
@ -119,7 +119,7 @@ def test_oidc_implicit_with_group(
|
|||
client.grant_type = ["token id_token"]
|
||||
client.token_endpoint_auth_method = "none"
|
||||
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
||||
res = testclient.get(
|
||||
"/oauth/authorize",
|
||||
|
|
@ -145,7 +145,7 @@ def test_oidc_implicit_with_group(
|
|||
params = parse_qs(urlsplit(res.location).fragment)
|
||||
|
||||
access_token = params["access_token"][0]
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token is not None
|
||||
|
||||
id_token = params["id_token"][0]
|
||||
|
|
@ -170,4 +170,4 @@ def test_oidc_implicit_with_group(
|
|||
|
||||
client.grant_type = ["code"]
|
||||
client.token_endpoint_auth_method = "client_secret_basic"
|
||||
client.save(slapd_connection)
|
||||
client.save()
|
||||
|
|
|
|||
|
|
@ -40,7 +40,6 @@ def test_generate_user_standard_claims_with_default_config(
|
|||
):
|
||||
User.ldap_object_classes(slapd_connection)
|
||||
|
||||
with testclient.app.app_context():
|
||||
data = generate_user_claims(user, STANDARD_CLAIMS, DEFAULT_JWT_MAPPING_CONFIG)
|
||||
|
||||
assert data == {
|
||||
|
|
@ -58,7 +57,6 @@ def test_custom_config_format_claim_is_well_formated(
|
|||
jwt_mapping_config = DEFAULT_JWT_MAPPING_CONFIG.copy()
|
||||
jwt_mapping_config["EMAIL"] = "{{ user.uid[0] }}@mydomain.tld"
|
||||
|
||||
with testclient.app.app_context():
|
||||
data = generate_user_claims(user, STANDARD_CLAIMS, jwt_mapping_config)
|
||||
|
||||
assert data["email"] == "user@mydomain.tld"
|
||||
|
|
@ -69,9 +67,8 @@ def test_claim_is_omitted_if_empty(testclient, slapd_connection, user):
|
|||
# it's better to not insert a null or empty string value
|
||||
User.ldap_object_classes(slapd_connection)
|
||||
user.mail = ""
|
||||
user.save(slapd_connection)
|
||||
user.save()
|
||||
|
||||
with testclient.app.app_context():
|
||||
data = generate_user_claims(user, STANDARD_CLAIMS, DEFAULT_JWT_MAPPING_CONFIG)
|
||||
|
||||
assert "email" not in data
|
||||
|
|
@ -86,7 +83,6 @@ def test_custom_format_claim_is_formatted_with_empty_value_and_not_omitted(
|
|||
jwt_mapping_config = DEFAULT_JWT_MAPPING_CONFIG.copy()
|
||||
jwt_mapping_config["EMAIL"] = "{{ user.givenName[0] }}@mydomain.tld"
|
||||
|
||||
with testclient.app.app_context():
|
||||
data = generate_user_claims(user, STANDARD_CLAIMS, jwt_mapping_config)
|
||||
|
||||
assert data["email"] == "@mydomain.tld"
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ def test_password_flow_basic(testclient, slapd_connection, user, client):
|
|||
assert res.json["token_type"] == "Bearer"
|
||||
access_token = res.json["access_token"]
|
||||
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token is not None
|
||||
|
||||
res = testclient.get(
|
||||
|
|
@ -38,7 +38,7 @@ def test_password_flow_basic(testclient, slapd_connection, user, client):
|
|||
|
||||
def test_password_flow_post(testclient, slapd_connection, user, client):
|
||||
client.token_endpoint_auth_method = "client_secret_post"
|
||||
client.save(conn=slapd_connection)
|
||||
client.save()
|
||||
|
||||
res = testclient.post(
|
||||
"/oauth/token",
|
||||
|
|
@ -57,7 +57,7 @@ def test_password_flow_post(testclient, slapd_connection, user, client):
|
|||
assert res.json["token_type"] == "Bearer"
|
||||
access_token = res.json["access_token"]
|
||||
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token is not None
|
||||
|
||||
res = testclient.get(
|
||||
|
|
|
|||
|
|
@ -59,7 +59,7 @@ def test_full_flow(
|
|||
assert res.location.startswith(client.redirect_uris[0])
|
||||
params = parse_qs(urlsplit(res.location).query)
|
||||
code = params["code"][0]
|
||||
authcode = AuthorizationCode.get(code=code, conn=slapd_connection)
|
||||
authcode = AuthorizationCode.get(code=code)
|
||||
assert authcode is not None
|
||||
|
||||
res = testclient.post(
|
||||
|
|
@ -75,7 +75,7 @@ def test_full_flow(
|
|||
)
|
||||
access_token = res.json["access_token"]
|
||||
|
||||
token = Token.get(access_token=access_token, conn=slapd_connection)
|
||||
token = Token.get(access_token=access_token)
|
||||
assert token.client == client.dn
|
||||
assert token.subject == logged_user.dn
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ def test_token_revocation(testclient, user, client, token, slapd_connection):
|
|||
)
|
||||
assert {} == res.json
|
||||
|
||||
token.reload(slapd_connection)
|
||||
token.reload()
|
||||
assert token.revokation_date
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -73,7 +73,7 @@ def test_user_without_password_first_login(testclient, slapd_connection):
|
|||
uid="temp",
|
||||
mail="john@doe.com",
|
||||
)
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
|
||||
res = testclient.get("/login", status=200)
|
||||
res.form["login"] = "Temp User"
|
||||
|
|
@ -81,7 +81,7 @@ def test_user_without_password_first_login(testclient, slapd_connection):
|
|||
|
||||
assert "First login" in res
|
||||
|
||||
u.delete(slapd_connection)
|
||||
u.delete()
|
||||
|
||||
|
||||
def test_user_deleted_in_session(testclient, slapd_connection):
|
||||
|
|
@ -94,14 +94,14 @@ def test_user_deleted_in_session(testclient, slapd_connection):
|
|||
mail="jake@doe.com",
|
||||
userPassword="{SSHA}fw9DYeF/gHTHuVMepsQzVYAkffGcU8Fz",
|
||||
)
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
testclient.get("/profile/jake", status=403)
|
||||
|
||||
with testclient.session_transaction() as session:
|
||||
session["user_dn"] = [u.dn]
|
||||
|
||||
testclient.get("/profile/jake", status=200)
|
||||
u.delete(conn=slapd_connection)
|
||||
u.delete()
|
||||
|
||||
testclient.get("/profile/jake", status=403)
|
||||
with testclient.session_transaction() as session:
|
||||
|
|
@ -157,7 +157,7 @@ def test_admin_self_deletion(testclient, slapd_connection):
|
|||
mail="temp@temp.com",
|
||||
userPassword="{SSHA}Vmgh2jkD0idX3eZHf8RzGos31oerjGiU",
|
||||
)
|
||||
admin.save(slapd_connection)
|
||||
admin.save()
|
||||
with testclient.session_transaction() as sess:
|
||||
sess["user_dn"] = [admin.dn]
|
||||
|
||||
|
|
@ -168,8 +168,7 @@ def test_admin_self_deletion(testclient, slapd_connection):
|
|||
.follow(status=200)
|
||||
)
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert User.get("temp", conn=slapd_connection) is None
|
||||
assert User.get("temp") is None
|
||||
|
||||
with testclient.session_transaction() as sess:
|
||||
assert not sess.get("user_dn")
|
||||
|
|
@ -187,7 +186,7 @@ def test_user_self_deletion(testclient, slapd_connection):
|
|||
mail="temp@temp.com",
|
||||
userPassword="{SSHA}Vmgh2jkD0idX3eZHf8RzGos31oerjGiU",
|
||||
)
|
||||
user.save(slapd_connection)
|
||||
user.save()
|
||||
with testclient.session_transaction() as sess:
|
||||
sess["user_dn"] = [user.dn]
|
||||
|
||||
|
|
@ -207,8 +206,7 @@ def test_user_self_deletion(testclient, slapd_connection):
|
|||
.follow(status=200)
|
||||
)
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert User.get("temp", conn=slapd_connection) is None
|
||||
assert User.get("temp") is None
|
||||
|
||||
with testclient.session_transaction() as sess:
|
||||
assert not sess.get("user_dn")
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
def test_ldap_connection_remote_ldap_unreachable(testclient):
|
||||
testclient.app.config["TESTING"] = False
|
||||
|
||||
testclient.app.config["LDAP"]["URI"] = "ldap://invalid-ldap.com"
|
||||
|
||||
testclient.app.config["DEBUG"] = True
|
||||
|
|
@ -11,6 +13,8 @@ def test_ldap_connection_remote_ldap_unreachable(testclient):
|
|||
|
||||
|
||||
def test_ldap_connection_remote_ldap_wrong_credentials(testclient):
|
||||
testclient.app.config["TESTING"] = False
|
||||
|
||||
testclient.app.config["LDAP"]["BIND_PW"] = "invalid-password"
|
||||
|
||||
testclient.app.config["DEBUG"] = True
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
def test_password_forgotten_disabled(smtpd, testclient, slapd_connection, user):
|
||||
def test_password_forgotten_disabled(smtpd, testclient, user):
|
||||
testclient.app.config["ENABLE_PASSWORD_RECOVERY"] = False
|
||||
|
||||
testclient.get("/reset", status=404)
|
||||
|
|
@ -8,7 +8,7 @@ def test_password_forgotten_disabled(smtpd, testclient, slapd_connection, user):
|
|||
assert "Forgotten password" not in res.text
|
||||
|
||||
|
||||
def test_password_forgotten(smtpd, testclient, slapd_connection, user):
|
||||
def test_password_forgotten(smtpd, testclient, user):
|
||||
res = testclient.get("/reset", status=200)
|
||||
|
||||
res.form["login"] = "user"
|
||||
|
|
@ -19,7 +19,7 @@ def test_password_forgotten(smtpd, testclient, slapd_connection, user):
|
|||
assert len(smtpd.messages) == 1
|
||||
|
||||
|
||||
def test_password_forgotten_invalid_form(smtpd, testclient, slapd_connection, user):
|
||||
def test_password_forgotten_invalid_form(smtpd, testclient, user):
|
||||
res = testclient.get("/reset", status=200)
|
||||
|
||||
res.form["login"] = ""
|
||||
|
|
@ -29,7 +29,7 @@ def test_password_forgotten_invalid_form(smtpd, testclient, slapd_connection, us
|
|||
assert len(smtpd.messages) == 0
|
||||
|
||||
|
||||
def test_password_forgotten_invalid(smtpd, testclient, slapd_connection, user):
|
||||
def test_password_forgotten_invalid(smtpd, testclient, user):
|
||||
testclient.app.config["HIDE_INVALID_LOGINS"] = True
|
||||
res = testclient.get("/reset", status=200)
|
||||
|
||||
|
|
@ -49,9 +49,7 @@ def test_password_forgotten_invalid(smtpd, testclient, slapd_connection, user):
|
|||
assert len(smtpd.messages) == 0
|
||||
|
||||
|
||||
def test_password_forgotten_invalid_when_user_cannot_self_edit(
|
||||
smtpd, testclient, slapd_connection, user
|
||||
):
|
||||
def test_password_forgotten_invalid_when_user_cannot_self_edit(smtpd, testclient, user):
|
||||
testclient.app.config["ACL"]["DEFAULT"]["PERMISSIONS"] = []
|
||||
|
||||
testclient.app.config["HIDE_INVALID_LOGINS"] = False
|
||||
|
|
|
|||
|
|
@ -3,31 +3,29 @@ from canaille.models import User
|
|||
|
||||
|
||||
def test_no_group(app, slapd_connection):
|
||||
with app.app_context():
|
||||
assert Group.all(conn=slapd_connection) == []
|
||||
assert Group.all() == []
|
||||
|
||||
|
||||
def test_set_groups(app, slapd_connection, user, foo_group, bar_group):
|
||||
with app.app_context():
|
||||
foo_dns = {m.dn for m in foo_group.get_members(conn=slapd_connection)}
|
||||
foo_dns = {m.dn for m in foo_group.get_members()}
|
||||
assert user.dn in foo_dns
|
||||
assert user.groups[0].dn == foo_group.dn
|
||||
|
||||
user.load_groups(conn=slapd_connection)
|
||||
user.set_groups([foo_group, bar_group], conn=slapd_connection)
|
||||
user.load_groups()
|
||||
user.set_groups([foo_group, bar_group])
|
||||
|
||||
bar_group = Group.get(bar_group.dn, conn=slapd_connection)
|
||||
bar_dns = {m.dn for m in bar_group.get_members(conn=slapd_connection)}
|
||||
bar_group = Group.get(bar_group.dn)
|
||||
bar_dns = {m.dn for m in bar_group.get_members()}
|
||||
assert user.dn in bar_dns
|
||||
assert user.groups[1].dn == bar_group.dn
|
||||
|
||||
user.load_groups(conn=slapd_connection)
|
||||
user.set_groups([foo_group], conn=slapd_connection)
|
||||
user.load_groups()
|
||||
user.set_groups([foo_group])
|
||||
|
||||
foo_group = Group.get(foo_group.dn, conn=slapd_connection)
|
||||
bar_group = Group.get(bar_group.dn, conn=slapd_connection)
|
||||
foo_dns = {m.dn for m in foo_group.get_members(conn=slapd_connection)}
|
||||
bar_dns = {m.dn for m in bar_group.get_members(conn=slapd_connection)}
|
||||
foo_group = Group.get(foo_group.dn)
|
||||
bar_group = Group.get(bar_group.dn)
|
||||
foo_dns = {m.dn for m in foo_group.get_members()}
|
||||
bar_dns = {m.dn for m in bar_group.get_members()}
|
||||
assert user.dn in foo_dns
|
||||
assert user.dn not in bar_dns
|
||||
|
||||
|
|
@ -42,23 +40,22 @@ def test_set_groups_with_leading_space_in_user_id_attribute(
|
|||
uid="user2",
|
||||
mail="john@doe.com",
|
||||
)
|
||||
user.save(slapd_connection)
|
||||
user.save()
|
||||
|
||||
with app.app_context():
|
||||
user.load_groups(conn=slapd_connection)
|
||||
user.set_groups([foo_group], conn=slapd_connection)
|
||||
user.load_groups()
|
||||
user.set_groups([foo_group])
|
||||
|
||||
foo_dns = {m.dn for m in foo_group.get_members(conn=slapd_connection)}
|
||||
foo_dns = {m.dn for m in foo_group.get_members()}
|
||||
assert user.dn in foo_dns
|
||||
|
||||
user.load_groups(conn=slapd_connection)
|
||||
user.set_groups([], conn=slapd_connection)
|
||||
user.load_groups()
|
||||
user.set_groups([])
|
||||
|
||||
foo_group = Group.get(foo_group.dn, conn=slapd_connection)
|
||||
foo_dns = {m.dn for m in foo_group.get_members(conn=slapd_connection)}
|
||||
foo_group = Group.get(foo_group.dn)
|
||||
foo_dns = {m.dn for m in foo_group.get_members()}
|
||||
assert user.dn not in foo_dns
|
||||
|
||||
user.delete(slapd_connection)
|
||||
user.delete()
|
||||
|
||||
|
||||
def test_moderator_can_create_edit_and_delete_group(
|
||||
|
|
@ -66,9 +63,8 @@ def test_moderator_can_create_edit_and_delete_group(
|
|||
):
|
||||
# The group does not exist
|
||||
res = testclient.get("/groups", status=200)
|
||||
with testclient.app.app_context():
|
||||
assert Group.get("bar", conn=slapd_connection) is None
|
||||
assert Group.get("foo", conn=slapd_connection) == foo_group
|
||||
assert Group.get("bar") is None
|
||||
assert Group.get("foo") == foo_group
|
||||
assert "bar" not in res.text
|
||||
assert "foo" in res.text
|
||||
|
||||
|
|
@ -80,13 +76,10 @@ def test_moderator_can_create_edit_and_delete_group(
|
|||
# Group has been created
|
||||
res = res.form.submit(status=302).follow(status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
bar_group = Group.get("bar", conn=slapd_connection)
|
||||
bar_group = Group.get("bar")
|
||||
assert bar_group.name == "bar"
|
||||
assert bar_group.description == ["yolo"]
|
||||
assert [
|
||||
member.dn for member in bar_group.get_members(conn=slapd_connection)
|
||||
] == [
|
||||
assert [member.dn for member in bar_group.get_members()] == [
|
||||
logged_moderator.dn
|
||||
] # Group cannot be empty so creator is added in it
|
||||
assert "bar" in res.text
|
||||
|
|
@ -98,19 +91,17 @@ def test_moderator_can_create_edit_and_delete_group(
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
bar_group = Group.get("bar", conn=slapd_connection)
|
||||
bar_group = Group.get("bar")
|
||||
assert bar_group.name == "bar"
|
||||
assert bar_group.description == ["yolo2"]
|
||||
assert Group.get("bar2", conn=slapd_connection) is None
|
||||
members = bar_group.get_members(conn=slapd_connection)
|
||||
assert Group.get("bar2") is None
|
||||
members = bar_group.get_members()
|
||||
for member in members:
|
||||
assert member.name in res.text
|
||||
|
||||
# Group is deleted
|
||||
res = res.form.submit(name="action", value="delete", status=302).follow(status=200)
|
||||
with testclient.app.app_context():
|
||||
assert Group.get("bar", conn=slapd_connection) is None
|
||||
assert Group.get("bar") is None
|
||||
assert "The group bar has been sucessfully deleted" in res.text
|
||||
|
||||
|
||||
|
|
@ -135,13 +126,11 @@ def test_get_members_filters_non_existent_user(
|
|||
testclient, slapd_connection, logged_moderator, foo_group, user
|
||||
):
|
||||
# an LDAP group can be inconsistent by containing members which doesn't exist
|
||||
with testclient.app.app_context():
|
||||
non_existent_user_dn = user.dn.replace(user.name, "yolo")
|
||||
foo_group.member = foo_group.member + [non_existent_user_dn]
|
||||
foo_group.save(conn=slapd_connection)
|
||||
foo_group.save()
|
||||
|
||||
with testclient.app.app_context():
|
||||
foo_members = foo_group.get_members(conn=slapd_connection)
|
||||
foo_members = foo_group.get_members()
|
||||
|
||||
assert foo_group.member == [user.dn, non_existent_user_dn]
|
||||
assert len(foo_members) == 1
|
||||
|
|
|
|||
|
|
@ -6,8 +6,7 @@ from canaille.models import User
|
|||
|
||||
|
||||
def test_invitation(testclient, slapd_connection, logged_admin, foo_group, smtpd):
|
||||
with testclient.app.app_context():
|
||||
assert User.get("someone", conn=slapd_connection) is None
|
||||
assert User.get("someone") is None
|
||||
|
||||
res = testclient.get("/invite", status=200)
|
||||
|
||||
|
|
@ -41,10 +40,9 @@ def test_invitation(testclient, slapd_connection, logged_admin, foo_group, smtpd
|
|||
|
||||
assert "You account has been created successfuly." in res
|
||||
|
||||
with testclient.app.app_context():
|
||||
user = User.get("someone", conn=slapd_connection)
|
||||
user.load_groups(conn=slapd_connection)
|
||||
foo_group.reload(slapd_connection)
|
||||
user = User.get("someone")
|
||||
user.load_groups()
|
||||
foo_group.reload()
|
||||
assert user.check_password("whatever")
|
||||
assert user.groups == [foo_group]
|
||||
|
||||
|
|
@ -58,9 +56,8 @@ def test_invitation(testclient, slapd_connection, logged_admin, foo_group, smtpd
|
|||
def test_invitation_editable_uid(
|
||||
testclient, slapd_connection, logged_admin, foo_group, smtpd
|
||||
):
|
||||
with testclient.app.app_context():
|
||||
assert User.get("jackyjack", conn=slapd_connection) is None
|
||||
assert User.get("djorje", conn=slapd_connection) is None
|
||||
assert User.get("jackyjack") is None
|
||||
assert User.get("djorje") is None
|
||||
|
||||
res = testclient.get("/invite", status=200)
|
||||
|
||||
|
|
@ -95,10 +92,9 @@ def test_invitation_editable_uid(
|
|||
|
||||
assert "You account has been created successfuly." in res
|
||||
|
||||
with testclient.app.app_context():
|
||||
user = User.get("djorje", conn=slapd_connection)
|
||||
user.load_groups(conn=slapd_connection)
|
||||
foo_group.reload(slapd_connection)
|
||||
user = User.get("djorje")
|
||||
user.load_groups()
|
||||
foo_group.reload()
|
||||
assert user.check_password("whatever")
|
||||
assert user.groups == [foo_group]
|
||||
|
||||
|
|
@ -108,8 +104,7 @@ def test_invitation_editable_uid(
|
|||
|
||||
|
||||
def test_generate_link(testclient, slapd_connection, logged_admin, foo_group, smtpd):
|
||||
with testclient.app.app_context():
|
||||
assert User.get("sometwo", conn=slapd_connection) is None
|
||||
assert User.get("sometwo") is None
|
||||
|
||||
res = testclient.get("/invite", status=200)
|
||||
|
||||
|
|
@ -139,10 +134,9 @@ def test_generate_link(testclient, slapd_connection, logged_admin, foo_group, sm
|
|||
res = res.form.submit(status=302)
|
||||
res = res.follow(status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
user = User.get("sometwo", conn=slapd_connection)
|
||||
user.load_groups(conn=slapd_connection)
|
||||
foo_group.reload(slapd_connection)
|
||||
user = User.get("sometwo")
|
||||
user.load_groups()
|
||||
foo_group.reload()
|
||||
assert user.check_password("whatever")
|
||||
assert user.groups == [foo_group]
|
||||
|
||||
|
|
@ -165,7 +159,6 @@ def test_invitation_login_already_taken(testclient, slapd_connection, logged_adm
|
|||
|
||||
|
||||
def test_registration(testclient, slapd_connection, foo_group):
|
||||
with testclient.app.app_context():
|
||||
invitation = Invitation(
|
||||
datetime.now().isoformat(),
|
||||
"someoneelse",
|
||||
|
|
@ -180,22 +173,16 @@ def test_registration(testclient, slapd_connection, foo_group):
|
|||
|
||||
|
||||
def test_registration_invalid_hash(testclient, slapd_connection, foo_group):
|
||||
with testclient.app.app_context():
|
||||
now = datetime.now().isoformat()
|
||||
invitation1 = Invitation(
|
||||
now, "someoneelse", False, "someone@mydomain.tld", [foo_group.dn]
|
||||
)
|
||||
hash = invitation1.profile_hash()
|
||||
invitation2 = Invitation(
|
||||
invitation = Invitation(
|
||||
now, "anything", False, "someone@mydomain.tld", [foo_group.dn]
|
||||
)
|
||||
b64 = invitation2.b64()
|
||||
b64 = invitation.b64()
|
||||
|
||||
testclient.get(f"/register/{b64}/invalid", status=302)
|
||||
|
||||
|
||||
def test_registration_invalid_data(testclient, slapd_connection, foo_group):
|
||||
with testclient.app.app_context():
|
||||
invitation = Invitation(
|
||||
datetime.now().isoformat(),
|
||||
"someoneelse",
|
||||
|
|
@ -211,7 +198,6 @@ def test_registration_invalid_data(testclient, slapd_connection, foo_group):
|
|||
def test_registration_more_than_48_hours_after_invitation(
|
||||
testclient, slapd_connection, foo_group
|
||||
):
|
||||
with testclient.app.app_context():
|
||||
two_days_ago = datetime.now() - timedelta(hours=48)
|
||||
invitation = Invitation(
|
||||
two_days_ago.isoformat(),
|
||||
|
|
@ -227,7 +213,6 @@ def test_registration_more_than_48_hours_after_invitation(
|
|||
|
||||
|
||||
def test_registration_no_password(testclient, slapd_connection, foo_group):
|
||||
with testclient.app.app_context():
|
||||
invitation = Invitation(
|
||||
datetime.now().isoformat(),
|
||||
"someoneelse",
|
||||
|
|
@ -246,8 +231,7 @@ def test_registration_no_password(testclient, slapd_connection, foo_group):
|
|||
res = res.form.submit(status=200)
|
||||
assert "This field is required." in res.text, res.text
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert not User.get("someoneelse", conn=slapd_connection)
|
||||
assert not User.get("someoneelse")
|
||||
|
||||
with testclient.session_transaction() as sess:
|
||||
assert "user_dn" not in sess
|
||||
|
|
@ -256,7 +240,6 @@ def test_registration_no_password(testclient, slapd_connection, foo_group):
|
|||
def test_no_registration_if_logged_in(
|
||||
testclient, slapd_connection, logged_user, foo_group
|
||||
):
|
||||
with testclient.app.app_context():
|
||||
invitation = Invitation(
|
||||
datetime.now().isoformat(),
|
||||
"someoneelse",
|
||||
|
|
@ -294,7 +277,6 @@ def test_groups_are_saved_even_when_user_does_not_have_read_permission(
|
|||
"uid"
|
||||
] # remove groups from default read permissions
|
||||
|
||||
with testclient.app.app_context():
|
||||
invitation = Invitation(
|
||||
datetime.now().isoformat(),
|
||||
"someoneelse",
|
||||
|
|
@ -318,8 +300,7 @@ def test_groups_are_saved_even_when_user_does_not_have_read_permission(
|
|||
res = res.form.submit(status=302)
|
||||
res = res.follow(status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
user = User.get("someoneelse", conn=slapd_connection)
|
||||
user.load_groups(conn=slapd_connection)
|
||||
foo_group.reload(slapd_connection)
|
||||
user = User.get("someoneelse")
|
||||
user.load_groups()
|
||||
foo_group.reload()
|
||||
assert user.groups == [foo_group]
|
||||
|
|
|
|||
|
|
@ -1,10 +1,9 @@
|
|||
from canaille.account import profile_hash
|
||||
|
||||
|
||||
def test_password_reset(testclient, slapd_connection, user):
|
||||
user.ldap_object_attributes(conn=slapd_connection)
|
||||
user.reload(conn=slapd_connection)
|
||||
with testclient.app.app_context():
|
||||
def test_password_reset(testclient, user):
|
||||
user.ldap_object_attributes()
|
||||
user.reload()
|
||||
hash = profile_hash("user", user.mail[0], user.userPassword[0])
|
||||
|
||||
res = testclient.get("/reset/user/" + hash, status=200)
|
||||
|
|
@ -15,10 +14,9 @@ def test_password_reset(testclient, slapd_connection, user):
|
|||
|
||||
res = res.follow(status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert user.check_password("foobarbaz")
|
||||
assert "Your password has been updated successfuly" in res.text
|
||||
user.set_password("correct horse battery staple", conn=slapd_connection)
|
||||
user.set_password("correct horse battery staple")
|
||||
|
||||
res = testclient.get("/reset/user/" + hash)
|
||||
res = res.follow()
|
||||
|
|
@ -26,9 +24,9 @@ def test_password_reset(testclient, slapd_connection, user):
|
|||
assert "The password reset link that brought you here was invalid." in res.text
|
||||
|
||||
|
||||
def test_password_reset_bad_link(testclient, slapd_connection, user):
|
||||
user.ldap_object_attributes(conn=slapd_connection)
|
||||
user.reload(conn=slapd_connection)
|
||||
def test_password_reset_bad_link(testclient, user):
|
||||
user.ldap_object_attributes()
|
||||
user.reload()
|
||||
|
||||
res = testclient.get("/reset/user/foobarbaz")
|
||||
res = res.follow()
|
||||
|
|
@ -36,10 +34,9 @@ def test_password_reset_bad_link(testclient, slapd_connection, user):
|
|||
assert "The password reset link that brought you here was invalid." in res.text
|
||||
|
||||
|
||||
def test_password_reset_bad_password(testclient, slapd_connection, user):
|
||||
user.ldap_object_attributes(conn=slapd_connection)
|
||||
user.reload(conn=slapd_connection)
|
||||
with testclient.app.app_context():
|
||||
def test_password_reset_bad_password(testclient, user):
|
||||
user.ldap_object_attributes()
|
||||
user.reload()
|
||||
hash = profile_hash("user", user.mail[0], user.userPassword[0])
|
||||
|
||||
res = testclient.get("/reset/user/" + hash, status=200)
|
||||
|
|
@ -48,7 +45,6 @@ def test_password_reset_bad_password(testclient, slapd_connection, user):
|
|||
res.form["confirmation"] = "typo"
|
||||
res = res.form.submit(status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert user.check_password("correct horse battery staple")
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -55,9 +55,8 @@ def test_edition(
|
|||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
assert "Profile updated successfuly." in res, str(res)
|
||||
|
||||
with testclient.app.app_context():
|
||||
logged_user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
logged_user.load_groups(conn=slapd_connection)
|
||||
logged_user = User.get(dn=logged_user.dn)
|
||||
logged_user.load_groups()
|
||||
|
||||
assert ["user"] == logged_user.uid
|
||||
assert ["given_name"] == logged_user.givenName
|
||||
|
|
@ -67,13 +66,12 @@ def test_edition(
|
|||
assert "666" == logged_user.employeeNumber
|
||||
assert [jpeg_photo] == logged_user.jpegPhoto
|
||||
|
||||
foo_group.reload(slapd_connection)
|
||||
bar_group.reload(slapd_connection)
|
||||
foo_group.reload()
|
||||
bar_group.reload()
|
||||
assert logged_user.groups == [foo_group]
|
||||
assert foo_group.member == [logged_user.dn]
|
||||
assert bar_group.member == [admin.dn]
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert logged_user.check_password("correct horse battery staple")
|
||||
|
||||
logged_user.uid = ["user"]
|
||||
|
|
@ -82,16 +80,15 @@ def test_edition(
|
|||
logged_user.mail = ["john@doe.com"]
|
||||
logged_user.givenName = None
|
||||
logged_user.jpegPhoto = None
|
||||
logged_user.save(conn=slapd_connection)
|
||||
logged_user.save()
|
||||
|
||||
|
||||
def test_field_permissions_none(
|
||||
testclient, slapd_server, slapd_connection, logged_user
|
||||
):
|
||||
testclient.get("/profile/user", status=200)
|
||||
with testclient.app.app_context():
|
||||
logged_user.telephoneNumber = ["555-666-777"]
|
||||
logged_user.save(conn=slapd_connection)
|
||||
logged_user.save()
|
||||
|
||||
testclient.app.config["ACL"]["DEFAULT"] = {
|
||||
"READ": ["uid"],
|
||||
|
|
@ -105,8 +102,7 @@ def test_field_permissions_none(
|
|||
testclient.post(
|
||||
"/profile/user", {"action": "edit", "telephoneNumber": "000-000-000"}
|
||||
)
|
||||
with testclient.app.app_context():
|
||||
user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
user = User.get(dn=logged_user.dn)
|
||||
assert user.telephoneNumber == ["555-666-777"]
|
||||
|
||||
|
||||
|
|
@ -114,9 +110,8 @@ def test_field_permissions_read(
|
|||
testclient, slapd_server, slapd_connection, logged_user
|
||||
):
|
||||
testclient.get("/profile/user", status=200)
|
||||
with testclient.app.app_context():
|
||||
logged_user.telephoneNumber = ["555-666-777"]
|
||||
logged_user.save(conn=slapd_connection)
|
||||
logged_user.save()
|
||||
|
||||
testclient.app.config["ACL"]["DEFAULT"] = {
|
||||
"READ": ["uid", "telephoneNumber"],
|
||||
|
|
@ -129,8 +124,7 @@ def test_field_permissions_read(
|
|||
testclient.post(
|
||||
"/profile/user", {"action": "edit", "telephoneNumber": "000-000-000"}
|
||||
)
|
||||
with testclient.app.app_context():
|
||||
user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
user = User.get(dn=logged_user.dn)
|
||||
assert user.telephoneNumber == ["555-666-777"]
|
||||
|
||||
|
||||
|
|
@ -138,9 +132,8 @@ def test_field_permissions_write(
|
|||
testclient, slapd_server, slapd_connection, logged_user
|
||||
):
|
||||
testclient.get("/profile/user", status=200)
|
||||
with testclient.app.app_context():
|
||||
logged_user.telephoneNumber = ["555-666-777"]
|
||||
logged_user.save(conn=slapd_connection)
|
||||
logged_user.save()
|
||||
|
||||
testclient.app.config["ACL"]["DEFAULT"] = {
|
||||
"READ": ["uid"],
|
||||
|
|
@ -153,8 +146,7 @@ def test_field_permissions_write(
|
|||
testclient.post(
|
||||
"/profile/user", {"action": "edit", "telephoneNumber": "000-000-000"}
|
||||
)
|
||||
with testclient.app.app_context():
|
||||
user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
user = User.get(dn=logged_user.dn)
|
||||
assert user.telephoneNumber == ["000-000-000"]
|
||||
|
||||
|
||||
|
|
@ -181,7 +173,7 @@ def test_bad_email(testclient, slapd_connection, logged_user):
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
|
||||
logged_user.reload(slapd_connection)
|
||||
logged_user.reload()
|
||||
|
||||
assert ["john@doe.com"] == logged_user.mail
|
||||
|
||||
|
|
@ -194,7 +186,7 @@ def test_surname_is_mandatory(testclient, slapd_connection, logged_user):
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
|
||||
logged_user.reload(slapd_connection)
|
||||
logged_user.reload()
|
||||
|
||||
assert ["Doe"] == logged_user.sn
|
||||
|
||||
|
|
@ -207,7 +199,6 @@ def test_password_change(testclient, slapd_connection, logged_user):
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert logged_user.check_password("new_password")
|
||||
|
||||
res = testclient.get("/profile/user", status=200)
|
||||
|
|
@ -218,7 +209,6 @@ def test_password_change(testclient, slapd_connection, logged_user):
|
|||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
assert "Profile updated successfuly" in res
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert logged_user.check_password("correct horse battery staple")
|
||||
|
||||
|
||||
|
|
@ -230,7 +220,6 @@ def test_password_change_fail(testclient, slapd_connection, logged_user):
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert logged_user.check_password("correct horse battery staple")
|
||||
|
||||
res = testclient.get("/profile/user", status=200)
|
||||
|
|
@ -240,7 +229,6 @@ def test_password_change_fail(testclient, slapd_connection, logged_user):
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
assert logged_user.check_password("correct horse battery staple")
|
||||
|
||||
|
||||
|
|
@ -254,8 +242,7 @@ def test_user_creation_edition_and_deletion(
|
|||
):
|
||||
# The user does not exist.
|
||||
res = testclient.get("/users", status=200)
|
||||
with testclient.app.app_context():
|
||||
assert User.get("george", conn=slapd_connection) is None
|
||||
assert User.get("george") is None
|
||||
assert "george" not in res.text
|
||||
|
||||
# Fill the profile for a new user.
|
||||
|
|
@ -271,10 +258,9 @@ def test_user_creation_edition_and_deletion(
|
|||
|
||||
# User have been created
|
||||
res = res.form.submit(name="action", value="edit", status=302).follow(status=200)
|
||||
with testclient.app.app_context():
|
||||
george = User.get("george", conn=slapd_connection)
|
||||
george.load_groups(conn=slapd_connection)
|
||||
foo_group.reload(slapd_connection)
|
||||
george = User.get("george")
|
||||
george.load_groups()
|
||||
foo_group.reload()
|
||||
assert "George" == george.givenName[0]
|
||||
assert george.groups == [foo_group]
|
||||
assert george.check_password("totoyolo")
|
||||
|
|
@ -290,14 +276,13 @@ def test_user_creation_edition_and_deletion(
|
|||
|
||||
# User have been edited
|
||||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
with testclient.app.app_context():
|
||||
george = User.get("george", conn=slapd_connection)
|
||||
george.load_groups(conn=slapd_connection)
|
||||
george = User.get("george")
|
||||
george.load_groups()
|
||||
assert "Georgio" == george.givenName[0]
|
||||
assert george.check_password("totoyolo")
|
||||
|
||||
foo_group.reload(slapd_connection)
|
||||
bar_group.reload(slapd_connection)
|
||||
foo_group.reload()
|
||||
bar_group.reload()
|
||||
assert george.dn in set(foo_group.member)
|
||||
assert george.dn in set(bar_group.member)
|
||||
assert set(george.groups) == {foo_group, bar_group}
|
||||
|
|
@ -306,8 +291,7 @@ def test_user_creation_edition_and_deletion(
|
|||
|
||||
# User have been deleted.
|
||||
res = res.form.submit(name="action", value="delete", status=302).follow(status=200)
|
||||
with testclient.app.app_context():
|
||||
assert User.get("george", conn=slapd_connection) is None
|
||||
assert User.get("george") is None
|
||||
assert "george" not in res.text
|
||||
|
||||
|
||||
|
|
@ -322,8 +306,7 @@ def test_cn_setting_with_given_name_and_surname(
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=302).follow(status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
george = User.get("george", conn=slapd_connection)
|
||||
george = User.get("george")
|
||||
assert george.cn[0] == "George Abitbol"
|
||||
|
||||
|
||||
|
|
@ -335,8 +318,7 @@ def test_cn_setting_with_surname_only(testclient, slapd_connection, logged_moder
|
|||
|
||||
res = res.form.submit(name="action", value="edit", status=302).follow(status=200)
|
||||
|
||||
with testclient.app.app_context():
|
||||
george = User.get("george", conn=slapd_connection)
|
||||
george = User.get("george")
|
||||
assert george.cn[0] == "Abitbol"
|
||||
|
||||
|
||||
|
|
@ -349,7 +331,7 @@ def test_first_login_mail_button(smtpd, testclient, slapd_connection, logged_adm
|
|||
uid="temp",
|
||||
mail="john@doe.com",
|
||||
)
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
|
||||
res = testclient.get("/profile/temp", status=200)
|
||||
assert "This user does not have a password yet" in res
|
||||
|
|
@ -365,14 +347,14 @@ def test_first_login_mail_button(smtpd, testclient, slapd_connection, logged_adm
|
|||
assert "Send again" in res
|
||||
assert len(smtpd.messages) == 1
|
||||
|
||||
u.reload(slapd_connection)
|
||||
u.reload()
|
||||
u.userPassword = ["{SSHA}fw9DYeF/gHTHuVMepsQzVYAkffGcU8Fz"]
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
|
||||
res = testclient.get("/profile/temp", status=200)
|
||||
assert "This user does not have a password yet" not in res
|
||||
|
||||
u.delete(slapd_connection)
|
||||
u.delete()
|
||||
|
||||
|
||||
def test_email_reset_button(smtpd, testclient, slapd_connection, logged_admin):
|
||||
|
|
@ -385,7 +367,7 @@ def test_email_reset_button(smtpd, testclient, slapd_connection, logged_admin):
|
|||
mail="john@doe.com",
|
||||
userPassword=["{SSHA}fw9DYeF/gHTHuVMepsQzVYAkffGcU8Fz"],
|
||||
)
|
||||
u.save(slapd_connection)
|
||||
u.save()
|
||||
|
||||
res = testclient.get("/profile/temp", status=200)
|
||||
assert "If the user has forgotten his password" in res, res.text
|
||||
|
|
@ -399,7 +381,7 @@ def test_email_reset_button(smtpd, testclient, slapd_connection, logged_admin):
|
|||
assert "Send again" in res
|
||||
assert len(smtpd.messages) == 1
|
||||
|
||||
u.delete(slapd_connection)
|
||||
u.delete()
|
||||
|
||||
|
||||
def test_photo_edition(
|
||||
|
|
@ -417,8 +399,7 @@ def test_photo_edition(
|
|||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
assert "Profile updated successfuly." in res, str(res)
|
||||
|
||||
with testclient.app.app_context():
|
||||
logged_user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
logged_user = User.get(dn=logged_user.dn)
|
||||
|
||||
assert [jpeg_photo] == logged_user.jpegPhoto
|
||||
|
||||
|
|
@ -428,8 +409,7 @@ def test_photo_edition(
|
|||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
assert "Profile updated successfuly." in res, str(res)
|
||||
|
||||
with testclient.app.app_context():
|
||||
logged_user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
logged_user = User.get(dn=logged_user.dn)
|
||||
|
||||
assert [jpeg_photo] == logged_user.jpegPhoto
|
||||
|
||||
|
|
@ -439,8 +419,7 @@ def test_photo_edition(
|
|||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
assert "Profile updated successfuly." in res, str(res)
|
||||
|
||||
with testclient.app.app_context():
|
||||
logged_user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
logged_user = User.get(dn=logged_user.dn)
|
||||
|
||||
assert [] == logged_user.jpegPhoto
|
||||
|
||||
|
|
@ -451,7 +430,6 @@ def test_photo_edition(
|
|||
res = res.form.submit(name="action", value="edit", status=200)
|
||||
assert "Profile updated successfuly." in res, str(res)
|
||||
|
||||
with testclient.app.app_context():
|
||||
logged_user = User.get(dn=logged_user.dn, conn=slapd_connection)
|
||||
logged_user = User.get(dn=logged_user.dn)
|
||||
|
||||
assert [] == logged_user.jpegPhoto
|
||||
|
|
|
|||
Loading…
Reference in a new issue