gestioCOF/kfet/open/tests.py
Aurélien Delobelle b8110c11a4 kfet.open
kfet.open app
- Base data (raw_open, last_update...) is stored and shared through cache system.
- 2 websockets groups: one for team users, one for other users.
- UI is initialized and kept up-to-date with WS.
- raw_open and force_close can be updated with standard HTTP requests.
  At this time, there isn't any restriction on raw_open view. Common sense tell us
  to change this behavior.

Misc
- Clean channels routing.
- 'PermConsumerMixin': user who sent the message is available as argument in
connection_groups method, which returns groups to which the user should be
appended on websocket connection (and discarded on disconnection).
- New kfet.utils module: should be used for mixins, whatever is useful and not concerns
the kfet app.
- Clean JS dependencies.
2017-06-21 07:08:28 +02:00

303 lines
10 KiB
Python

import json
from django.contrib.auth.models import AnonymousUser, Permission, User
from django.test import Client
from channels.channel import Group
from channels.test import ChannelTestCase, WSClient
from . import kfet_open, OpenKfet
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_export_user(self):
"""Export is limited for an anonymous user."""
export = self.kfet_open.export(AnonymousUser())
self.assertSetEqual(
set(['is_open', 'last_update']),
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(['is_open', 'last_update', 'raw_open', 'force_close']),
set(export),
)
def test_export(self):
"""Export all by default."""
export = self.kfet_open.export()
self.assertSetEqual(
set(['is_open', 'last_update', 'raw_open', '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(['is_open', 'last_update']),
set(base),
)
recv_admin = self.get_next_message('test.open.team', require=True)
admin = json.loads(recv_admin['text'])
self.assertSetEqual(
set(['is_open', 'last_update', 'raw_open', '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})
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(['is_open', 'last_update']),
set(msg),
)
# test for root user
msg = self.ws_connect(self.r_c_ws)
self.assertSetEqual(
set(['is_open', 'last_update', 'raw_open', '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})
# anonymous user agree
msg = self.c_ws.receive(json=True)
self.assertTrue(msg['is_open'])
# root user too
msg = self.r_c_ws.receive(json=True)
self.assertTrue(msg['is_open'])
self.assertTrue(msg['raw_open'])
# 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.assertFalse(msg['is_open'])
# root user too
msg = self.r_c_ws.receive(json=True)
self.assertFalse(msg['is_open'])
# but root knows things
self.assertTrue(msg['raw_open'])
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.assertFalse(msg['is_open'])
msg = self.ws_connect(self.r_c_ws)
self.assertFalse(msg['is_open'])
self.assertTrue(msg['raw_open'])
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.assertTrue(msg['is_open'])
msg = self.r_c_ws.receive(json=True)
self.assertTrue(msg['is_open'])
self.assertTrue(msg['raw_open'])
self.assertFalse(msg['force_close'])