forked from DGNum/gestioCOF
Better status management.
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.
This commit is contained in:
Aurélien Delobelle
parent
19847ac9d8
commit
5673fabeff
3 changed files with 78 additions and 53 deletions
|
|
@ -1,3 +1,5 @@
|
|||
from datetime import timedelta
|
||||
|
||||
from django.utils import timezone
|
||||
|
||||
from ..decorators import kfet_is_team
|
||||
|
|
@ -13,6 +15,17 @@ class OpenKfet(CachedMixin, object):
|
|||
Current state persists through cache.
|
||||
|
||||
"""
|
||||
# status is unknown after this duration
|
||||
time_unknown = timedelta(minutes=15)
|
||||
|
||||
# status
|
||||
OPENED = 'opened'
|
||||
CLOSED = 'closed'
|
||||
UNKNOWN = 'unknown'
|
||||
# admin status
|
||||
FAKE_CLOSED = 'fake_closed'
|
||||
|
||||
# cached attributes config
|
||||
cached = {
|
||||
'_raw_open': False,
|
||||
'_last_update': None,
|
||||
|
|
@ -40,6 +53,19 @@ class OpenKfet(CachedMixin, object):
|
|||
"""Take into account force_close."""
|
||||
return False if self.force_close else self.raw_open
|
||||
|
||||
def status(self):
|
||||
if (self.last_update is None or
|
||||
timezone.now() - self.last_update >= self.time_unknown):
|
||||
return self.UNKNOWN
|
||||
return self.OPENED if self.is_open else self.CLOSED
|
||||
|
||||
def admin_status(self, status=None):
|
||||
if status is None:
|
||||
status = self.status()
|
||||
if status == self.CLOSED and self.raw_open:
|
||||
return self.FAKE_CLOSED
|
||||
return status
|
||||
|
||||
def _export(self):
|
||||
"""Export internal state.
|
||||
|
||||
|
|
@ -51,27 +77,26 @@ class OpenKfet(CachedMixin, object):
|
|||
- base for others.
|
||||
|
||||
"""
|
||||
status = self.status()
|
||||
base = {
|
||||
'is_open': self.is_open,
|
||||
'last_update': self.last_update,
|
||||
'status': status,
|
||||
}
|
||||
restrict = {
|
||||
'raw_open': self.raw_open,
|
||||
'admin_status': self.admin_status(status),
|
||||
'force_close': self.force_close,
|
||||
}
|
||||
return base, {**base, **restrict}
|
||||
|
||||
def export(self, user=None):
|
||||
"""Export internal state.
|
||||
def export(self, user):
|
||||
"""Export internal state for a given user.
|
||||
|
||||
Returns:
|
||||
(dict): Internal state. Only variables visible for the user are
|
||||
exported, according to its permissions. If no user is given, it
|
||||
returns all available values.
|
||||
exported, according to its permissions.
|
||||
|
||||
"""
|
||||
base, team = self._export()
|
||||
return team if user is None or kfet_is_team(user) else base
|
||||
return team if kfet_is_team(user) else base
|
||||
|
||||
def send_ws(self):
|
||||
"""Send internal state to websocket channels."""
|
||||
|
|
|
|||
|
|
@ -54,26 +54,13 @@ OpenKfet.prototype = {
|
|||
},
|
||||
|
||||
refresh: function(data) {
|
||||
if (data)
|
||||
if (data) {
|
||||
$.extend(this, data);
|
||||
this.refresh_status();
|
||||
this.refresh_dom();
|
||||
},
|
||||
|
||||
refresh_status: function() {
|
||||
// find new status
|
||||
let status = this.UNKNOWN;
|
||||
if (this.is_recent)
|
||||
status = this.is_open ? this.OPENED : this.CLOSED;
|
||||
this.status = status;
|
||||
|
||||
// admin specific
|
||||
if (this.admin) {
|
||||
let admin_status = status;
|
||||
if (status == this.CLOSED && this.raw_open)
|
||||
admin_status = this.FAKE_CLOSED;
|
||||
this.admin_status = admin_status;
|
||||
this.last_update = moment();
|
||||
}
|
||||
if (!this.is_recent)
|
||||
this.status = this.UNKNOWN;
|
||||
this.refresh_dom();
|
||||
},
|
||||
|
||||
refresh_dom: function() {
|
||||
|
|
@ -96,7 +83,7 @@ OpenKfet.prototype = {
|
|||
}
|
||||
},
|
||||
|
||||
toggle_force_close: function(new_value, password) {
|
||||
toggle_force_close: function(password) {
|
||||
$.post({
|
||||
url: this.force_close_url,
|
||||
data: {force_close: !this.force_close},
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
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
|
||||
|
|
@ -54,11 +56,30 @@ class OpenKfetTest(ChannelTestCase):
|
|||
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(['is_open', 'last_update']),
|
||||
set(['status']),
|
||||
set(export),
|
||||
)
|
||||
|
||||
|
|
@ -68,15 +89,7 @@ class OpenKfetTest(ChannelTestCase):
|
|||
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(['status', 'admin_status', 'force_close']),
|
||||
set(export),
|
||||
)
|
||||
|
||||
|
|
@ -89,14 +102,14 @@ class OpenKfetTest(ChannelTestCase):
|
|||
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(['status']),
|
||||
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(['status', 'admin_status', 'force_close']),
|
||||
set(admin),
|
||||
)
|
||||
|
||||
|
|
@ -240,14 +253,14 @@ class OpenKfetScenarioTest(ChannelTestCase):
|
|||
# test for anonymous user
|
||||
msg = self.ws_connect(self.c_ws)
|
||||
self.assertSetEqual(
|
||||
set(['is_open', 'last_update']),
|
||||
set(['status']),
|
||||
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(['status', 'admin_status', 'force_close']),
|
||||
set(msg),
|
||||
)
|
||||
|
||||
|
|
@ -264,25 +277,25 @@ class OpenKfetScenarioTest(ChannelTestCase):
|
|||
|
||||
# anonymous user agree
|
||||
msg = self.c_ws.receive(json=True)
|
||||
self.assertTrue(msg['is_open'])
|
||||
self.assertEqual(OpenKfet.OPENED, msg['status'])
|
||||
|
||||
# root user too
|
||||
msg = self.r_c_ws.receive(json=True)
|
||||
self.assertTrue(msg['is_open'])
|
||||
self.assertTrue(msg['raw_open'])
|
||||
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.assertFalse(msg['is_open'])
|
||||
self.assertEqual(OpenKfet.CLOSED, msg['status'])
|
||||
|
||||
# root user too
|
||||
msg = self.r_c_ws.receive(json=True)
|
||||
self.assertFalse(msg['is_open'])
|
||||
self.assertEqual(OpenKfet.CLOSED, msg['status'])
|
||||
# but root knows things
|
||||
self.assertTrue(msg['raw_open'])
|
||||
self.assertEqual(OpenKfet.FAKE_CLOSED, msg['admin_status'])
|
||||
self.assertTrue(msg['force_close'])
|
||||
|
||||
def test_scenario_2(self):
|
||||
|
|
@ -291,19 +304,19 @@ class OpenKfetScenarioTest(ChannelTestCase):
|
|||
kfet_open.force_close = True
|
||||
|
||||
msg = self.ws_connect(self.c_ws)
|
||||
self.assertFalse(msg['is_open'])
|
||||
self.assertEqual(OpenKfet.CLOSED, msg['status'])
|
||||
|
||||
msg = self.ws_connect(self.r_c_ws)
|
||||
self.assertFalse(msg['is_open'])
|
||||
self.assertTrue(msg['raw_open'])
|
||||
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.assertTrue(msg['is_open'])
|
||||
self.assertEqual(OpenKfet.OPENED, msg['status'])
|
||||
|
||||
msg = self.r_c_ws.receive(json=True)
|
||||
self.assertTrue(msg['is_open'])
|
||||
self.assertTrue(msg['raw_open'])
|
||||
self.assertEqual(OpenKfet.OPENED, msg['status'])
|
||||
self.assertEqual(OpenKfet.OPENED, msg['admin_status'])
|
||||
self.assertFalse(msg['force_close'])
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue