forked from DGNum/gestioCOF
05eeb6a25c
Refer to allauth doc for an accurate features list: http://django-allauth.readthedocs.io/en/latest/ Users can now change their password, ask for a password reset, or set one if they don't have one. In particular, it allows users whose account has been created via a clipper authentication to configure a password before losing their clipper. Even if they have already lost it, they are able to get one using the "Reset password" functionality. Allauth multiple emails management is deactivated. Requests to the related url redirect to the home page. All the login and logout views are replaced by the allauth' ones. It also concerns the Django and Wagtail admin sites. Note that users are no longer logged out of the clipper CAS server when they authenticated via this server. Instead a message suggests the user to disconnect. Clipper connections and `login_clipper` --------------------------------------- - Non-empty `login_clipper` are now unique among `CofProfile` instances. - They are created once for users with a non-empty 'login_clipper' (with the data migration 0014_create_clipper_connections). - The `login_clipper` of CofProfile instances are sync with their clipper connections: * `CofProfile.sync_clipper_connections` method updates the connections based on `login_clipper`. * Signals receivers `sync_clipper…` update `login_clipper` based on connections creations/updates/deletions. Misc ---- - Add NullCharField (model field) which allows to use `unique=True` on CharField (even with empty strings). - Parts of kfet mixins for TestCase are now in shared.tests.testcase, as they are used elsewhere than in the kfet app.
301 lines
11 KiB
Python
301 lines
11 KiB
Python
import json
|
|
from datetime import timedelta
|
|
from unittest import mock
|
|
|
|
from channels.channel import Group
|
|
from channels.test import ChannelTestCase, WSClient
|
|
from django.contrib.auth.models import AnonymousUser, Permission, User
|
|
from django.test import Client
|
|
from django.utils import timezone
|
|
|
|
from . import OpenKfet, kfet_open
|
|
from .consumers import OpenKfetConsumer
|
|
|
|
|
|
class OpenKfetTest(ChannelTestCase):
|
|
"""OpenKfet object unit-tests suite."""
|
|
|
|
def setUp(self):
|
|
self.kfet_open = OpenKfet()
|
|
|
|
def tearDown(self):
|
|
self.kfet_open.clear_cache()
|
|
|
|
def test_defaults(self):
|
|
"""Default values."""
|
|
self.assertFalse(self.kfet_open.raw_open)
|
|
self.assertIsNone(self.kfet_open.last_update)
|
|
self.assertFalse(self.kfet_open.force_close)
|
|
self.assertFalse(self.kfet_open.is_open)
|
|
|
|
def test_raw_open(self):
|
|
"""Get and set raw_open; last_update is renewed."""
|
|
for raw_open in [True, False]:
|
|
prev_update = self.kfet_open.last_update
|
|
self.kfet_open.raw_open = raw_open
|
|
self.assertEqual(raw_open, self.kfet_open.raw_open)
|
|
self.assertNotEqual(prev_update, self.kfet_open.last_update)
|
|
|
|
def test_force_close(self):
|
|
"""Get and set force_close."""
|
|
for force_close in [True, False]:
|
|
self.kfet_open.force_close = force_close
|
|
self.assertEqual(force_close, self.kfet_open.force_close)
|
|
|
|
def test_is_open(self):
|
|
"""If force_close is disabled, is_open is raw_open."""
|
|
self.kfet_open.force_close = False
|
|
for raw_open in [True, False]:
|
|
self.kfet_open.raw_open = raw_open
|
|
self.assertEqual(raw_open, self.kfet_open.is_open)
|
|
|
|
def test_is_open_force_close(self):
|
|
"""If force_close is enabled, is_open is False."""
|
|
self.kfet_open.force_close = True
|
|
for raw_open in [True, False]:
|
|
self.kfet_open.raw_open = raw_open
|
|
self.assertFalse(self.kfet_open.is_open)
|
|
|
|
def test_status(self):
|
|
# (raw_open, force_close, expected status, expected admin)
|
|
cases = [
|
|
(False, False, OpenKfet.CLOSED, OpenKfet.CLOSED),
|
|
(False, True, OpenKfet.CLOSED, OpenKfet.CLOSED),
|
|
(True, False, OpenKfet.OPENED, OpenKfet.OPENED),
|
|
(True, True, OpenKfet.CLOSED, OpenKfet.FAKE_CLOSED),
|
|
]
|
|
for raw_open, force_close, exp_stat, exp_adm_stat in cases:
|
|
self.kfet_open.raw_open = raw_open
|
|
self.kfet_open.force_close = force_close
|
|
self.assertEqual(exp_stat, self.kfet_open.status())
|
|
self.assertEqual(exp_adm_stat, self.kfet_open.admin_status())
|
|
|
|
def test_status_unknown(self):
|
|
self.kfet_open.raw_open = True
|
|
self.kfet_open._last_update = timezone.now() - timedelta(days=30)
|
|
self.assertEqual(OpenKfet.UNKNOWN, self.kfet_open.status())
|
|
|
|
def test_export_user(self):
|
|
"""Export is limited for an anonymous user."""
|
|
export = self.kfet_open.export(AnonymousUser())
|
|
self.assertSetEqual(set(["status"]), set(export))
|
|
|
|
def test_export_team(self):
|
|
"""Export all values for a team member."""
|
|
user = User.objects.create_user("team", "", "team")
|
|
user.user_permissions.add(Permission.objects.get(codename="is_team"))
|
|
export = self.kfet_open.export(user)
|
|
self.assertSetEqual(set(["status", "admin_status", "force_close"]), set(export))
|
|
|
|
def test_send_ws(self):
|
|
Group("kfet.open.base").add("test.open.base")
|
|
Group("kfet.open.team").add("test.open.team")
|
|
|
|
self.kfet_open.send_ws()
|
|
|
|
recv_base = self.get_next_message("test.open.base", require=True)
|
|
base = json.loads(recv_base["text"])
|
|
self.assertSetEqual(set(["status"]), set(base))
|
|
|
|
recv_admin = self.get_next_message("test.open.team", require=True)
|
|
admin = json.loads(recv_admin["text"])
|
|
self.assertSetEqual(set(["status", "admin_status", "force_close"]), set(admin))
|
|
|
|
|
|
class OpenKfetViewsTest(ChannelTestCase):
|
|
"""OpenKfet views unit-tests suite."""
|
|
|
|
def setUp(self):
|
|
# get some permissions
|
|
perms = {
|
|
"kfet.is_team": Permission.objects.get(codename="is_team"),
|
|
"kfet.can_force_close": Permission.objects.get(codename="can_force_close"),
|
|
}
|
|
|
|
# authenticated user and its client
|
|
self.u = User.objects.create_user("user", "", "user")
|
|
self.c = Client()
|
|
self.c.login(username="user", password="user")
|
|
|
|
# team user and its clients
|
|
self.t = User.objects.create_user("team", "", "team")
|
|
self.t.user_permissions.add(perms["kfet.is_team"])
|
|
self.c_t = Client()
|
|
self.c_t.login(username="team", password="team")
|
|
|
|
# admin user and its client
|
|
self.a = User.objects.create_user("admin", "", "admin")
|
|
self.a.user_permissions.add(
|
|
perms["kfet.is_team"], perms["kfet.can_force_close"]
|
|
)
|
|
self.c_a = Client()
|
|
self.c_a.login(username="admin", password="admin")
|
|
|
|
def tearDown(self):
|
|
kfet_open.clear_cache()
|
|
|
|
def test_door(self):
|
|
"""Edit raw_status."""
|
|
for sent, expected in [(1, True), (0, False)]:
|
|
resp = Client().post(
|
|
"/k-fet/open/raw_open", {"raw_open": sent, "token": "plop"}
|
|
)
|
|
self.assertEqual(200, resp.status_code)
|
|
self.assertEqual(expected, kfet_open.raw_open)
|
|
|
|
def test_force_close(self):
|
|
"""Edit force_close."""
|
|
for sent, expected in [(1, True), (0, False)]:
|
|
resp = self.c_a.post("/k-fet/open/force_close", {"force_close": sent})
|
|
self.assertEqual(200, resp.status_code)
|
|
self.assertEqual(expected, kfet_open.force_close)
|
|
|
|
def test_force_close_forbidden(self):
|
|
"""Can't edit force_close without kfet.can_force_close permission."""
|
|
clients = [Client(), self.c, self.c_t]
|
|
for client in clients:
|
|
resp = client.post("/k-fet/open/force_close", {"force_close": 0})
|
|
self.assertEqual(403, resp.status_code)
|
|
|
|
|
|
class OpenKfetConsumerTest(ChannelTestCase):
|
|
"""OpenKfet consumer unit-tests suite."""
|
|
|
|
def test_standard_user(self):
|
|
"""Lambda user is added to kfet.open.base group."""
|
|
# setup anonymous client
|
|
c = WSClient()
|
|
|
|
# connect
|
|
c.send_and_consume(
|
|
"websocket.connect", path="/ws/k-fet/open", fail_on_none=True
|
|
)
|
|
|
|
# initialization data is replied on connection
|
|
self.assertIsNotNone(c.receive())
|
|
|
|
# client belongs to the 'kfet.open' group...
|
|
OpenKfetConsumer.group_send("kfet.open.base", {"test": "plop"})
|
|
self.assertEqual(c.receive(), {"test": "plop"})
|
|
|
|
# ...but not to the 'kfet.open.admin' one
|
|
OpenKfetConsumer.group_send("kfet.open.team", {"test": "plop"})
|
|
self.assertIsNone(c.receive())
|
|
|
|
def test_team_user(self):
|
|
"""Team user is added to kfet.open.team group."""
|
|
# setup team user and its client
|
|
t = User.objects.create_user("team", "", "team")
|
|
t.user_permissions.add(Permission.objects.get(codename="is_team"))
|
|
c = WSClient()
|
|
c.force_login(t)
|
|
|
|
# connect
|
|
c.send_and_consume(
|
|
"websocket.connect", path="/ws/k-fet/open", fail_on_none=True
|
|
)
|
|
|
|
# initialization data is replied on connection
|
|
self.assertIsNotNone(c.receive())
|
|
|
|
# client belongs to the 'kfet.open.admin' group...
|
|
OpenKfetConsumer.group_send("kfet.open.team", {"test": "plop"})
|
|
self.assertEqual(c.receive(), {"test": "plop"})
|
|
|
|
# ... but not to the 'kfet.open' one
|
|
OpenKfetConsumer.group_send("kfet.open.base", {"test": "plop"})
|
|
self.assertIsNone(c.receive())
|
|
|
|
|
|
class OpenKfetScenarioTest(ChannelTestCase):
|
|
"""OpenKfet functionnal tests suite."""
|
|
|
|
def setUp(self):
|
|
|
|
# anonymous client (for views)
|
|
self.c = Client()
|
|
# anonymous client (for websockets)
|
|
self.c_ws = WSClient()
|
|
|
|
# root user
|
|
self.r = User.objects.create_superuser("root", "", "root")
|
|
# its client (for views)
|
|
self.r_c = Client()
|
|
self.r_c.login(username="root", password="root")
|
|
# its client (for websockets)
|
|
self.r_c_ws = WSClient()
|
|
self.r_c_ws.force_login(self.r)
|
|
|
|
def tearDown(self):
|
|
kfet_open.clear_cache()
|
|
|
|
def ws_connect(self, ws_client):
|
|
ws_client.send_and_consume(
|
|
"websocket.connect", path="/ws/k-fet/open", fail_on_none=True
|
|
)
|
|
return ws_client.receive(json=True)
|
|
|
|
def test_scenario_0(self):
|
|
"""Clients connect."""
|
|
# test for anonymous user
|
|
msg = self.ws_connect(self.c_ws)
|
|
self.assertSetEqual(set(["status"]), set(msg))
|
|
|
|
# test for root user
|
|
msg = self.ws_connect(self.r_c_ws)
|
|
self.assertSetEqual(set(["status", "admin_status", "force_close"]), set(msg))
|
|
|
|
def test_scenario_1(self):
|
|
"""Clients connect, door opens, enable force close."""
|
|
self.ws_connect(self.c_ws)
|
|
self.ws_connect(self.r_c_ws)
|
|
|
|
# door sent "I'm open!"
|
|
self.c.post("/k-fet/open/raw_open", {"raw_open": True, "token": "plop"})
|
|
|
|
# anonymous user agree
|
|
msg = self.c_ws.receive(json=True)
|
|
self.assertEqual(OpenKfet.OPENED, msg["status"])
|
|
|
|
# root user too
|
|
msg = self.r_c_ws.receive(json=True)
|
|
self.assertEqual(OpenKfet.OPENED, msg["status"])
|
|
self.assertEqual(OpenKfet.OPENED, msg["admin_status"])
|
|
|
|
# admin says "no it's closed"
|
|
self.r_c.post("/k-fet/open/force_close", {"force_close": True})
|
|
|
|
# so anonymous user see it's closed
|
|
msg = self.c_ws.receive(json=True)
|
|
self.assertEqual(OpenKfet.CLOSED, msg["status"])
|
|
|
|
# root user too
|
|
msg = self.r_c_ws.receive(json=True)
|
|
self.assertEqual(OpenKfet.CLOSED, msg["status"])
|
|
# but root knows things
|
|
self.assertEqual(OpenKfet.FAKE_CLOSED, msg["admin_status"])
|
|
self.assertTrue(msg["force_close"])
|
|
|
|
def test_scenario_2(self):
|
|
"""Starting falsely closed, clients connect, disable force close."""
|
|
kfet_open.raw_open = True
|
|
kfet_open.force_close = True
|
|
|
|
msg = self.ws_connect(self.c_ws)
|
|
self.assertEqual(OpenKfet.CLOSED, msg["status"])
|
|
|
|
msg = self.ws_connect(self.r_c_ws)
|
|
self.assertEqual(OpenKfet.CLOSED, msg["status"])
|
|
self.assertEqual(OpenKfet.FAKE_CLOSED, msg["admin_status"])
|
|
self.assertTrue(msg["force_close"])
|
|
|
|
self.r_c.post("/k-fet/open/force_close", {"force_close": False})
|
|
|
|
msg = self.c_ws.receive(json=True)
|
|
self.assertEqual(OpenKfet.OPENED, msg["status"])
|
|
|
|
msg = self.r_c_ws.receive(json=True)
|
|
self.assertEqual(OpenKfet.OPENED, msg["status"])
|
|
self.assertEqual(OpenKfet.OPENED, msg["admin_status"])
|
|
self.assertFalse(msg["force_close"])
|