celery/celery/tests/events/test_cursesmon.py

from __future__ import absolute_import

from nose import SkipTest

from celery.tests.case import AppCase


class MockWindow(object):

    def getmaxyx(self):
        return self.y, self.x


class test_CursesDisplay(AppCase):

    def setup(self):
        try:
            import curses  # noqa
        except ImportError:
            raise SkipTest('curses monitor requires curses')

        from celery.events import cursesmon
        self.monitor = cursesmon.CursesMonitor(object(), app=self.app)
        self.win = MockWindow()
        self.monitor.win = self.win

    def test_format_row_with_default_widths(self):
        self.win.x, self.win.y = 91, 24
        row = self.monitor.format_row(
            '783da208-77d0-40ca-b3d6-37dd6dbb55d3',
            'task.task.task.task.task.task.task.task.task.tas',
            'workerworkerworkerworkerworkerworkerworkerworker',
            '21:13:20',
            'SUCCESS')
        self.assertEqual('783da208-77d0-40ca-b3d6-37dd6dbb55d3 '
                         'workerworker... task.task.[.]tas 21:13:20 SUCCESS ',
                         row)

    def test_format_row_with_truncated_uuid(self):
        self.win.x, self.win.y = 80, 24
        row = self.monitor.format_row(
            '783da208-77d0-40ca-b3d6-37dd6dbb55d3',
            'task.task.task.task.task.task.task.task.task.tas',
            'workerworkerworkerworkerworkerworkerworkerworker',
            '21:13:20',
            'SUCCESS')
        self.assertEqual('783da208-77d0-40ca-b3d... workerworker... '
                         'task.task.[.]tas 21:13:20 SUCCESS ',
                         row)

    def test_format_title_row(self):
        self.win.x, self.win.y = 80, 24
        row = self.monitor.format_row('UUID', 'TASK',
                                      'WORKER', 'TIME', 'STATE')
        self.assertEqual('UUID                      WORKER          '
                         'TASK             TIME     STATE   ',
                         row)

    def test_format_row_for_wide_screen_with_short_uuid(self):
        self.win.x, self.win.y = 140, 24
        row = self.monitor.format_row(
            '783da208-77d0-40ca-b3d6-37dd6dbb55d3',
            'task.task.task.task.task.task.task.task.task.tas',
            'workerworkerworkerworkerworkerworkerworkerworker',
            '21:13:20',
            'SUCCESS')
        self.assertEqual(136, len(row))
        self.assertEqual('783da208-77d0-40ca-b3d6-37dd6dbb55d3 '
                         'workerworkerworkerworkerworkerworker... '
                         'task.task.task.task.task.task.task.[.]tas '
                         '21:13:20 SUCCESS ',
                         row)