Source

bloodhound-rpc / trunk / tracrpc / tests / admin.py

Full commit
# -*- coding: utf-8 -*-
"""
License: BSD

(c) 2014      ::: Olemis Lang (olemis+trac@gmail.com)
"""


import datetime
import os
import shutil
from StringIO import StringIO
import time
import unittest
import urllib
import xmlrpclib

from trac.util.datefmt import to_datetime, utc
from tracrpc.tests import TracRpcTestCase

class RpcAdminTicketTestCase(TracRpcTestCase):
    BH_IN_DEFAULT_PRODUCT = True

    def setUp(self):
        TracRpcTestCase.setUp(self)
        self.anon = self._tester.server_proxy(self.url_anon)
        self.user = self._tester.server_proxy(self.url_auth, user='user')
        self.admin = self._tester.server_proxy(self.url_auth, user='admin')

    def tearDown(self):
        TracRpcTestCase.tearDown(self)

    def test_admin_ticket_crud_as_user(self):
        timestamp = datetime.datetime(2000, 1, 1, 10, 20, 30)
        timestamp = xmlrpclib.DateTime(to_datetime(timestamp, utc))
        r = self.admin.admin.exec_as([{'methodName' : 'ticket.create', 
                                       'params' : ["create_get_delete_as_user",
                                                   "fooy", {}]
                                      }], 'user', timestamp)
        self.assertEquals(1, len(r))
        self.assertTrue(isinstance(r[0][0], int), "Expected 'int' but got '%s'" %
                                   (r[0][0],))
        tid = r[0][0]

        tid, time_created, time_changed, attributes = self.admin.ticket.get(tid)
        self.assertEquals(timestamp, time_created)
        self.assertEquals(timestamp, time_changed)
        self.assertEquals('fooy', attributes['description'])
        self.assertEquals('create_get_delete_as_user', attributes['summary'])
        self.assertEquals('new', attributes['status'])
        self.assertEquals('user', attributes['reporter'])
        self.admin.ticket.delete(tid)

    def test_admin_ticket_getRecentChanges(self):
        timestamp1 = datetime.datetime(2000, 1, 1, 10, 20, 30)
        timestamp1 = xmlrpclib.DateTime(to_datetime(timestamp1, utc))
        r = self.admin.admin.exec_as([{'methodName' : 'ticket.create', 
                                       'params' : ["admin_ticket_getRecentChanges",
                                                   "one", {}]
                                      }], 'admin', timestamp1)
        self.assertEquals(1, len(r))
        self.assertTrue(isinstance(r[0][0], int), "Expected 'int' but got '%s'" %
                                               (r[0][0],))
        tid1 = r[0][0]

        timestamp2 = datetime.datetime(2001, 2, 2, 11, 21, 31)
        timestamp2 = xmlrpclib.DateTime(to_datetime(timestamp2, utc))
        r = self.admin.admin.exec_as([{'methodName' : 'ticket.create', 
                                       'params' : ["admin_ticket_getRecentChanges",
                                                   "two", {}]
                                      }], 'admin', timestamp2)
        self.assertEquals(1, len(r))
        self.assertTrue(isinstance(r[0][0], int), "Expected 'int' but got '%s'" %
                                               (r[0][0],))
        tid2 = r[0][0]

        timestamp = datetime.datetime(2001, 1, 1)
        timestamp = xmlrpclib.DateTime(to_datetime(timestamp, utc))
        changes = self.admin.ticket.getRecentChanges(timestamp)
        try:
            self.assertEquals(changes, [tid2])
        finally:
            self.admin.ticket.delete(tid1)
            self.admin.ticket.delete(tid2)

    def test_admin_ticket_update_author(self):
        tid = self.admin.ticket.create("", "one", {})

        r = self.admin.admin.exec_as([{'methodName' : 'ticket.update', 
                                       'params' : [tid, "comment1", {}]
                                      }], 'admin')
        self.assertEquals(1, len(r))
        self.assertFalse(isinstance(r[0], xmlrpclib.Fault), "Unexpected failure")
        time.sleep(1)

        r = self.admin.admin.exec_as([{'methodName' : 'ticket.update', 
                                       'params' : [tid, "comment2", {}]
                                      }], 'foo')
        self.assertEquals(1, len(r))
        self.assertFalse(isinstance(r[0], xmlrpclib.Fault), "Unexpected failure")
        time.sleep(1)

        r = self.admin.admin.exec_as([{'methodName' : 'ticket.update', 
                                       'params' : [tid, "comment3", {},
                                                   False, 'user']
                                      }], 'should_be_rejected')
        self.assertEquals(1, len(r))
        self.assertFalse(isinstance(r[0], xmlrpclib.Fault), "Unexpected failure")
        changes = self.admin.ticket.changeLog(tid)
        self.assertEquals(3, len(changes))
        for when, who, what, cnum, comment, _tid in changes:
            self.assertTrue(comment in ('comment1', 'comment2', 'comment3'))
            if comment == 'comment1':
                self.assertEquals('admin', who)
            if comment == 'comment2':
                self.assertEquals('foo', who)
            if comment == 'comment3':
                self.assertEquals('user', who)
        self.admin.ticket.delete(tid)

    def test_admin_create_at_time(self):
        now = to_datetime(None, utc)
        minus1 = now - datetime.timedelta(days=1)
        minus2 = now - datetime.timedelta(days=2)
        # create the tickets
        one, two = self.admin.admin.exec_as([{'methodName' : 'ticket.create',
                                              'params' : ["create_at_time1",
                                                          "ok", {}, False]},
                                             {'methodName' : 'ticket.create',
                                              'params' : ["create_at_time3",
                                                          "ok", {}, False,
                                                          xmlrpclib.DateTime(minus2)
                                                         ]}
                                            ], 'auser', xmlrpclib.DateTime(minus1))
        self.assertTrue(all(not isinstance(x, xmlrpclib.Fault) for x in (one, two)))
        one, = one
        two, = two
        # get the tickets
        t1 = self.admin.ticket.get(one)
        t2 = self.admin.ticket.get(two)
        # check timestamps
        self.assertTrue(t2[1] < t1[1])
        self.admin.ticket.delete(one)
        self.admin.ticket.delete(two)

    def test_update_at_time(self):
        from trac.util.datefmt import to_datetime, utc
        now = to_datetime(None, utc)
        minus1 = now - datetime.timedelta(hours=1)
        minus2 = now - datetime.timedelta(hours=2)
        minus3 = now - datetime.timedelta(minutes=3)
        tid = self.admin.ticket.create("ticket_update_at_time", "ok", {})
        self.admin.admin.exec_as([{'methodName' : 'ticket.update',
                                  'params' : [tid, 'one', {}, False, '',
                                              xmlrpclib.DateTime(minus2)]},
                                  {'methodName' : 'ticket.update',
                                   'params' : [tid, 'two', {}, False, '',
                                              xmlrpclib.DateTime(minus1)]},
                                  {'methodName' : 'ticket.update',
                                   'params' : [tid, 'three', {}, False, 'usr']
                                   }], 'auser', minus3)
        self.user.ticket.update(tid, 'four', {})
        changes = self.admin.ticket.changeLog(tid)
        self.assertEquals(4, len(changes))
        # quick test to make sure each is older than previous
        self.assertTrue(changes[0][0] < changes[1][0] < changes[2][0])
        # assertions for user names
        self.assertEquals('auser', changes[0][1])
        self.assertEquals('auser', changes[1][1])
        self.assertEquals('usr', changes[2][1])
        self.assertEquals('user', changes[3][1])
        # margin of 2 seconds for tests
        justnow = xmlrpclib.DateTime(now - datetime.timedelta(minutes=4))
        self.assertTrue(justnow <= changes[2][0])
        self.assertTrue(justnow <= changes[3][0])
        self.admin.ticket.delete(tid)


class RpcAdminWikiTestCase(TracRpcTestCase):

    def setUp(self):
        TracRpcTestCase.setUp(self)
        self.anon = self._tester.server_proxy(self.url_anon)
        self.user = self._tester.server_proxy(self.url_auth, user='user')
        self.admin = self._tester.server_proxy(self.url_auth, user='admin')

    def tearDown(self):
        TracRpcTestCase.tearDown(self)


class RpcAdminWikiTestCase(TracRpcTestCase):

    def setUp(self):
        TracRpcTestCase.setUp(self)
        self.anon = self._tester.server_proxy(self.url_anon)
        self.user = self._tester.server_proxy(self.url_auth, user='user')
        self.admin = self._tester.server_proxy(self.url_auth, user='admin')

    def tearDown(self):
        TracRpcTestCase.tearDown(self)

    def test_admin_wiki_attachments(self):
        # Note: Quite similar to the tracrpc.tests.json.JsonTestCase.test_binary
        image_url = os.path.join(self._testenv.trac_src, 'trac',
                             'htdocs', 'feed.png')
        image_in = StringIO(open(image_url, 'r').read())
        # Create wiki page
        timestamp = datetime.datetime(2000, 1, 1, 1, 1, 1)
        delta = datetime.timedelta(days=5, minutes=5)
        self.admin.admin.exec_as([{'methodName' : 'wiki.putPage',
                                   'params' : ['TestAdminAttachments', 
                                               'update 1\n', 
                                               dict(comment='update 1')]
                                   }], 'user1', xmlrpclib.DateTime(timestamp))
        info = self.admin.wiki.getPageInfo('TestAdminAttachments')
        self.assertTrue(isinstance(info, dict))
        self.assertEquals(xmlrpclib.DateTime(datetime.datetime(2000, 1, 1,
                                                               1, 1, 1)),
                          info.get('lastModified'))
        self.assertEquals('user1', info.get('author'))
        self.assertEquals('update 1', info.get('comment'))
        # Create attachment
        timestamp += delta
        self.admin.admin.exec_as([{'methodName' : 'wiki.putAttachmentEx',
                                   'params' : ['TestAdminAttachments', 
                                               'feed2.png', 'update 2\n',
                                               xmlrpclib.Binary(image_in.getvalue())]
                                   }], 'user2', xmlrpclib.DateTime(timestamp))
        # TODO: Assertions for attachment meta-data
        self.assertEquals(image_in.getvalue(), self.admin.wiki.getAttachment(
                                                'TestAdminAttachments/feed2.png').data)
        # Update attachment (adding new)
        timestamp += delta
        self.admin.admin.exec_as([{'methodName' : 'wiki.putAttachmentEx',
                                   'params' : ['TestAdminAttachments', 
                                               'feed2.png', 'update 3\n',
                                               xmlrpclib.Binary(image_in.getvalue()),
                                               False]
                                   }], 'user2', xmlrpclib.DateTime(timestamp))
        self.assertEquals(image_in.getvalue(), self.admin.wiki.getAttachment(
                                                'TestAdminAttachments/feed2.2.png').data)
        # TODO: Assertions for attachment meta-data

        # List attachments
        self.assertEquals(['TestAdminAttachments/feed2.2.png', 'TestAdminAttachments/feed2.png'],
                        sorted(self.admin.wiki.listAttachments('TestAdminAttachments')))
        # Delete both attachments
        self.admin.wiki.deleteAttachment('TestAdminAttachments/feed2.png')
        self.admin.wiki.deleteAttachment('TestAdminAttachments/feed2.2.png')
        # List attachments again
        self.assertEquals([], self.admin.wiki.listAttachments('TestAdminAttachments'))


class RpcAdminMilestoneTestCase(TracRpcTestCase):
    BH_IN_DEFAULT_PRODUCT = True

    def setUp(self):
        TracRpcTestCase.setUp(self)
        self.anon = self._tester.server_proxy(self.url_anon)
        self.user = self._tester.server_proxy(self.url_auth, user='user')
        self.admin = self._tester.server_proxy(self.url_auth, user='admin')

    def tearDown(self):
        TracRpcTestCase.tearDown(self)

    def test_admin_milestone_attachments(self):
        # Note: Quite similar to the tracrpc.tests.json.JsonTestCase.test_binary
        image_url = os.path.join(self._testenv.trac_src, 'trac',
                                 'htdocs', 'feed.png')
        image_in = StringIO(open(image_url, 'r').read())
        # Add new milestone
        mname = 'test_admin_milestone_attachments'
        self.admin.ticket.milestone.create(mname, {})
        # Create attachment
        now = to_datetime(None, utc)
        minus1 = now - datetime.timedelta(hours=1)
        minus2 = now - datetime.timedelta(hours=2)
        self.admin.admin.exec_as([{'methodName' : 'ticket.milestone.putAttachment',
                                   'params' : [mname, 'feed2.png', 'upload1',
                                               xmlrpclib.Binary(image_in.getvalue())]
                                   }], 'user1', minus2)
        self.assertEquals(image_in.getvalue(), 
                          self.admin.ticket.milestone.getAttachment(mname,
                                                           'feed2.png').data)
        # Update attachment (adding new)
        self.admin.admin.exec_as([{'methodName' : 'ticket.milestone.putAttachment',
                                   'params' : [mname, 'feed2.png', 'upload2',
                                               xmlrpclib.Binary(image_in.getvalue()),
                                               False]}], 'user2', minus1)
        self.assertEquals(image_in.getvalue(), 
                          self.admin.ticket.milestone.getAttachment(mname,
                                                            'feed2.2.png').data)
        # List attachments
        attachments = sorted(self.admin.ticket.milestone.listAttachments(mname),
                             key=lambda x : x[3])

        self.assertEquals(2, len(attachments))

        attachment = attachments[0]
        self.assertEquals('feed2.png', attachment[0])
        self.assertEquals('upload1', attachment[1])
        self.assertEquals(len(image_in.getvalue()), attachment[2])
        self.assertEquals(xmlrpclib.DateTime(minus2), attachment[3])
        self.assertEquals('user1', attachment[4])

        attachment = attachments[1]
        self.assertEquals('feed2.2.png', attachment[0])
        self.assertEquals('upload2', attachment[1])
        self.assertEquals(len(image_in.getvalue()), attachment[2])
        self.assertEquals(xmlrpclib.DateTime(minus1), attachment[3])
        self.assertEquals('user2', attachment[4])

        # Delete both attachments
        self.admin.ticket.milestone.deleteAttachment(mname, 'feed2.png')
        self.admin.ticket.milestone.deleteAttachment(mname, 'feed2.2.png')
        # List attachments again
        self.assertEquals([],
                          self.admin.ticket.milestone.listAttachments(mname))


def test_suite(suite=None):
    if suite is None:
        from tracrpc.tests import functionalSuite
        suite = functionalSuite()
    # FIXME: Improve
    for ts in [unittest.makeSuite(RpcAdminTicketTestCase),
               unittest.makeSuite(RpcAdminWikiTestCase),
               unittest.makeSuite(RpcAdminMilestoneTestCase),
               ]:
        for test in ts._tests:
            suite.addTest(test)
    return suite

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