forked from DGNum/gestioCOF
5673fabeff
Status is mainly computed in Python. That fix inconsistent datetime between client and server. Client only receives status and keep timestamp of last received ws msg.
322 lines
11 KiB
Python
322 lines
11 KiB
Python
import json
|
|
from datetime import timedelta
|
|
|
|
from django.contrib.auth.models import AnonymousUser, Permission, User
|
|
from django.test import Client
|
|
from django.utils import timezone
|
|
|
|
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_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'])
|