diff --git a/tests/test_api.py b/tests/test_api.py index e8631c6..2283790 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,188 +1,199 @@ -import httplib, json, base64, hmac, hashlib +import httplib, json, base64, hmac, hashlib, email from twisted.trial import unittest class TestAPI(unittest.TestCase): reqid = 0 # test config config = { 'api_host': '127.0.0.1', 'api_port': 8080, 'api_user': 'webclient', 'api_pass': 'Welcome2KolabSystems', 'api_secret': '8431f19170e7f90d4107bf4b169baf' } @classmethod def setUp(self): # TODO: reset riak buckets and fill with sample data # TODO: fire up the web service pass def _api_request(self, user, method, sign=True, **kwargs): """ Helper method to send JSON-RPC calls to the API """ headers = { 'Authorization': 'Basic ' + base64.b64encode(self.config['api_user'] + ':' + self.config['api_pass']), 'Content-type': "application/json", 'X-Request-User': user } self.rpcerror = None self.reqid += 1 body = { 'jsonrpc': '2.0', 'id': self.reqid, 'method': method, 'params': kwargs } sbody = json.dumps(body) if sign: headers['X-Request-Sign'] = hmac.new( key=self.config['api_secret'], msg=headers.get('X-Request-User') + ':' + sbody, digestmod=hashlib.sha256 ).hexdigest() try: conn = httplib.HTTPConnection(self.config['api_host'], self.config['api_port']) conn.request('POST', '/api/rpc', sbody, headers) result = None response = conn.getresponse() if response.status == 200: result = json.loads(response.read()) if result.has_key('error'): self.rpcerror = result['error'] if result.has_key('result') and result.has_key('id') and result['id'] == self.reqid: return result['result'] # print "JSON-RPC response: %d %s; %r" % (response.status, response.reason, result) except Exception, e: print "JSON-RPC error: %r" % (e) return False def assertRPCError(self, code=None): self.assertIsInstance(self.rpcerror, dict) if code is not None: self.assertEqual(self.rpcerror['code'], code) def test_001_json_rpc(self): res = self._api_request('', 'system.keygen', sign=False) self.assertFalse(res) self.assertRPCError(-32600) res = self._api_request('', 'system.keygen', sign=True) self.assertIsInstance(res, dict) self.assertTrue(res.has_key('key')) self.assertTrue(res.has_key('expires')) def test_002_invalid_request(self): res = self._api_request('john.doe@example.org', 'event.bogus', uid='6EE0570E8CA21DDB67FC9ADE5EE38E7F-XXXXXXXXXXXXXXX' ) self.assertFalse(res) self.assertRPCError(-32601) # Method not found res = self._api_request('john.doe@example.org', 'event.created', uid='6EE0570E8CA21DDB67FC9ADE5EE38E7F-A4BF5BBB9FEAA271' ) self.assertFalse(res) self.assertRPCError(-32602) # Invalid params res = self._api_request('john.doe@example.org', 'event.created', uid='6EE0570E8CA21DDB67FC9ADE5EE38E7F-INVALID-UID', mailbox='Calendar' ) self.assertFalse(res) def test_003_created_lastmodified(self): created = self._api_request('john.doe@example.org', 'event.created', uid='6EE0570E8CA21DDB67FC9ADE5EE38E7F-A4BF5BBB9FEAA271', mailbox='Calendar', msguid=4 ) self.assertIsInstance(created, dict) self.assertTrue(created.has_key('date')) self.assertTrue(created.has_key('user')) self.assertTrue(created.has_key('rev')) self.assertIn('john.doe@example.org', created['user']) changed = self._api_request('john.doe@example.org', 'event.lastmodified', uid='6EE0570E8CA21DDB67FC9ADE5EE38E7F-A4BF5BBB9FEAA271', mailbox='Calendar' ) self.assertIsInstance(changed, dict) self.assertTrue(changed.has_key('rev')) self.assertTrue(changed['rev'] > created['rev']) self.assertTrue(changed['date'] > created['date']) def test_004_changelog(self): changelog = self._api_request('john.doe@example.org', 'event.changelog', uid='5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271', mailbox='Calendar', msguid=22 ) # print json.dumps(changelog, indent=4) self.assertIsInstance(changelog, dict) self.assertTrue(changelog.has_key('changes')) self.assertTrue(changelog.has_key('uid')) self.assertEqual(len(changelog['changes']), 2) one = changelog['changes'][0] two = changelog['changes'][1] self.assertTrue(two['rev'] > one['rev']) def test_005_get_revision(self): changelog = self._api_request('john.doe@example.org', 'event.changelog', uid='5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271', mailbox='Calendar' ) self.assertIsInstance(changelog, dict) self.assertEqual(len(changelog.get('changes', [])), 2) rev1 = self._api_request('john.doe@example.org', 'event.get', uid='5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271', mailbox='Calendar', rev=changelog['changes'][0]['rev'] ) self.assertIsInstance(rev1, dict) self.assertTrue(rev1.has_key('xml')) self.assertNotIn('', rev1['xml']) rev2 = self._api_request('john.doe@example.org', 'event.get', uid='5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271', mailbox='Calendar', rev=changelog['changes'][1]['rev'] ) self.assertIsInstance(rev2, dict) self.assertTrue(rev2.has_key('xml')) self.assertIn('Somewhere else', rev2['xml']) + msgdata = self._api_request('john.doe@example.org', 'event.rawdata', + uid='5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271', + mailbox='Calendar', + rev=changelog['changes'][0]['rev'] + ) + self.assertIsInstance(msgdata, unicode) + + message = email.message_from_string(msgdata.encode('utf8','replace')) + self.assertIsInstance(message, email.message.Message) + self.assertTrue(message.is_multipart()) + def test_006_diff(self): changelog = self._api_request('john.doe@example.org', 'event.changelog', uid='5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271', mailbox='Calendar' ) self.assertIsInstance(changelog, dict) self.assertEqual(len(changelog.get('changes', [])), 2) rev1 = changelog['changes'][0]['rev'] rev2 = changelog['changes'][1]['rev'] diff = self._api_request('john.doe@example.org', 'event.diff', uid='5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271', mailbox='Calendar', rev=rev1+':'+rev2 ) # print json.dumps(diff, indent=4) self.assertIsInstance(diff, dict) self.assertTrue(diff.has_key('changes')) self.assertTrue(diff.has_key('uid')) self.assertEqual(len(diff['changes']), 6) \ No newline at end of file diff --git a/tests/test_storage.py b/tests/test_storage.py index bb9f9d5..efdfb96 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -1,77 +1,80 @@ import os, flask, json, email from app import storage from twisted.trial import unittest class FlaskCurrentApp(object): config = dict( STORAGE=dict( backend='riak', riak_host='127.0.0.1', riak_port='10018' ), CONFIG_DIR=os.path.join(os.path.abspath(os.path.dirname(__file__)), '..', 'config') ) def __init__(self): import logging.config logging.config.fileConfig(self.config['CONFIG_DIR'] + '/bonnie-flask.conf') class TestStorage(unittest.TestCase): def setUp(self): # patch current_app to return static config self.patch(flask, 'current_app', FlaskCurrentApp()) def test_000_instance(self): strg = storage.factory() self.assertIsInstance(strg, storage.AbstractStorage) self.assertIsInstance(strg, storage.riak_storage.RiakStorage) def test_001_get_user_by_name(self): strg = storage.factory() user = strg.get_user(username='john.doe@example.org') self.assertIsInstance(user, dict) self.assertEqual(user['id'], '55475201-bdc211e4-881c96ef-f248ab46') self.assertEqual(user['user'], 'john.doe@example.org') def test_002_get_user_by_id(self): strg = storage.factory() user = strg.get_user(id='55475201-bdc211e4-881c96ef-f248ab46') self.assertIsInstance(user, dict) self.assertEqual(user['user'], 'john.doe@example.org') self.assertEqual(user['id'], '55475201-bdc211e4-881c96ef-f248ab46') def test_010_get_folder_id(self): strg = storage.factory() folder = strg.get_folder('user/john.doe/Calendar@example.org') self.assertIsInstance(folder, dict) self.assertEqual(folder['id'], 'a5660caa-3165-4a84-bacd-ef4b58ef3663') def test_020_get_events(self): strg = storage.factory() events = strg.get_events('6EE0570E8CA21DDB67FC9ADE5EE38E7F-A4BF5BBB9FEAA271', 'user/john.doe/Calendar@example.org', 4) self.assertEqual(len(events), 3) self.assertEqual(events[0]['event'], 'MessageAppend') self.assertEqual(events[0]['uidset'], '3') self.assertEqual(events[1]['event'], 'MessageAppend') self.assertEqual(events[1]['uidset'], '4') self.assertEqual(events[2]['event'], 'MessageTrash') self.assertEqual(events[2]['uidset'], '3') #print json.dumps(events, indent=4) def test_025_get_revision(self): strg = storage.factory() uid = '5A637BE7895D785671E1732356E65CC8-A4BF5BBB9FEAA271' mailbox = 'user/john.doe/Calendar@example.org' events = strg.get_events(uid, mailbox, None, limit=1) self.assertEqual(len(events), 1) rec = strg.get_revision(uid, mailbox, None, events[0]['revision']) self.assertIsInstance(rec, dict) - self.assertTrue(rec.has_key('message')) self.assertEqual(rec['event'], 'MessageAppend') - message = email.message_from_string(rec['message'].encode('utf8','replace')) + msgsource = strg.get_message_data(rec) + self.assertIsInstance(msgsource, unicode) + + message = email.message_from_string(msgsource.encode('utf8','replace')) self.assertIsInstance(message, email.message.Message) + self.assertTrue(message.is_multipart())