Synnefo » ./kamaki

import sys

from StringIO import StringIO

from kamaki.clients.connection.kamakicon import KamakiHTTPConnection

class testKamakiCon(unittest.TestCase):

    def setUp(self):

        self.async_pool = None

        self.conn1 = KamakiHTTPConnection()

        self.conn2 = KamakiHTTPConnection()

        self.conn3 = KamakiHTTPConnection()

        self.conn4 = KamakiHTTPConnection()

        account = 'saxtouri@grnet.gr'

        self.conn1.url =\

            'https://pithos.okeanos.io/v1/%s/pithos?path=files' % account

        self.conn1.set_header('X-Auth-Token', '0TpoyAXqJSPxLdDuZHiLOA==')

        self.conn2.url = 'https://pithos.okeanos.io/v1/%s/pithos' % account

        self.conn2.set_header('X-Auth-Token', '0TpoyAXqJSPxLdDuZHiLOA==')

        self.conn3.url =\

            'https://pithos.okeanos.io/v1/%s/pithos?path=subdir' % account

        self.conn3.set_header('X-Auth-Token', '0TpoyAXqJSPxLdDuZHiLOA==')

        self.conn4.url = 'https://pithos.okeanos.io/v1/%s' % account

        self.conn4.set_header('X-Auth-Token', '0TpoyAXqJSPxLdDuZHiLOA==')

    def tearDown(self):

        pass

    def _get_async_content(self, con, **kwargs):

        class SilentGreenlet(gevent.Greenlet):

            def _report_error(self, exc_info):

                _stderr = None

                try:

                    _stderr = sys._stderr

                    sys.stderr = StringIO()

                    gevent.Greenlet._report_error(self, exc_info)

                finally:

                    sys.stderr = _stderr

        POOL_SIZE = 2

        if self.async_pool is None:

            self.async_pool = gevent.pool.Pool(size=POOL_SIZE)

        g = SilentGreenlet(self._get_content_len, con, **kwargs)

        self.async_pool.start(g)

        return g

    def _get_content_len(self, con, **kwargs):

        r = con.perform_request('GET', **kwargs)

        return len(r.content)

    def test_gevents(self):

        h1 = self._get_async_content(self.conn1)

        h2 = self._get_async_content(self.conn2)

        h3 = self._get_async_content(self.conn3)

        h4 = self._get_async_content(self.conn2,

            async_headers={'X-Auth-Token': 'FAKETOKEN'})

        h5 = self._get_async_content(self.conn1)

        while not (h1.ready()\

            and h2.ready()\

            and h3.ready()\

            and h4.ready()\

            and h5.ready()):

            sleep(.000001)

        r1 = h1.value

        r2 = h2.value

        # r3 = h3.value

        r4 = h4.value

        r5 = h5.value

        self.assertEqual(r1, r5)

        self.assertNotEqual(r2, r4)

        #print('1:%s 2:%s 3:%s 4:%s 5:%s'%(r1, r2, r3, r4, r5))

        gevent.joinall([h1, h2, h3, h4, h5])

    suiteFew = unittest.TestSuite()

    suiteFew.addTest(unittest.makeSuite(testKamakiCon))

    unittest.TextTestRunner(verbosity=2).run(suiteFew)