Did I find the right examples for you? yes no      Crawl my project      Python Jobs

All Samples(17)  |  Call(17)  |  Derive(0)  |  Import(0)

src/q/l/qless-py-0.10.1/test/test_worker.py   qless-py(Download)
    def test_proctitle(self):
        '''Make sure we can get / set the process title'''
        before = Worker.title()
        Worker.title('Foo')
        self.assertNotEqual(before, Worker.title())
        with open('test/tmp/file.out', 'w+'):
            pass
        self.assertNotEqual(os.listdir('test/tmp'), [])
        Worker.clean('test/tmp')
        self.assertEqual(os.listdir('test/tmp'), [])
                    pass
            # Now it's sullied. Clean it up
            self.assertNotEqual(os.listdir(path), [])
            with Worker.sandbox(path):
                self.assertEqual(os.listdir(path), [])

src/q/l/qless-py-HEAD/test/test_worker.py   qless-py(Download)
    def test_proctitle(self):
        '''Make sure we can get / set the process title'''
        before = Worker.title()
        Worker.title('Foo')
        self.assertNotEqual(before, Worker.title())
        with open('test/tmp/file.out', 'w+'):
            pass
        self.assertNotEqual(os.listdir('test/tmp'), [])
        Worker.clean('test/tmp')
        self.assertEqual(os.listdir('test/tmp'), [])
                    pass
            # Now it's sullied. Clean it up
            self.assertNotEqual(os.listdir(path), [])
            with Worker.sandbox(path):
                self.assertEqual(os.listdir(path), [])

src/q/l/qless-core-HEAD/test/test_locks.py   qless-core(Download)
        # However, once the grace period passes, we should be fine
        self.assertNotEqual(
            self.lua('pop', expires + self.grace, 'queue', 'another', 10), {})
 
    def test_repeated(self):

src/q/l/qless-py-0.10.1/test/test_client.py   qless-py(Download)
    def test_basic(self):
        '''Can give us access to jobs'''
        self.assertEqual(self.client.jobs['jid'], None)
        self.client.queues['foo'].put('Foo', {}, jid='jid')
        self.assertNotEqual(self.client.jobs['jid'], None)
    def test_recurring(self):
        '''Can give us access to recurring jobs'''
        self.assertEqual(self.client.jobs['jid'], None)
        self.client.queues['foo'].recur('Foo', {}, 60, jid='jid')
        self.assertNotEqual(self.client.jobs['jid'], None)
    def test_basic(self):
        '''Gives us access to queues'''
        self.assertNotEqual(self.client.queues['foo'], None)
 
    def test_counts(self):

src/q/l/qless-py-HEAD/test/test_client.py   qless-py(Download)
    def test_basic(self):
        '''Can give us access to jobs'''
        self.assertEqual(self.client.jobs['jid'], None)
        self.client.queues['foo'].put('Foo', {}, jid='jid')
        self.assertNotEqual(self.client.jobs['jid'], None)
    def test_recurring(self):
        '''Can give us access to recurring jobs'''
        self.assertEqual(self.client.jobs['jid'], None)
        self.client.queues['foo'].recur('Foo', {}, 60, jid='jid')
        self.assertNotEqual(self.client.jobs['jid'], None)
    def test_basic(self):
        '''Gives us access to queues'''
        self.assertNotEqual(self.client.queues['foo'], None)
 
    def test_counts(self):

src/q/l/qless-py-0.10.1/test/test_config.py   qless-py(Download)
    def test_update(self):
        '''We can use dictionary-style update'''
        updated = dict((key, '1') for key in self.client.config)
        self.assertNotEqual(self.client.config.all, updated)
        self.client.config.update(updated)

src/q/l/qless-py-HEAD/test/test_config.py   qless-py(Download)
    def test_update(self):
        '''We can use dictionary-style update'''
        updated = dict((key, '1') for key in self.client.config)
        self.assertNotEqual(self.client.config.all, updated)
        self.client.config.update(updated)

src/q/l/qless-py-0.10.1/test/test_queue.py   qless-py(Download)
    def test_heartbeat(self):
        '''Provided access to heartbeat configuration'''
        original = self.client.queues['foo'].heartbeat
        self.client.queues['foo'].heartbeat = 10
        self.assertNotEqual(original, self.client.queues['foo'].heartbeat)

src/q/l/qless-py-HEAD/test/test_queue.py   qless-py(Download)
    def test_heartbeat(self):
        '''Provided access to heartbeat configuration'''
        original = self.client.queues['foo'].heartbeat
        self.client.queues['foo'].heartbeat = 10
        self.assertNotEqual(original, self.client.queues['foo'].heartbeat)