test_labdiscoverylib.py

import os
import sys
import json
import time
import base64
import datetime
import requests
import threading

from io import StringIO
from flask import Flask, url_for, render_template_string, g, session
import flask.cli as flask_cli

import labdiscoverylib
from labdiscoverylib.tasks import _TaskRunner
import unittest
from click.testing import CliRunner

try:
    from flask_socketio import SocketIO, SocketIOTestClient
except ImportError:
    print()
    print("Important: if you do not have SocketIO you can't run these tests")
    print()
    raise


os.environ['FLASK_APP'] = 'fake.py' # Overrided later
cmp = lambda a, b: a.__cmp__(b)

class StdWrap(object):
    def __enter__(self):
        self.sysout = sys.stdout
        self.syserr = sys.stderr
        self.fake_stdout = sys.stdout = StringIO()
        self.fake_stderr = sys.stderr = StringIO()

    def __exit__(self, *args, **kwargs):
        sys.stdout = self.sysout
        sys.stderr = self.syserr

class WithoutWebLabTest(unittest.TestCase):
    def test_extension(self):
        app = Flask(__name__)
        with self.assertRaises(labdiscoverylib.WebLabNotInitializedError):
            with app.app_context():
                labdiscoverylib._current_weblab()

class BaseWebLabTest(unittest.TestCase):
    def get_config(self):
        return {
            'SECRET_KEY': 'super-secret',
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': self.server_name,
            'WEBLAB_SCHEME': 'https',
            'WEBLAB_AUTOCLEAN_THREAD': False, # No thread
            'WEBLAB_TASK_THREADS_PROCESS': 0, # No thread
        }

    def create_weblab(self):
        labdiscoverylib._CleanerThread.created = False
        self.weblab = labdiscoverylib.WebLab()
        self.app = Flask(__name__)
        flask_cli.locate_app = lambda *args: self.app
        self.server_name = 'localhost:5000'
        self.app.config.update(self.get_config())
        self.auth_headers = {
            'Authorization': 'Basic ' + base64.encodebytes(b'lde:password').decode('utf8').strip(),
        }
        self.wrong_auth_headers = {
            'Authorization': 'Basic ' + base64.encodebytes(b'wrong_lde:wrong_password').decode('utf8').strip(),
        }

        @self.weblab.task(unique='global')
        def task_before_init():
            return True

        self.weblab.init_app(self.app)
        self.weblab._backend.client.flushall()

        @self.weblab.on_start
        def on_start(client_data, server_data):
            self.on_start(client_data, server_data)

        @self.weblab.on_dispose
        def on_dispose():
            self.on_dispose()

        @self.weblab.initial_url
        def initial_url():
            return url_for('lab')

        @self.app.route('/lab/')
        @labdiscoverylib.requires_login
        def lab():
            return self.lab()

        @self.app.route('/lab/active')
        @labdiscoverylib.requires_active
        def lab_active():
            return self.lab()

        @self.app.route('/logout')
        @labdiscoverylib.requires_active
        def logout():
            labdiscoverylib.logout()
            return "logout"

        @self.app.route('/poll')
        @labdiscoverylib.requires_active
        def poll():
            labdiscoverylib.poll()
            labdiscoverylib.poll() # Twice so as to test g.poll_requested
            return "poll"

        @self.weblab.task()
        def task():
            return self.task()

        with self.assertRaises(ValueError) as cm:
            @self.weblab.task()
            def task():
                self.task()

        self.assertIn("same name", str(cm.exception))

        with self.assertRaises(ValueError) as cm:
            # 43 characters is the length of create_token()
            self.assertEquals(43, len(labdiscoverylib.create_token()))

            # Therefore, having a task with that name length is forbidden
            # to avoid any potential issue in get_task
            @self.weblab.task()
            def abc0123456789012345678901234567890123456789():
                pass
           
        self.assertIn("number of characters", str(cm.exception))

        self.current_task = task

    def get_json(self, rv):
        data = rv.get_data(as_text=True)
        try:
            return json.loads(data)
        except Exception as e:
            raise Exception(f"Error loading json: {e} in {data}")

    def get_text(self, rv):
        return rv.get_data(as_text=True)

    def on_start(self, client_data, server_data):
        pass

    def on_dispose(self):
        pass

    def lab(self):
        return ":-)"

    def task(self):
        pass

    def setUp(self):
        self.create_weblab()

    def tearDown(self):
        self.weblab._cleanup()

class WebLabApiTest(BaseWebLabTest):
    def test_api(self):
        with self.app.test_client() as client:
            result = self.get_json(client.get('/ldl/sessions/api'))
            self.assertEquals(result['api_version'], '1')

    def test_weblab_test_without_auth(self):
        with self.app.test_client() as client:
            result = self.get_json(client.get('/ldl/sessions/test'))
            self.assertEquals(result['valid'], False)
            self.assertIn("no username", result['error_messages'][0])

    def test_weblab_test_with_wrong_auth(self):
        with StdWrap():
            with self.app.test_client() as client:
                result = self.get_json(client.get('/ldl/sessions/test', headers=self.wrong_auth_headers))
                self.assertEquals(result['valid'], False)
                self.assertIn("wrong username", result['error_messages'][0])

    def test_weblab_test_with_right_auth(self):
        with self.app.test_client() as client:
            result = self.get_json(client.get('/ldl/sessions/test', headers=self.auth_headers))
            self.assertEquals(result['valid'], True)

    def test_weblab_status_with_wrong_auth(self):
        with StdWrap():
            with self.app.test_client() as client:
                result = self.get_text(client.get('/ldl/sessions/<invalid>/status', headers=self.wrong_auth_headers))
                self.assertIn("seem to be", result)

class SimpleUnauthenticatedTest(BaseWebLabTest):
    def test_token(self):
        token1 = self.weblab.create_token()
        token2 = self.weblab.create_token()
        self.assertNotEquals(token1, token2)

    def test_task_not_found(self):
        task = self.weblab.get_task("does.not.exist")
        self.assertIsNone(task)

    def test_callback_initial_url(self):
        self.weblab._initial_url = None

        with StdWrap():
            with self.app.test_client() as client:
                result = self.get_text(client.get('/callback/session.not.found'))

        self.assertIn('ERROR', result)
        self.assertIn('weblab.initial_url', result)

    def test_callback(self):
        with self.app.test_client() as client:
            result = self.get_text(client.get('/callback/session.not.found'))
            self.assertIn('forbidden', result)

    def test_anonymous(self):
        with self.app.test_client() as client:
            client.get('/lab/')
            self.assertTrue(labdiscoverylib.weblab_user.is_anonymous)
            self.assertFalse(labdiscoverylib.weblab_user.active)
            self.assertIsNone(labdiscoverylib.weblab_user.locale)
            self.assertIsNotNone(labdiscoverylib.weblab_user.data)
            with self.assertRaises(TypeError):
                self.assertEquals(0, len(labdiscoverylib.weblab_user.data))

    def test_anonymous_on_active(self):
        with self.app.test_client() as client:
            rv = client.get('/lab/active')
            self.assertIn("forbidden", self.get_text(rv))

    def test_poll_url(self):
        with self.app.test_client() as client:
            client.get('/lab/')
            url = url_for('weblab_poll_url', session_id='does.not.exist')
            result = self.get_json(client.get(url))
            self.assertIn("Different session", result['reason'])

            with client.session_transaction() as sess:
                sess[self.weblab._session_id_name] = 'does.not.exist'

            result = self.get_json(client.get(url))
            self.assertIn("Not found", result['reason'])

    def test_logout_url(self):
        with self.app.test_client() as client:
            client.get('/lab/')
            url = url_for('weblab_logout_url', session_id='does.not.exist')
            result = self.get_json(client.get(url))
            self.assertIn("Different session", result['reason'])

    def test_poll_script(self):
        with self.app.test_client() as client:
            client.get('/lab/')
            result = render_template_string("{{ weblab_poll_script() }}")
            self.assertIn('session_id not found', result)

    def test_unauthorized(self):
         with self.app.test_client() as client:
            result = self.get_text(client.get('/lab/'))
            self.assertIn("Access forbidden", result)

    def test_dispose_wrong_requests(self):
        with self.app.test_client() as client:
            request_data = {
            }
            rv = client.post('/ldl/sessions/{}'.format('foo'), data=json.dumps(request_data), headers=self.auth_headers)
            self.assertIn("Unknown", self.get_json(rv)['message'])

            request_data = {
                'action': 'look at the mountains'
            }
            rv = client.post('/ldl/sessions/{}'.format('foo'), data=json.dumps(request_data), headers=self.auth_headers)
            self.assertIn("Unknown", self.get_json(rv)['message'])

            request_data = {
                'action': 'delete'
            }
            rv = client.post('/ldl/sessions/{}'.format('does.not.exist'), data=json.dumps(request_data), headers=self.auth_headers)
            self.assertIn("Not found", self.get_json(rv)['message'])

class SimpleNoTimeoutUnauthenticatedTest(BaseWebLabTest):
    def get_config(self):
        config = super(SimpleNoTimeoutUnauthenticatedTest, self).get_config()
        config['WEBLAB_TIMEOUT'] = 0
        return config

    def test_poll_script_timeout(self):
        with self.app.test_client() as client:
            client.get('/lab/')
            result = render_template_string("{{ weblab_poll_script() }}")
            self.assertIn('timeout is 0', result)

class UnauthorizedLinkSimpleTest(BaseWebLabTest):
    def get_config(self):
        config = super(UnauthorizedLinkSimpleTest, self).get_config()
        config['WEBLAB_UNAUTHORIZED_LINK'] = 'http://mylink'
        return config

    def test_unauthorized_link(self):
         with self.app.test_client() as client:
            rv = client.get('/lab/', follow_redirects=False)
            self.assertEquals(rv.location, 'http://mylink')

class UnauthorizedTemplateSimpleTest(BaseWebLabTest):
    def get_config(self):
        config = super(UnauthorizedTemplateSimpleTest, self).get_config()
        config['WEBLAB_UNAUTHORIZED_TEMPLATE'] = 'mytemplate.html'
        return config

    def test_unauthorized_template(self):
        def new_render_template(location):
            return location

        old_render_template = labdiscoverylib.render_template

        with self.app.test_client() as client:
            labdiscoverylib.render_template = new_render_template
            try:
                lab_result = self.get_text(client.get('/lab/'))
            finally:
                labdiscoverylib.render_template = old_render_template

            self.assertEquals(lab_result, 'mytemplate.html')

class TestNewUserError(Exception):
    pass

class BaseSessionWebLabTest(BaseWebLabTest):
    def setUp(self):
        super(BaseSessionWebLabTest, self).setUp()
        # self.weblab_client is stateless, sessionless
        self.weblab_client = self.app.test_client()
        # while self.client represents the user web browser
        self.client = self.app.test_client(use_cookies=True)
        self.client.__enter__()

    def tearDown(self):
        self.client.__exit__(None, None, None)
        super(BaseSessionWebLabTest, self).tearDown()

    def new_user(self, name='Jim Smith', username='jim.smith', username_unique='jim.smith@labsland', 
                 assigned_time=300, back='http://labsland.com', language='en',
                 experiment_name='mylab', category_name='Lab experiments', use_timestamp=True):
        assigned_time = float(assigned_time)
        
        now = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)
        self._start_date = now

        request_data = {
            'request': {
                'locale': language,
                'ldeReservationId': 'foobar',
                'user': {
                    'in_test': True,
                },
                'server': {
                    'myvar': 'myvalue',
                },
                'backUrl': back,
            },
            'laboratory': {
                'name': experiment_name,
                'category': category_name,
            },
            'user': {
                'username': username,
                'fullName': name,
                'unique': username_unique,
            },
            'schedule': {
                'start': now.isoformat(),
                'length': assigned_time,
            }
        }

        rv = self.weblab_client.post('/ldl/sessions/', data=json.dumps(request_data), headers=self.auth_headers)
        response = self.get_json(rv)
        if 'session_id' in response:
            self.session_id = response['session_id']
            launch_url = response['url']
            relative_launch_url = launch_url.split(self.server_name, 1)[1]
            return launch_url, self.session_id
        raise TestNewUserError(response['message'])

    
    def status(self, session_id = None):
        if session_id is None:
            session_id = self.session_id

        rv = self.weblab_client.get('/ldl/sessions/{}/status'.format(session_id), headers=self.auth_headers)
        return self.get_json(rv)

    def dispose(self, session_id = None):
        if session_id is None:
            session_id = self.session_id

        request_data = {
            'action': 'delete',
        }
        rv = self.weblab_client.post('/ldl/sessions/{}'.format(session_id), data=json.dumps(request_data), headers=self.auth_headers)
        return self.get_json(rv)

class UserTest(BaseSessionWebLabTest):

    def lab(self):
        task = self.current_task.delay()
        # check that it's a dictionary
        if labdiscoverylib.weblab_user.active:
            labdiscoverylib.weblab_user.data['foo'] = 'bar'

        # And in any case build another
        if labdiscoverylib.weblab_user.active:
            # Optional, forces an immediate synchronization
            labdiscoverylib.weblab_user.data = {'foo': 'bar'}
        
        return render_template_string("""@@task@@%s@@task@@{{ weblab_poll_script() }}
            {{ weblab_poll_script(logout_on_close=True, callback='myfunc') }}""" % task.task_id)

    def task(self):
        self.counter += 1
        if not labdiscoverylib.current_task_stopping:
            return "ERROR"
        labdiscoverylib.current_task.data = {'inside': 'previous'}
        labdiscoverylib.current_task.update_data({'inside': 'yes'})
        labdiscoverylib.current_task.store()
        labdiscoverylib.weblab_user.data.retrieve()
        return [ self.counter, labdiscoverylib.weblab_user.data['foo'] ]

    def test_simple(self):
        # New user 
        launch_url1, session_id1 = self.new_user(language='es_ES')

        # counter is zero
        self.counter = 0
        
        # We call the relative_launch_url. It is redirected to the lab, which
        # starts a new task, which establishes that counter is zero
        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))

        self.assertEquals(labdiscoverylib.weblab_user.session_id, session_id1)
        self.assertEquals(labdiscoverylib.weblab_user.locale, 'es_ES')
        self.assertEquals(labdiscoverylib.weblab_user.full_name, 'Jim Smith')
        self.assertEquals(labdiscoverylib.weblab_user.experiment_name, 'mylab')
        self.assertEquals(labdiscoverylib.weblab_user.category_name, 'Lab experiments')
        self.assertEquals(labdiscoverylib.weblab_user.experiment_id, 'mylab@Lab experiments')
        self.assertEquals(labdiscoverylib.weblab_user.request_client_data['in_test'], True)
        self.assertEquals(labdiscoverylib.weblab_user.request_server_data['myvar'], 'myvalue')
        self.assertEquals(labdiscoverylib.weblab_user.start_date, self._start_date)

        task_id = response.split('@@task@@')[1]

        # There is one task, which is running
        self.assertEquals(len(self.weblab.tasks), 1)
        self.assertEquals(len(self.weblab.running_tasks), 1)

        task1 = self.weblab.get_task(task_id)
        self.assertIsNotNone(task1)
        self.assertEquals(task1.name, 'task')
        self.assertEquals(task1.status, 'submitted')
        self.assertTrue(task1.submitted)
        self.assertFalse(task1.finished)
        self.assertFalse(task1.stopping)
        self.assertFalse(task1.done)
        self.assertFalse(task1.failed)
        self.assertFalse(task1.running)
        self.assertEquals(task1.session_id, session_id1)
        self.assertIsNone(task1.result)
        self.assertIsNone(task1.error)
        task1.stop()
        task1.retrieve()
        self.assertTrue(task1.stopping)
        
        task1b = self.weblab.get_task(self.current_task)
        task1c = self.weblab.get_task('task')
        task1d = self.weblab.get_running_task(self.current_task)
        task1e = self.weblab.get_running_task('task')

        self.assertEquals(task1b, task1)
        self.assertEquals(task1c, task1)
        self.assertEquals(task1d, task1)
        self.assertEquals(task1e, task1)

        tasks_b = self.weblab.get_tasks(self.current_task)
        tasks_c = self.weblab.get_tasks('task')
        tasks_d = self.weblab.get_running_tasks(self.current_task)
        tasks_e = self.weblab.get_running_tasks('task')

        self.assertEquals(1, len(tasks_b))
        self.assertEquals(1, len(tasks_c))
        self.assertEquals(1, len(tasks_d))
        self.assertEquals(1, len(tasks_e))

        self.assertEquals(0, len(self.weblab.get_tasks('foo')))
        self.assertEquals(0, len(self.weblab.get_running_tasks('foo')))

        self.assertEquals(tasks_b[0], task1)
        self.assertEquals(tasks_c[0], task1)
        self.assertEquals(tasks_d[0], task1)
        self.assertEquals(tasks_e[0], task1)

        # We're outside a task
        self.assertFalse(labdiscoverylib.current_task_stopping)

        self.weblab.join_tasks(self.current_task, timeout=0.01, stop=True)

        # But the counter is still zero
        self.assertEquals(self.counter, 0)
    
        global started
        started = False

        def wait_for_thread():
            global started
            started = True
            task1.join(timeout=3)

        background_thread = threading.Thread(target=wait_for_thread)
        background_thread.daemon = True
        background_thread.start()

        initial = time.time()
        while not started:
            time.sleep(0.05)
            if time.time() - initial > 2:
                self.fail("Error waiting for thread to start")

        # Run the tasks
        self.weblab.run_tasks()

        background_thread.join(timeout=5)
        self.assertFalse(background_thread.is_alive())

        # The task has been run
        self.assertEquals(self.counter, 1)

        # There is still 1 task in this session, but no running task
        self.assertEquals(len(self.weblab.tasks), 1)
        self.assertEquals(len(self.weblab.running_tasks), 0)
        
        # Let's retrieve the task again
        task2 = self.weblab.get_task(task_id)
        self.assertEquals(task2.status, 'done')
        self.assertTrue(task2.done)
        self.assertTrue(task2.finished)
        self.assertFalse(task2.failed)
        self.assertFalse(task2.submitted)
        self.assertFalse(task2.running)
        self.assertIsNone(task2.error)
        self.assertEquals(task2.result, [1, 'bar'])
        self.assertIn('inside', task2.data)
        self.assertEquals(task2.data['inside'], 'yes')

        self.assertFalse(labdiscoverylib.current_task)

        # And let's see how it's the same task as before
        self.assertEquals(task1, task2)
        self.assertEquals(hash(task1), hash(task2))
        self.assertEquals(cmp(task1, task2), 0)
        self.assertFalse(task1 < task2)
        self.assertFalse(task2 < task1)

        task2b = self.weblab.get_task(self.current_task)
        task2c = self.weblab.get_task('task')
        task2d = self.weblab.get_running_task(self.current_task)
        task2e = self.weblab.get_running_task('task')
        self.assertEquals(task2b, task2)
        self.assertEquals(task2c, task2)
        self.assertIsNone(task2d)
        self.assertIsNone(task2e)

        # In python 3 it's quite difficult to find the largest hashable value
        # ( sys.hash_info.modulus - 1 is the largest number where hash(n) == n, but 
        # for many other hash(x) > hash(2 ^ 61 - 1))

        self.assertIn(task1.task_id, repr(task1))
        self.assertNotEquals(task1, task1.task_id)
        self.assertNotEquals(cmp(task1, task1.task_id), 0)

        # Cool!

        self.client.get(url_for('weblab_poll_url', session_id=session_id1))
        self.client.get('/poll')
        self.client.get('/logout')

        # Even before cleaning expired users
        rv = self.client.get('/lab/active')
        self.assertEquals(rv.location, 'http://labsland.com')


        self.weblab.clean_expired_users()

        self.status(session_id1)
        self.dispose(session_id1)

        rv = self.client.get('/lab/active')
        self.assertEquals(rv.location, 'http://labsland.com')
        
        self.client.get('/lab/')
        self.assertFalse(labdiscoverylib.weblab_user.active)
        self.assertFalse(labdiscoverylib.weblab_user.is_anonymous)
        self.assertEquals(labdiscoverylib.weblab_user.time_left, 0)
        self.assertEquals(labdiscoverylib.weblab_user.session_id, session_id1)
        self.assertIn(session_id1, str(labdiscoverylib.weblab_user))
        with self.assertRaises(NotImplementedError):
            labdiscoverylib.weblab_user.data = {}

        with self.assertRaises(NotImplementedError):
            labdiscoverylib.weblab_user.update_data()
        self.assertEquals(labdiscoverylib.weblab_user.locale, 'es_ES')
        self.assertEquals(labdiscoverylib.weblab_user.full_name, 'Jim Smith')
        self.assertEquals(labdiscoverylib.weblab_user.experiment_name, 'mylab')
        self.assertEquals(labdiscoverylib.weblab_user.category_name, 'Lab experiments')
        self.assertEquals(labdiscoverylib.weblab_user.experiment_id, 'mylab@Lab experiments')

    def test_status_concrete_time_left_without_timestamp(self):
        # New user, with 3 seconds
        launch_url1, session_id1 = self.new_user(assigned_time=3, use_timestamp=False)
        
        should_finish = self.status()['should_finish']

        # Ask in 2..3 seconds (not 5)
        self.assertGreaterEqual(should_finish, 1)
        self.assertLessEqual(should_finish, 3)

    def test_status_concrete_time_left(self):
        # New user, with 3 seconds
        launch_url1, session_id1 = self.new_user(assigned_time=3)
        
        should_finish = self.status()['should_finish']

        # Ask in 2..3 seconds (not 5)
        self.assertGreaterEqual(should_finish, 1)
        self.assertLessEqual(should_finish, 3)

    def test_status_exited(self):
        # New user, with 3 seconds
        launch_url1, session_id1 = self.new_user(assigned_time=3)

        self.client.get(launch_url1, follow_redirects=True)
        self.client.get(url_for('weblab_logout_url', session_id=session_id1))
        
        should_finish = self.status()['should_finish']

        # Logged out
        self.assertEquals(should_finish, -1)

    def test_status_time_left_passed(self):
        # New user, with 3 seconds
        launch_url1, session_id1 = self.new_user(assigned_time=0.1)

        self.client.get(launch_url1, follow_redirects=True)
        time.sleep(0.2)
        
        should_finish = self.status()['should_finish']

        # time passed
        self.assertEquals(should_finish, -1)

    def test_status_timeout(self):
        # New user, with 3 seconds
        self.weblab.timeout = 0.1
        launch_url1, session_id1 = self.new_user()

        self.client.get(launch_url1, follow_redirects=True)
        time.sleep(0.2)
        
        should_finish = self.status()['should_finish']

        # time passed
        self.assertEquals(should_finish, -1)

class TaskFailTest(BaseSessionWebLabTest):

    def lab(self):
        task = self.current_task.delay()
        return str(task.task_id)

    def task(self):
        10 / 0
        return -1

    def test_task_fail(self):
        # New user 
        launch_url1, session_id1 = self.new_user()

        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))

        task_id = response

        task = self.weblab.get_task(response)
        self.assertEquals(task.status, 'submitted')
        self.assertTrue(task.submitted)
        self.assertFalse(task.failed)
        self.assertFalse(task.finished)
        self.assertFalse(task.stopping)
        self.assertFalse(task.running)
        self.assertFalse(task.done)
        
        with StdWrap():
            self.weblab.run_tasks()

        task.retrieve()
        self.assertEquals(task.status, 'failed')
        self.assertTrue(task.failed)
        self.assertTrue(task.finished)
        self.assertFalse(task.done)
        self.assertFalse(task.running)
        self.assertFalse(task.submitted)
        self.assertIsNone(task.result)
        self.assertIsNotNone(task.error)
        self.assertEqual(task.error['code'], 'exception')
        self.assertIn('zero', task.error['message'])
        self.assertEqual(task.error['class'], 'ZeroDivisionError')

class TaskJoinTimeoutTest(BaseSessionWebLabTest):

    def lab(self):
        task = self.current_task.delay()
        return str(task.task_id)

    def task(self):
        return -1

    def test_task_join_timeout(self):
        # New user 
        launch_url1, session_id1 = self.new_user()

        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))

        task_id = response

        task = self.weblab.get_task(response)
        
        # Task has not started. Perfect for running joins with timeout:
        with self.assertRaises(labdiscoverylib.TimeoutError):
            task.join(timeout=0.1)

        self.assertFalse(task.finished)
        task.join(timeout=0.1, error_on_timeout=False)
        self.assertFalse(task.finished)

class TaskJoinSameThreadTest(BaseSessionWebLabTest):

    def lab(self):
        task = self.current_task.delay()
        return str(task.task_id)

    def task(self):
        labdiscoverylib.current_task.join(timeout=0.5)

    def test_task_join_timeout(self):
        # New user 
        launch_url1, session_id1 = self.new_user()

        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))

        task_id = response

        with StdWrap():
            self.weblab.run_tasks()

        task = self.weblab.get_task(response)
        self.assertTrue(task.failed)
        self.assertEquals('exception', task.error['code'])
        self.assertEquals('RuntimeError', task.error['class'])
        self.assertIn('Deadlock', task.error['message'])
        

class MyLabUser(object):
    def __init__(self, username_unique, username):
        self.username_unique = username_unique
        self.username = username

class LoadUserTest(BaseSessionWebLabTest):
    def test_load_no_loader(self):
        launch_url1, session_id1 = self.new_user()
        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))

        user = labdiscoverylib.weblab_user.user
        self.assertIsNone(user)

    def test_load_simple(self):
        @self.weblab.user_loader
        def user_loader(username_unique):
            return MyLabUser(username_unique, labdiscoverylib.weblab_user.username)

        launch_url1, session_id1 = self.new_user(username='user1', username_unique='unique1')
        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))
        user = labdiscoverylib.weblab_user.user
        self.assertIsInstance(user, MyLabUser)
        self.assertEquals(user.username, 'user1')
        self.assertEquals(user.username_unique, 'unique1')

        user2 = labdiscoverylib.weblab_user.user
        self.assertIsInstance(user2, MyLabUser)
        self.assertEquals(user2.username, 'user1')
        self.assertEquals(user2.username_unique, 'unique1')

        launch_url2, session_id2 = self.new_user(username='user2', username_unique='unique2')
        response = self.get_text(self.client.get(launch_url2, follow_redirects=True))
        user3 = labdiscoverylib.weblab_user.user
        self.assertIsInstance(user3, MyLabUser)
        self.assertEquals(user3.username, 'user2')
        self.assertEquals(user3.username_unique, 'unique2')

    def test_user_loader_fail(self):
        @self.weblab.user_loader
        def user_loader(username_unique):
            raise Exception("Random error")

        launch_url1, session_id1 = self.new_user(username='user1', username_unique='unique1')
        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))
        with self.assertRaises(Exception) as cm:
            labdiscoverylib.weblab_user.user
        
        self.assertIn("Random error", str(cm.exception))


class LongTaskTest(BaseSessionWebLabTest):

    def get_config(self):
        config = super(LongTaskTest, self).get_config()
        config['WEBLAB_AUTOCLEAN_THREAD'] = True
        config['WEBLAB_TASK_THREADS_PROCESS'] = 3
        return config

    def lab(self):
        if self.run_delayed:
            task_object = self.current_task.delay()
        else:
            task_object = self.current_task.run_sync(timeout=self.sync_timeout)
        return str(task_object.task_id)

    def task(self):
        time.sleep(self.sleep_time)
        return 0

    def test_long_task(self):
        self.sleep_time = 0.6
        self.weblab.cleaner_thread_interval = 0.1
        self.run_delayed = True
        launch_url1, session_id1 = self.new_user()
        response = self.get_text(self.client.get(launch_url1, follow_redirects=True))
        task_id = response

        max_time = 3
        t0 = time.time()
        while True:
            task = self.weblab.get_task(task_id)
            # Temporal variable since sometimes the change happens
            # between task.status and the next task.status
            task_status = task.status
            if task_status == 'running':
                self.assertTrue(task.running)
                self.assertFalse(task.finished)
                self.assertFalse(task.done)
                self.assertFalse(task.failed)
                self.assertFalse(task.submitted)
                break
            self.assertEquals(task_status, 'submitted')
            time.sleep(0.03)

            if time.time() - t0 > max_time:
                self.fail("Too long checking for a submitted thread")

        self.client.get('/logout')
        time.sleep(0.2) # So other thread calls clean
        self.dispose()
        labdiscoverylib._on_exit()

    def test_run_sync_short(self):
        self.sleep_time = 0.1
        self.weblab.cleaner_thread_interval = 0.1
        self.run_delayed = False
        self.sync_timeout = None # Forever
        launch_url1, session_id1 = self.new_user()
        task_id = self.get_text(self.client.get(launch_url1, follow_redirects=True))
        task = self.weblab.get_task(task_id)
        self.assertTrue(task.done)
        self.weblab._cleanup()

    def test_run_sync_long(self):
        self.sleep_time = 0.4
        self.weblab.cleaner_thread_interval = 0.1
        self.run_delayed = False
        self.sync_timeout = 0.1 # Forever
        launch_url1, session_id1 = self.new_user()
        task_id = self.get_text(self.client.get(launch_url1, follow_redirects=True))
        task = self.weblab.get_task(task_id)
        self.assertFalse(task.done)
        task.join()
        self.assertTrue(task.done)
        self.weblab._cleanup()

class GlobalUniqueTaskTest(BaseSessionWebLabTest):

    def get_config(self):
        config = super(GlobalUniqueTaskTest, self).get_config()
        config['WEBLAB_AUTOCLEAN_THREAD'] = True
        config['WEBLAB_TASK_THREADS_PROCESS'] = 3
        return config

    def lab(self):
        # We can't know which one will be running first, so we force one to be running first
        task_object = self.unique_task.delay()
        while not self.task_is_running:
            time.sleep(0.05)
        
        # Now we now that the task is running. Let's run it again
        task2 = self.unique_task.delay()
        task2.join() # Should be fast
        if not task2.failed:
            raise Exception("unique=global should have avoided the second task")
        if task2.error['class'] != 'AlreadyRunningError':
            raise Exception("unique=global expected an AlreadyRunningError; got {}".format(task2.error))

        return str(task_object.task_id)

    def task(self):
        initial = time.time()
        self.task_is_running = True

        while not labdiscoverylib.current_task_stopping:
            time.sleep(0.05)

            elapsed = time.time() - initial
            if elapsed > 6:
                raise Exception("Error, nobody calling stop")

    def test_global_unique(self):
        original_steps_waiting = _TaskRunner._STEPS_WAITING
        _TaskRunner._STEPS_WAITING = 1
        try:
            self.weblab.cleaner_thread_interval = 0.1
            self.task_is_running = False
    
            @self.weblab.task(unique='global')
            def task_global_unique():
                return self.task()
    
            self.unique_task = task_global_unique
    
            launch_url1, session_id1 = self.new_user()
    
            with StdWrap():
                task_id = self.get_text(self.client.get(launch_url1, follow_redirects=True))
    
            task = self.weblab.get_task(task_id)
            task.stop()
            task.join()
            self.assertTrue(task.done)
        finally:
            _TaskRunner._STEPS_WAITING = original_steps_waiting

class UserUniqueTaskTest(BaseSessionWebLabTest):

    def get_config(self):
        config = super(UserUniqueTaskTest, self).get_config()
        config['WEBLAB_AUTOCLEAN_THREAD'] = True
        config['WEBLAB_TASK_THREADS_PROCESS'] = 3
        return config

    def lab(self):
        # We can't know which one will be running first, so we force one to be running first
        task_object = self.unique_task.delay()
        while not self.task_is_running:
            time.sleep(0.05)

        task2 = self.unique_task.delay()
        task2.join() # Should be fast
        if not task2.failed:
            raise Exception("unique=user should have avoided the second task")
        if task2.error['class'] != 'AlreadyRunningError':
            raise Exception("unique=user expected an AlreadyRunningError; got {}".format(task2.error))
        return str(task_object.task_id)

    def task(self):
        initial = time.time()
        self.task_is_running = True

        while not labdiscoverylib.current_task_stopping:
            time.sleep(0.05)

            elapsed = time.time() - initial
            if elapsed > 6:
                raise Exception("Error, nobody calling stop")

    def test_user_unique(self):
        self.weblab.cleaner_thread_interval = 0.1
        self.task_is_running = False

        @self.weblab.task(unique='user')
        def task_user_unique():
            return self.task()

        self.unique_task = task_user_unique

        launch_url1, session_id1 = self.new_user()

        with StdWrap():
            task_id = self.get_text(self.client.get(launch_url1, follow_redirects=True))

        task = self.weblab.get_task(task_id)
        task.stop()
        task.join()
        self.assertTrue(task.done)


class LongDisposeErrorTest(BaseSessionWebLabTest):

    def setUp(self):
        super(LongDisposeErrorTest, self).setUp()
        self.blocker_lock = threading.Lock()
        self.first_notifier_lock = threading.Lock()
        self.second_notifier_lock = threading.Lock()

    def lab(self):
        return ":-)"

    def on_dispose(self):
        self.first_notifier_lock.release()
        self.blocker_lock.acquire()
        self.blocker_lock.release()
        self.second_notifier_lock.release()

    def test_long_dispose(self):
        # New user
        launch_url1, session_id1 = self.new_user()

        # So it will be blocked on the dispose
        self.blocker_lock.acquire()

        self.first_notifier_lock.acquire()
        self.second_notifier_lock.acquire()

        background_dispose = threading.Thread(target=lambda : self.dispose())
        background_dispose.daemon = True
        background_dispose.start()

        # We acquire twice so we wait for the other thread to be running
        self.first_notifier_lock.acquire()

        # Now we know that dispose is blocked in the blocker_lock:

        # We run status, as weblab would do to know if the student finished.
        # It should not return -1, because it is still in dispose. It should
        # wait for dispose to finish
        status = self.status()
        self.assertGreater(status['should_finish'], 0)

        self.first_notifier_lock.release()
        self.blocker_lock.release()

        self.second_notifier_lock.acquire()
        time.sleep(0.1) # If we are here we know that the thread has finished. Wait a bit

        status = self.status()
        self.assertEquals(status['should_finish'], -1)



class DisposeErrorTest(BaseSessionWebLabTest):

    def lab(self):
        return ":-)"

    def on_dispose(self):
        raise Exception("Testing error in the dispose method")

    def test_task_fail(self):
        # New user
        launch_url1, session_id1 = self.new_user()

        with StdWrap():
            self.dispose()

class StartErrorTest(BaseSessionWebLabTest):

    def lab(self):
        return ":-)"

    def on_start(self, client_data, server_data):
        raise Exception("Testing error in the start method")

    def test_task_fail(self):
        import labdiscoverylib.views as labdiscoverylib_views
        old_dispose_user = labdiscoverylib_views.dispose_user
        def new_dispose_user(*args, **kwargs):
            raise Exception("weird error")

        labdiscoverylib_views.dispose_user = new_dispose_user
        try:
            with self.assertRaises(TestNewUserError):
                with StdWrap():
                    self.new_user()
        finally:
            labdiscoverylib_views.dispose_user = old_dispose_user

class BaseCLITest(BaseSessionWebLabTest):

    def setUp(self):
        super(BaseCLITest, self).setUp()

        json_library = json
        class FakeResponse(object):
            def __init__(self, json_contents):
                self.json_contents = json_contents

            def json(self):
                return self.json_contents

        class FakeRequests(object):
            @staticmethod
            def post(url, json, auth):
                json_contents = json_library.dumps(json)
                rv = self.client.post('/' + url.split('/', 3)[-1], data=json_contents, headers=self.auth_headers)
                json_contents = self.get_json(rv)
                return FakeResponse(json_contents)
            
        labdiscoverylib.requests = FakeRequests

    def tearDown(self):
        super(BaseCLITest, self).tearDown()
        labdiscoverylib.requests = requests


class CLITest(BaseCLITest):

    @unittest.skip("Pending")
    def test_cli_flow(self):
        runner = CliRunner()
        
        class webbrowser(object):
            @staticmethod
            def open(*args, **kwargs):
                pass

        labdiscoverylib.webbrowser = webbrowser

        with runner.isolated_filesystem():
            result = runner.invoke(self.app.cli, ["weblab", "fake", "new"])
            self.assertEquals(result.exit_code, 0)

            result = runner.invoke(self.app.cli, ["weblab", "fake", "status"])
            self.assertIn("Should finish: 5", result.output)
            self.assertEquals(result.exit_code, 0)

            result = runner.invoke(self.app.cli, ["weblab", "fake", "dispose"])
            self.assertIn("Deleted", result.output)
            self.assertEquals(result.exit_code, 0)

            result = runner.invoke(self.app.cli, ["weblab", "fake", "dispose"])
            self.assertIn("Session not found", result.output)
            self.assertEquals(result.exit_code, 0)

            result = runner.invoke(self.app.cli, ["weblab", "fake", "status"])
            self.assertIn("Session not found", result.output)
            self.assertEquals(result.exit_code, 0)

            result = runner.invoke(self.app.cli, ["weblab", "fake", "new", "--dont-open-browser"])
            self.assertEquals(result.exit_code, 0)

            request_data = {
                'action': 'delete',
            }
            session_id_line = [ line for line in result.output.splitlines() if self.server_name in line ][0]
            session_id = session_id_line.strip().split('/')[-1]
            self.weblab._backend._tests_delete_user(session_id)

            result = runner.invoke(self.app.cli, ["weblab", "fake", "dispose"])
            self.assertIn("Not found", result.output)
            self.assertEquals(result.exit_code, 0)

    @unittest.skip("Pending")
    def test_other_cli(self):
        runner = CliRunner()

        result = runner.invoke(self.app.cli, ["weblab", "clean-expired-users"])
        self.assertEquals(result.exit_code, 0)

        result = runner.invoke(self.app.cli, ["weblab", "run-tasks"])
        self.assertEquals(result.exit_code, 0)

    @unittest.skip("Pending")
    def test_loop_cli(self):
        runner = CliRunner()

        labdiscoverylib._TESTING_LOOP = True
        result = runner.invoke(self.app.cli, ["weblab", "loop"])
        self.assertEquals(result.exit_code, 0)

class CLIFailTest(BaseCLITest):
    def on_start(self, client_data, server_data):
        raise Exception("Error initializing laboratory")

    @unittest.skip("Pending")
    def test_cli_error(self):

        runner = CliRunner()

        with runner.isolated_filesystem():
            result = runner.invoke(self.app.cli, ["weblab", "fake", "new", "--dont-open-browser"])
            self.assertIn("Error processing", result.output)

class WebLabSocketIOTest(BaseSessionWebLabTest):

    def create_socketio(self):
        self.socketio = SocketIO(self.app)

        @self.socketio.on('connect')
        @labdiscoverylib.socket_requires_login
        def on_connect():
            self.socketio.emit('results', {'result': 'onconnected'}, namespace='/test')

        @self.socketio.on('my-active-test', namespace='/test')
        @labdiscoverylib.socket_requires_active
        def active_message(message):
            self.socketio.emit('results', {'result': 'onactive'}, namespace='/test')

        @self.socketio.on('my-login-test', namespace='/test')
        @labdiscoverylib.socket_requires_login
        def login_message(message):
            self.socketio.emit('results', {'result': 'onlogin'}, namespace='/test')

    def setUp(self):
        super(WebLabSocketIOTest, self).setUp()
        self.create_socketio()

    @unittest.skip("Pending")
    def test_socket_requires_active(self):
        socket_client = self.socketio.test_client(app=self.app, namespace='/test')
        socket_client.emit('my-login-test', "hi login", namespace='/test')
        socket_client.emit('my-active-test', "hi active", namespace='/test')
        results = socket_client.get_received(namespace='/test')
        self.assertEquals(len(results), 0)

        launch_url1, session_id1 = self.new_user()
        response = self.client.get(launch_url1, follow_redirects=False)
        cookie = response.headers['Set-Cookie']
        headers = {
            'Cookie': cookie.split(';')[0]
        }

        socket_client = self.socketio.test_client(app=self.app, namespace='/test', headers=headers)
        socket_client.connect(namespace='/test', headers = headers)
        socket_client.emit('my-login-test', "hi login", namespace='/test')
        socket_client.emit('my-active-test', "hi active", namespace='/test')
        results = socket_client.get_received(namespace='/test')
        self.assertEquals(len(results), 3)
        self.assertEquals(results[0]['args'][0]['result'], 'onconnected')
        self.assertEquals(results[1]['args'][0]['result'], 'onlogin')
        self.assertEquals(results[2]['args'][0]['result'], 'onactive')
        socket_client.disconnect()

        self.client.get('/logout')

        socket_client = self.socketio.test_client(app=self.app, namespace='/test', headers=headers)
        socket_client.emit('my-login-test', "hi login", namespace='/test')
        socket_client.emit('my-active-test', "hi active", namespace='/test')
        results = socket_client.get_received(namespace='/test')
        self.assertEquals(len(results), 1)
        self.assertEquals(results[0]['args'][0]['result'], 'onlogin')


class WebLabConfigErrorsTest(unittest.TestCase):

    def _check_error(self, config, error_class, message):
        with self.assertRaises(error_class) as cm:
            self.weblab = labdiscoverylib.WebLab()
            self.app = Flask(__name__)
            self.app.config.update(config)
            self.weblab.init_app(self.app)

        self.assertIn(message, str(cm.exception))

    def test_callback(self):
        self._check_error({
                'WEBLAB_CALLBACK_URL': '',
                'WEBLAB_USERNAME': 'lde',
                'WEBLAB_PASSWORD': 'password',
                'SERVER_NAME': 'localhost:5000',
            }, ValueError, "Empty URL")

    def test_username(self):
        self._check_error({
                'WEBLAB_PASSWORD': 'password',
                'SERVER_NAME': 'localhost:5000',
            }, ValueError, "Missing WEBLAB_USERNAME")

    def test_password(self):
        self._check_error({
                'WEBLAB_USERNAME': 'lde',
                'SERVER_NAME': 'localhost:5000',
            }, ValueError, "Missing WEBLAB_PASSWORD")

class WebLabSetupErrorsTest(unittest.TestCase):
    def test_empty_app(self):
        with self.assertRaises(ValueError) as cm:
            labdiscoverylib.WebLab().init_app(None)

        self.assertIn("Flask app", str(cm.exception))

    def test_app_trailing_slashes(self):
        app = Flask(__name__)
        app.config.update({
            'WEBLAB_BASE_URL': '/mylab/',
            'WEBLAB_CALLBACK_URL': '/mylab/callback/',
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': 'localhost:5000',
        })
        with StdWrap():
            weblab = labdiscoverylib.WebLab(app)
        weblab._cleanup()

    def test_missing_server_name(self):
        app = Flask(__name__)
        app.config.update({
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
        })
        with StdWrap():
            sysargv = sys.argv
            sys.argv = list(sys.argv) + [ 'weblab', 'fake', 'new', '--dont-open-browser']
            try:
                weblab = labdiscoverylib.WebLab(app)
            finally:
                sys.argv = sysargv
        weblab._cleanup()

    def test_app_twice(self):
        app = Flask(__name__)
        app.config.update({
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': 'localhost:5000',
        })
        weblab = labdiscoverylib.WebLab(app)
        weblab.init_app(app) # No problem
        weblab._cleanup()

    def test_app_twice_different_apps(self):
        app1 = Flask(__name__)
        app1.config.update({
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': 'localhost:5000',
        })
        app2 = Flask(__name__)
        weblab = labdiscoverylib.WebLab(app1)
        with self.assertRaises(ValueError) as cm:
            weblab.init_app(app2)

        self.assertIn('different app', str(cm.exception))
        weblab._cleanup()

    def test_app_no_thread_and_auto_clean(self):
        app = Flask(__name__)
        app.config.update({
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': 'localhost:5000',
            'WEBLAB_NO_THREAD': True,
            'WEBLAB_AUTOCLEAN_THREAD': True,
        })
        weblab = labdiscoverylib.WebLab()
        with self.assertRaises(ValueError) as cm:
            weblab.init_app(app)

        self.assertIn('incompatible with WEBLAB_AUTOCLEAN_THREAD', str(cm.exception))
        weblab._cleanup()

    def test_app_no_thread_and_task_threads(self):
        app = Flask(__name__)
        app.config.update({
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': 'localhost:5000',
            'WEBLAB_NO_THREAD': True,
            'WEBLAB_TASK_THREADS_PROCESS': 5,
        })
        weblab = labdiscoverylib.WebLab()
        with self.assertRaises(ValueError) as cm:
            weblab.init_app(app)

        self.assertIn('incompatible with WEBLAB_TASK_THREADS_PROCESS', str(cm.exception))
        weblab._cleanup()

    def test_app_two_weblabs_same_app(self):
        app = Flask(__name__)
        app.config.update({
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'WEBLAB_BASE_URL': '/foo',
            'SERVER_NAME': 'localhost:5000',
        })
        weblab1 = labdiscoverylib.WebLab(app)
        with self.assertRaises(ValueError) as cm:
            weblab2 = labdiscoverylib.WebLab(app)

        self.assertIn('already installed', str(cm.exception))

        weblab1._cleanup()

    def test_app_twice_different_config(self):
        app = Flask(__name__)
        app.config.update({
            'WEBLAB_CALLBACK_URL': '/mylab/callback',
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': 'localhost:5000',
        })
        weblab = labdiscoverylib.WebLab(app)
        app.config.update({
            'WEBLAB_CALLBACK_URL': '/mylab/callback2',
        })

        with self.assertRaises(ValueError) as cm:
            weblab.init_app(app)

        self.assertIn('different config', str(cm.exception))
        labdiscoverylib._cleanup_all()
        weblab._cleanup()

    def _create_weblab(self):
        self.app = Flask(__name__)
        self.app.config.update({
            'WEBLAB_USERNAME': 'lde',
            'WEBLAB_PASSWORD': 'password',
            'SERVER_NAME': 'localhost:5000',
        })
        self.weblab = labdiscoverylib.WebLab(self.app)
        self.weblab.init_app(self.app) # No problem

    def test_initial_url_duplicated(self):
        self._create_weblab()

        try:
            @self.weblab.initial_url
            def foo():
                pass
            
            with self.assertRaises(ValueError):
                @self.weblab.initial_url
                def bar():
                    pass
        finally:
            self.weblab._cleanup()

    def test_user_loader_duplicated(self):
        self._create_weblab()

        try:
            @self.weblab.user_loader
            def user_loader(username_unique):
                pass
            
            with self.assertRaises(ValueError):
                @self.weblab.user_loader
                def user_loader2(username_unique):
                    pass
        finally:
            self.weblab._cleanup()       

    def test_on_start_duplicated(self):
        self._create_weblab()

        try:
            @self.weblab.on_start
            def foo():
                pass
            
            with self.assertRaises(ValueError):
                @self.weblab.on_start
                def bar():
                    pass
        finally:
            self.weblab._cleanup()

    def test_on_dispose_duplicated(self):
        self._create_weblab()

        try:
            @self.weblab.on_dispose
            def foo():
                pass
            
            with self.assertRaises(ValueError):
                @self.weblab.on_dispose
                def bar():
                    pass
        finally:
            self.weblab._cleanup()

    def test_socket_requires_login_without_socketio(self):
        past_value = labdiscoverylib._FLASK_SOCKETIO
        labdiscoverylib._FLASK_SOCKETIO = False
        try:
            with StdWrap():
                @labdiscoverylib.socket_requires_login
                def f():
                    pass
        finally:
            labdiscoverylib._FLASK_SOCKETIO = past_value

    def test_socket_requires_active_without_socketio(self):
        past_value = labdiscoverylib._FLASK_SOCKETIO
        labdiscoverylib._FLASK_SOCKETIO = False
        try:
            with StdWrap():
                @labdiscoverylib.socket_requires_active
                def f():
                    pass
        finally:
            labdiscoverylib._FLASK_SOCKETIO = past_value

def _to_timestamp(dtime):
    return str(int(time.mktime(dtime.timetuple()))) + str(dtime.microsecond / 1e6)[1:]