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, '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(['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, 'token': 'plop', }) # 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'])