gestioCOF/kfet/open/tests.py

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'])