1. Martin Czygan
  2. audrid

Source

audrid / tests / test_api_pools.py

# coding: utf-8

from audrid.models import (
    Pool,
    User,
)

from audrid import (app, db, utils)
import base64
import json
import time
import unittest

class PoolsAPITests(unittest.TestCase):
    """ Test DB.
    """
    def setUp(self):
        app.config['TESTING'] = True
        app.config['DEBUG'] = True
        app.config['CSRF_ENABLED'] = False
        # app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
        app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://audrid:audrid@localhost/audrid_test'
        self.app = app.test_client()
        db.drop_all(app=app)
        db.create_all(app=app)

        admin = User(username='admin', email='martin.czygan@gmail.com', password='admin')
        guest = User(username='guest', email='guest@example.com', password='guest')
        db.session.add(admin)
        db.session.add(guest)
        db.session.commit()

    def tearDown(self):
        db.session.remove()
        db.drop_all(app=app)

    def open_with_auth(self, url, method='GET', username='admin', password='admin', data=None, follow_redirects=False):
        headers = {
            'Authorization' : 'Basic %s' % (
            base64.b64encode("%s:%s" % (username, password)))
        }
        return self.app.open(url, method=method, headers=headers, data=data, follow_redirects=follow_redirects)

    def test_pools_post_reject_invalid_data(self):
        """ POST   /pools should reject invalid request data
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({}))
        self.assertEqual(400, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : 'x'}))
        self.assertEqual(400, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : '123456'}))
        self.assertEqual(400, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'tasks' : '123456'}))
        self.assertEqual(400, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'tasks' : []}))
        self.assertEqual(400, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'tasks' : ["Hello", "World"]}))
        self.assertEqual(400, rv.status_code)

    def test_pools_post_reject_duplicate_id(self):
        """ POST   /pools should reject duplicate pool id
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        self.assertEqual(302, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        self.assertEqual(400, rv.status_code)

    def test_pools_post(self):
        """ POST   /pools should accept valid request data 
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        self.assertEqual(302, rv.status_code)

    def test_pools_get_all(self):
        """ GET    /pools should report correct number of pools
        """
        rv = self.open_with_auth('/api/v2/pools', method='GET')
        self.assertEqual(0, len(json.loads(rv.data)['pools']))

        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v2/pools', method='GET')
        self.assertEqual(1, len(json.loads(rv.data)['pools']))

    def test_pools_get_one_reject_invalid_id(self):
        """ GET    /pools/<id> should return 404 on unknown id
        """
        rv = self.open_with_auth('/api/v2/pools/this_is_not_here', method='GET')
        self.assertEqual(404, rv.status_code)

    def test_pools_get_one_returns_200_on_hit(self):
        """ GET    /pools/<id> should return 200 on hit
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v2/pools/123456', method='GET')
        self.assertEqual(200, rv.status_code)

    def test_pools_get_one_returns_pool_data(self):
        """ GET    /pools/<id> should return pool data on hit
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v2/pools/123456', method='GET')
        self.assertEqual('123456', json.loads(rv.data)['id'])
        self.assertEqual([], json.loads(rv.data)['tasks'])
        self.assertEqual(2, len(json.loads(rv.data).keys()))

    def test_pools_delete_returns_404_on_unknown_id(self):
        """ DELETE /pools/<id> returns 404 on unknown id
        """
        rv = self.open_with_auth('/api/v2/pools/xxx', method='DELETE')
        self.assertEqual(404, rv.status_code)

    def test_pools_delete_actually_deletes(self):
        """ DELETE /pools/<id> sets deleted flag correctly
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v2/pools/123456', method='DELETE')
        rv = self.open_with_auth('/api/v2/pools', method='GET')
        self.assertEqual(0, len(json.loads(rv.data)['pools']))

    def test_pools_tasks_get_all_returns_404_on_missing_pool(self):
        """ GET    /pools/<id>/tasks returns 404 on missing pool
        """
        rv = self.open_with_auth('/api/v2/pools/xxx/tasks', method='GET')
        self.assertEqual(404, rv.status_code)        

    def test_pools_tasks_get_all_returns_200_on_hit(self):
        """ GET    /pools/<id>/tasks returns 200 on hit
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v2/pools/123456/tasks', method='GET')
        self.assertEqual(200, rv.status_code)        

    def test_pools_tasks_get_all_returns_tasks_on_hit(self):
        """ GET    /pools/<id>/tasks returns tasks
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v2/pools/123456/tasks', method='GET')
        self.assertEqual(0, len(json.loads(rv.data)['tasks']))

        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'zoo', 'tasks' : [
            {'kind' : 'text', 'id' : 'zzzxxx', 'task' : 'Hello WOrld!', 'answer' : ''}] }))
        self.assertEqual(302, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks', method='GET')
        self.assertEqual(1, len(json.loads(rv.data)['tasks']))

    def test_pools_tasks_get_one_returns_404_on_missing_pool(self):
        """ GET    /pools/<id>/tasks/<id> returns 404 on missing pool
        """
        rv = self.open_with_auth('/api/v2/pools/xxx/tasks/yyy', method='GET')
        self.assertEqual(404, rv.status_code)        

    def test_pools_tasks_get_one_returns_404_on_missing_task(self):
        """ GET    /pools/<id>/tasks/<id> returns 404 on existing pool but missing task
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'zoo', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks/hellow', method='GET')
        self.assertEqual(404, rv.status_code)

    def test_pools_tasks_get_one_returns_200_on_hit(self):
        """ GET    /pools/<id>/tasks/<id> returns 200 on hit
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'zoo', 'tasks' : [
            {'kind' : 'text', 'id' : 'zzzxxx', 'task' : 'Hello WOrld!', 'answer' : ''}] }))
        self.assertEqual(302, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks/zzzxxx', method='GET')
        self.assertEqual(200, rv.status_code)

    def test_pools_tasks_get_one_returns_correct_task(self):
        """ GET    /pools/<id>/tasks/<id> returns correct task
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'zoo', 'tasks' : [
            {'kind' : 'text', 'id' : 'zzzxxx', 'task' : 'Hello WOrld!', 'answer' : ''}] }))
        self.assertEqual(302, rv.status_code)
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks/zzzxxx', method='GET')
        self.assertEqual({'kind' : 'text', 'id' : 'zzzxxx', 'task' : 'Hello WOrld!', 'answer' : ''}, 
            json.loads(rv.data))

    def test_pools_tasks_post_returns_404_on_missing_pool(self):
        """ POST   /pools/<id>/tasks returns 404 on missing pool
        """
        rv = self.open_with_auth('/api/v2/pools/missing/tasks', method='POST')
        self.assertEqual(404, rv.status_code)

    def test_pools_tasks_post_returns_200_on_correct_task(self):
        """ POST   /pools/<id>/tasks returns 200 on correct task
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'zoo', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks', method='POST', 
            data=json.dumps({'kind' : 'text', 'id' : 'zzzxxx', 'task' : 'Hello WOrld!', 'answer' : ''}))
        self.assertEqual(200, rv.status_code)

    def test_pools_tasks_post_returns_200_on_correct_task(self):
        """ POST   /pools/<id>/tasks actually adds the task to the pool
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'zoo', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks', method='POST', 
            data=json.dumps({'kind' : 'text', 'id' : 'zzzxxx', 'task' : 'Hello WOrld!', 'answer' : ''}))

        # one way to check ...
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks/zzzxxx', method='GET')
        self.assertEqual(200, rv.status_code)        
        # and another
        rv = self.open_with_auth('/api/v2/pools/zoo/tasks', method='GET')
        self.assertEqual(1, len(json.loads(rv.data)['tasks']))

    def test_pools_delete(self):
        """ DELETE /pools should set the deleted flag on all pools
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'one', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'two', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'three', 'tasks' : [] }))

        rv = self.open_with_auth('/api/v2/pools', method='DELETE')

        rv = self.open_with_auth('/api/v2/pools', method='GET')
        self.assertEqual(0, len(json.loads(rv.data)['pools']))

    def test_pools_post_deleted_id(self):
        """ POST   /pools should return 400 also when the existing pool is marked deleted
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'one', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools', method='DELETE')
        
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'one', 'tasks' : [] }))
        self.assertEqual(400, rv.status_code)        

    def test_pools_versions_returns_404_on_missing_pool(self):
        """ GET    /pools/<id>/versions return 404 on missing pool
        """
        rv = self.open_with_auth('/api/v2/pools/xxx/versions', method='GET')
        self.assertEqual(404, rv.status_code)

    def test_pools_versions_returns_200_on_existing_pool(self):
        """ GET    /pools/<id>/versions return 200 on existing pool
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/123456/versions', method='GET')
        self.assertEqual(200, rv.status_code)

    def test_pools_versions_returns_versions(self):
        """ GET    /pools/<id>/versions return versions
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/123456/versions', method='GET')
        self.assertEqual(1, len(json.loads(rv.data).get('versions')))

    def test_pools_versions_returns_correct_number_ofversions(self):
        """ GET    /pools/<id>/versions returns correct number of versions
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/123456/tasks', method='POST', 
            data=json.dumps({'kind' : 'text', 'id' : 'grammar1', 'task' : 'Hello WOrld!', 'answer' : ''}))

        rv = self.open_with_auth('/api/v2/pools/123456/versions', method='GET')
        self.assertEqual(2, len(json.loads(rv.data).get('versions')))

        rv = self.open_with_auth('/api/v2/pools/123456/tasks', method='POST', 
            data=json.dumps({'kind' : 'text', 'id' : 'grammar2', 'task' : 'Hello WOrld!', 'answer' : ''}))

        rv = self.open_with_auth('/api/v2/pools/123456/versions', method='GET')
        self.assertEqual(3, len(json.loads(rv.data).get('versions')))

    def test_pools_versions_404_or_200_on_version_hit_or_miss(self):
        """ GET    /pools/<id>/versions/<version> returns 404 or 200 correctly
        """
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/123456/tasks', method='POST', 
            data=json.dumps({'kind' : 'text', 'id' : 'grammar1', 'task' : 'Hello WOrld!', 'answer' : ''}))

        rv = self.open_with_auth('/api/v2/pools/123456/versions/1', method='GET')
        self.assertEqual(200, rv.status_code)

        rv = self.open_with_auth('/api/v2/pools/123456/versions/2', method='GET')
        self.assertEqual(200, rv.status_code)

        rv = self.open_with_auth('/api/v2/pools/123456/versions/3', method='GET')
        self.assertEqual(404, rv.status_code)

    def test_pools_versions_version_altering_operations(self):
        """ GET    /pools/<id>/versions/<version> correctly tracks operations (POST and DELETE)
        """
        # create a pool
        rv = self.open_with_auth('/api/v2/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : [] }))
        rv = self.open_with_auth('/api/v2/pools/123456/versions/1', method='GET')
        self.assertEqual(200, rv.status_code)

        # add a task
        rv = self.open_with_auth('/api/v2/pools/123456/tasks', method='POST', 
            data=json.dumps({'kind' : 'text', 'id' : 'grammar1', 'task' : 'Hello WOrld!', 'answer' : ''}))
        rv = self.open_with_auth('/api/v2/pools/123456/versions/2', method='GET')
        self.assertEqual(200, rv.status_code)

        # delete a specific task
        rv = self.open_with_auth('/api/v2/pools/123456/tasks/grammar1', method='DELETE')
        self.assertEqual(200, rv.status_code)

        rv = self.open_with_auth('/api/v2/pools/123456/versions/3', method='GET')
        self.assertEqual(200, rv.status_code)

        rv = self.open_with_auth('/api/v2/pools/123456/versions/4', method='GET')
        self.assertEqual(404, rv.status_code)