Source

audrid / tests / test_api.py

Full commit
from audrid.models import (
    Audit,
    Document, 
    Exam, 
    Log,
    User,
)
from audrid import (app, db, utils)
import base64
import json
import time
import unittest

class APITests(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://ezri:ezri@localhost/ezri'
        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):
        headers = {
            'Authorization' : 'Basic %s' % (
            base64.b64encode("%s:%s" % (username, password)))
        }
        return self.app.open(url, method=method, headers=headers, data=data)

    def test_pools(self):
        rv = self.open_with_auth('/api/v1/pools')
        self.assertEqual(200, rv.status_code)
        self.assertEqual({"pools" : []}, json.loads(rv.data))

        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({}))
        self.assertEqual(400, rv.status_code)

        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : 'x'}))
        self.assertEqual(400, rv.status_code)

        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : '123456'}))
        self.assertEqual(400, rv.status_code)

        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        self.assertEqual(200, rv.status_code)
        self.assertEqual(1, Document.query.count())
        self.assertEqual(1, Log.query.count())
        self.assertIsNotNone(Document.query.filter_by(id='123456').first())

        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        self.assertEqual(400, rv.status_code)

    def test_pools_delete(self):
        rv = self.open_with_auth('/api/v1/pools', method='DELETE')
        self.assertEqual(200, rv.status_code)
        
        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'000000', 'tasks' : []}))
        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'000001', 'tasks' : []}))
        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'000002', 'tasks' : []}))

        self.assertEqual(0, Document.query.filter_by(deleted=True).count())

        rv = self.open_with_auth('/api/v1/pools', method='DELETE')
        self.assertEqual(200, rv.status_code)
        self.assertEqual(3, Document.query.filter_by(deleted=True).count())

    def test_exams(self):
        rv = self.open_with_auth('/api/v1/exams', method='GET')
        self.assertEqual(200, rv.status_code)
        self.assertEqual({u"exams" : []}, json.loads(rv.data))

        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v1/exams', method='POST', data=json.dumps({"pool" : u'123456'}))
        self.assertEqual(400, rv.status_code)

        rv = self.open_with_auth('/api/v1/exams', method='POST', data=json.dumps({"name" : u'123456'}))
        self.assertEqual(400, rv.status_code)

        rv = self.open_with_auth('/api/v1/exams', method='POST', data=json.dumps({"pool_id" : u'123456', 'name' : u'This is a test pool'}))
        self.assertEqual(200, rv.status_code)

        rv = self.open_with_auth('/api/v1/exams', method='GET')
        self.assertEqual(200, rv.status_code)
        self.assertTrue('exams' in json.loads(rv.data))
        self.assertEqual(1, len(json.loads(rv.data)['exams']))

    def test_exams_delete(self):
        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v1/exams', method='POST', data=json.dumps({"pool_id" : u'123456', "name" : "Test exam 12"}))
        self.assertEqual(1, Exam.query.filter_by(deleted=False).count())
        rv = self.open_with_auth('/api/v1/exams', method='DELETE')
        self.assertEqual(0, Exam.query.filter_by(deleted=False).count())

    def test_audits(self):
        rv = self.open_with_auth('/api/v1/audits', method='GET')
        self.assertEqual(200, rv.status_code)
        self.assertEqual({u"audits" : []}, json.loads(rv.data))

        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v1/exams', method='POST', data=json.dumps({"pool_id" : u'123456', 'name' : u'This is a test pool'}))
        rv = self.open_with_auth('/api/v1/audits', method='POST', data=json.dumps({"exam_id" : Exam.query.first().id, "user_id" : 1}))
        self.assertEqual(200, rv.status_code)

        rv = self.open_with_auth('/api/v1/audits', method='GET')
        self.assertEqual(200, rv.status_code)
        self.assertEqual(1 , len(json.loads(rv.data)['audits']))

        self.assertEqual(2, Document.query.count())
        self.assertEqual(2, Log.query.count())
        self.assertEqual(1, Exam.query.count())
        self.assertEqual(1, Audit.query.count())

    def test_audits_delete(self):
        rv = self.open_with_auth('/api/v1/pools', method='POST', data=json.dumps({'id' : u'123456', 'tasks' : []}))
        rv = self.open_with_auth('/api/v1/exams', method='POST', data=json.dumps({"pool_id" : u'123456', 'name' : u'This is a test pool'}))
        rv = self.open_with_auth('/api/v1/audits', method='POST', data=json.dumps({"exam_id" : Exam.query.first().id, "user_id" : 1}))
        rv = self.open_with_auth('/api/v1/audits', method='DELETE')

        self.assertEqual(2, Document.query.count())
        self.assertEqual(3, Log.query.count())
        self.assertEqual(1, Exam.query.count())
        self.assertEqual(0, Audit.query.count())