test_psiturk.py

# -*- coding: utf-8 -*-
""" This module tests the psiTurk suite.  """

import os
import unittest
import tempfile
import psiturk
import json
from faker import Faker


fake = Faker()  # Fake data generator


class FlaskTestClientProxy(object):
    '''Spoof user agent (Chrome)'''
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        environ['REMOTE_ADDR'] = environ.get('REMOTE_ADDR', fake.ipv4())
        environ['HTTP_USER_AGENT'] = 'Mozilla/5.0 (Macintosh; Intel Mac OS X\
            10_9_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1944.0\
            Safari/537.36'
        return self.app(environ, start_response)


class BadFlaskTestClientProxy(object):
    '''Spoof user agent (iPad)'''
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        environ['REMOTE_ADDR'] = environ.get('REMOTE_ADDR', fake.ipv4())
        environ['HTTP_USER_AGENT'] = 'Mozilla/5.0 (iPad; U; CPU OS 3_2 like \
            Mac OS X; en-us) AppleWebKit/531.21.10 (KHTML, like Gecko) \
            Version/4.0.4 Mobile/7B334b Safari/531.21.10'
        return self.app(environ, start_response)


class PsiturkUnitTest(unittest.TestCase):

    def setUp(self, case=None):
        '''Build up fixtures'''
        os.chdir('psiturk-example')
        import psiturk.db
        import psiturk.experiment
        reload(psiturk.experiment)

        fd, db_path = tempfile.mkstemp()
        self.db_fd = fd
        psiturk.experiment.app.config['DATABASE'] = db_path
        psiturk.experiment.app.wsgi_app = FlaskTestClientProxy(
            psiturk.experiment.app.wsgi_app)
        self.app = psiturk.experiment.app.test_client()
        self.config = psiturk.experiment.CONFIG
        psiturk.db.init_db()

        # Fake MTurk data
        self.worker_id = fake.md5(raw_output=False)
        self.hit_id = fake.md5(raw_output=False)
        self.assignment_id = fake.md5(raw_output=False)

    def tearDown(self):
        '''Tear down fixtures'''
        os.close(self.db_fd)
        os.chdir('..')
        os.unlink(psiturk.experiment.app.config['DATABASE'])
        self.app = None

    def set_config(self, section, field, value):
        self.config.parent.set(self.config, section, field, str(value))


class PsiTurkStandardTests(PsiturkUnitTest):

    # Test experiment.py
    # ==================

    def test_default_page(self):
        '''Test that root page works.'''
        rv = self.app.get('/')
        assert 'Welcome to psiTurk!' in rv.data

    def test_exp_debug_no_url_vars(self):
        '''Test that exp page throws Error #1003 with no url vars.'''
        rv = self.app.get('/exp')
        assert '<b>Error</b>: 1003' in rv.data

    def test_ad_no_url_vars(self):
        '''Test that ad page throws Error #1001 with no url vars.'''
        rv = self.app.get('/ad')
        assert '<b>Error</b>: 1001' in rv.data

    def test_ad_with_all_urls(self):
        '''Test that ad page throws Error #1003 with no url vars.'''
        args = '&'.join([
            'assignmentId=debug%s' % self.assignment_id,
            'workerId=debug%s' % self.worker_id,
            'hitId=debug%s' % self.hit_id,
            'mode=sandbox'])
        rv = self.app.get('/ad?%s' % args)
        assert 'Thank you for accepting this HIT!' in rv.data

    def test_exp_with_all_url_vars_not_registered_on_ad_server(self):
        '''Test that exp page throws Error #1018 with all url vars but not registered.'''
        args = '&'.join([
            'assignmentId=debug%s' % self.assignment_id,
            'workerId=debug%s' % self.worker_id,
            'hitId=debug%s' % self.hit_id,
            'mode=sandbox'])
        rv = self.app.get('/exp?%s' % args)
        assert '<b>Error</b>: 1018' in rv.data

    def test_sync_put(self):
        request = "&".join([
           "assignmentId=debug%s" % self.assignment_id,
           "workerId=debug%s" % self.worker_id,
           "hitId=debug%s" % self.hit_id,
           "mode=debug"])

        # put the user in the database
        rv = self.app.get("/exp?%s" % request)

        # try putting the sync
        uniqueid = "debug%s:debug%s" % (self.worker_id, self.assignment_id)
        rv = self.app.put('/sync/%s' % uniqueid)
        status = json.loads(rv.data).get("status", "")
        assert status == "user data saved"

    def test_sync_get(self):
        request = "&".join([
           "assignmentId=debug%s" % self.assignment_id,
           "workerId=debug%s" % self.worker_id,
           "hitId=debug%s" % self.hit_id,
           "mode=debug"])

        # put the user in the database
        rv = self.app.get("/exp?%s" % request)

        # save data with sync PUT
        uniqueid = "debug%s:debug%s" % (self.worker_id, self.assignment_id)
        rv = self.app.put('/sync/%s' % uniqueid)

        # get data with sync GET
        uniqueid = "debug%s:debug%s" % (self.worker_id, self.assignment_id)
        rv = self.app.get('/sync/%s' % uniqueid)

        response = json.loads(rv.data)
        assert response.get("assignmentId", "") == "debug%s" % self.assignment_id
        assert response.get("workerId", "") == "debug%s" % self.worker_id
        assert response.get("hitId", "") == "debug%s" % self.hit_id
        assert response.get("condition", None) == 0
        assert response.get("counterbalance", None) == 0
        assert response.get("bonus", None) == 0.0

    def test_favicon(self):
        '''Test that favicon loads.'''
        rv = self.app.get('/favicon.ico')
        assert rv.status_code == 200

    def test_complete_experiment(self):
        '''Test that a participant can start and finish the experiment.'''
        request = "&".join([
            "assignmentId=debug%s" % self.assignment_id,
            "workerId=debug%s" % self.worker_id,
            "hitId=debug%s" % self.hit_id,
            "mode=debug"])

        # put the user in the database
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200

        # save data with sync PUT
        uniqueid = "debug%s:debug%s" % (self.worker_id, self.assignment_id)
        rv = self.app.put('/sync/%s' % uniqueid)
        assert rv.status_code == 200

        # complete experiment
        mode = 'debug'
        rv = self.app.get('/complete?uniqueId=%s&mode=%s' % (uniqueid, mode))
        assert rv.status_code == 200

    def test_repeat_experiment_fail(self):
        '''Test that a participant cannot repeat the experiment.'''
        request = "&".join([
            "assignmentId=%s" % self.assignment_id,
            "workerId=%s" % self.worker_id,
            "hitId=%s" % self.hit_id,
            "mode=debug"])

        # put the user in the database
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200

        # save data with sync PUT
        uniqueid = "%s:%s" % (self.worker_id, self.assignment_id)
        rv = self.app.put('/sync/%s' % uniqueid)
        assert rv.status_code == 200

        # complete experiment
        mode = 'debug'
        rv = self.app.get('/complete?uniqueId=%s&mode=%s' % (uniqueid, mode))
        assert rv.status_code == 200

        # choose new assignment and hit ids
        self.assignment_id = fake.md5(raw_output=False)
        self.hit_id = fake.md5(raw_output=False)
        request = "&".join([
            "assignmentId=%s" % self.assignment_id,
            "workerId=%s" % self.worker_id,
            "hitId=%s" % self.hit_id,
            "mode=debug"])

        # make sure they are blocked on the ad page
        rv = self.app.get('/ad?%s' % request)
        assert ': 1010' in rv.data

        # make sure they are blocked on the experiment page
        rv = self.app.get("/exp?%s" % request)
        assert ': 1010' in rv.data

    def test_repeat_experiment_success(self):
        '''Test that a participant can repeat the experiment.'''
        self.set_config('HIT Configuration', 'allow_repeats', 'true')
        request = "&".join([
            "assignmentId=%s" % self.assignment_id,
            "workerId=%s" % self.worker_id,
            "hitId=%s" % self.hit_id,
            "mode=debug"])

        # put the user in the database
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200

        # save data with sync PUT
        uniqueid = "%s:%s" % (self.worker_id, self.assignment_id)
        rv = self.app.put('/sync/%s' % uniqueid)
        assert rv.status_code == 200

        # complete experiment
        mode = 'debug'
        rv = self.app.get('/complete?uniqueId=%s&mode=%s' % (uniqueid, mode))
        assert rv.status_code == 200

        # choose new assignment and hit ids
        self.assignment_id = fake.md5(raw_output=False)
        self.hit_id = fake.md5(raw_output=False)
        request = "&".join([
            "assignmentId=%s" % self.assignment_id,
            "workerId=%s" % self.worker_id,
            "hitId=%s" % self.hit_id,
            "mode=debug"])

        # make sure they are not blocked on the ad page
        rv = self.app.get('/ad?%s' % request)
        assert rv.status_code == 200
        assert ': 1010' not in rv.data

        # make sure they are not blocked on the experiment page
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200
        assert ': 1010' not in rv.data

        # save data with sync PUT
        uniqueid = "%s:%s" % (self.worker_id, self.assignment_id)
        rv = self.app.put('/sync/%s' % uniqueid)
        assert rv.status_code == 200

        # complete experiment
        mode = 'debug'
        rv = self.app.get('/complete?uniqueId=%s&mode=%s' % (uniqueid, mode))
        assert rv.status_code == 200

    def test_repeat_experiment_quit(self):
        '''Test that a participant cannot restart the experiment.'''
        request = "&".join([
            "assignmentId=%s" % self.assignment_id,
            "workerId=%s" % self.worker_id,
            "hitId=%s" % self.hit_id,
            "mode=debug"])

        # put the user in the database
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200

        # put the in the experiment
        uniqueid = "%s:%s" % (self.worker_id, self.assignment_id)
        rv = self.app.post("/inexp", data=dict(uniqueId=uniqueid))
        assert rv.status_code == 200

        # make sure they are blocked on the ad page
        rv = self.app.get('/ad?%s' % request)
        assert rv.status_code == 200
        assert ': 1009' in rv.data

        # make sure they are blocked on the experiment page
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200
        assert ': 1008' in rv.data

        # have them quit the experiment
        rv = self.app.post("/quitter", data=dict(uniqueId=uniqueid))
        assert rv.status_code == 200

        # make sure they are blocked on the ad page
        rv = self.app.get('/ad?%s' % request)
        assert rv.status_code == 200
        assert ': 1009' in rv.data

        # make sure they are blocked on the experiment page
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200
        assert ': 1008' in rv.data

    def test_repeat_experiment_quit_allow_repeats(self):
        '''Test that a participant cannot restart the experiment, even when repeats are allowed.'''
        self.set_config('HIT Configuration', 'allow_repeats', 'true')
        request = "&".join([
            "assignmentId=%s" % self.assignment_id,
            "workerId=%s" % self.worker_id,
            "hitId=%s" % self.hit_id,
            "mode=debug"])

        # put the user in the database
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200

        # put the in the experiment
        uniqueid = "%s:%s" % (self.worker_id, self.assignment_id)
        rv = self.app.post("/inexp", data=dict(uniqueId=uniqueid))
        assert rv.status_code == 200

        # make sure they are blocked on the ad page
        rv = self.app.get('/ad?%s' % request)
        assert rv.status_code == 200
        assert ': 1009' in rv.data

        # make sure they are blocked on the experiment page
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200
        assert ': 1008' in rv.data

        # have them quit the experiment
        rv = self.app.post("/quitter", data=dict(uniqueId=uniqueid))
        assert rv.status_code == 200

        # make sure they are blocked on the ad page
        rv = self.app.get('/ad?%s' % request)
        assert rv.status_code == 200
        assert ': 1009' in rv.data

        # make sure they are blocked on the experiment page
        rv = self.app.get("/exp?%s" % request)
        assert rv.status_code == 200
        assert ': 1008' in rv.data


class BadUserAgent(PsiturkUnitTest):
    '''Setup test blocked user agent (iPad/tablets)'''

    def setUp(self):
        '''Build up fixtures'''
        os.chdir('psiturk-example')
        import psiturk.db
        import psiturk.experiment
        reload(psiturk.experiment)

        fd, db_path = tempfile.mkstemp()
        self.db_fd = fd
        psiturk.experiment.app.config['DATABASE'] = db_path
        psiturk.experiment.app.wsgi_app = BadFlaskTestClientProxy(
            psiturk.experiment.app.wsgi_app)
        self.app = psiturk.experiment.app.test_client()
        psiturk.db.init_db()

        # Fake MTurk data
        self.worker_id = fake.md5(raw_output=False)
        self.hit_id = fake.md5(raw_output=False)
        self.assignment_id = fake.md5(raw_output=False)

    def test_ad_with_bad_user_agent(self):
        '''Test that ad page throws Error when user agent is blocked.'''
        rv = self.app.get(
            '/ad' + '?assignmentId=debug' + self.assignment_id + '&workerId=debug' +
            self.worker_id + '&hitId=debug' + self.hit_id + '&mode=sandbox'
        )
        assert '<b>Error</b>: 1014' in rv.data


class PsiTurkTestPsiturkJS(PsiturkUnitTest):
    ''' Setup test for missing psiturk.js file. '''

    def setUp(self):
        '''Build up fixtures'''
        self.PSITURK_JS_FILE = '../psiturk/psiturk_js/psiturk.js'
        os.chdir('psiturk-example')
        os.rename(self.PSITURK_JS_FILE, self.PSITURK_JS_FILE + '.bup')
        import psiturk.db
        import psiturk.experiment
        reload(psiturk.experiment)

        fd, db_path = tempfile.mkstemp()
        self.db_fd = fd
        psiturk.experiment.app.config['DATABASE'] = db_path
        psiturk.experiment.app.wsgi_app = FlaskTestClientProxy(
            psiturk.experiment.app.wsgi_app)
        self.app = psiturk.experiment.app.test_client()
        psiturk.db.init_db()

    def test_psiturk_js_is_missing(self):
        ''' Test for missing psiturk.js '''
        rv = self.app.get('static/js/psiturk.js')
        assert 'file not found' in rv.data

    def tearDown(self):
        '''Tear down fixtures'''
        super(PsiTurkTestPsiturkJS, self).tearDown()
        os.chdir('psiturk-example')
        os.rename(self.PSITURK_JS_FILE + '.bup', self.PSITURK_JS_FILE)
        os.chdir('..')


class ExperimentErrorsTest(PsiturkUnitTest):

    def test_experiment_errors(self):
        """Make sure every error has a description"""
        error_cls = psiturk.experiment_errors.ExperimentError

        for error_name in error_cls.experiment_errors:
            assert error_name in error_cls.error_descriptions

        for error_name in error_cls.error_descriptions:
            assert error_name in error_cls.experiment_errors


if __name__ == '__main__':
    unittest.main()