How to use the kobo.hub.models.TaskLogs function in kobo

def test_get_chunk_small_file(self):
        _, filepath = tempfile.mkstemp(prefix='kobo-test-', suffix='.log', text=True)
        filename = os.path.basename(filepath)
        content = b'This is a log message.'

        with open(filepath, 'wb') as f:
            f.write(content)

        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        s = task_log.get_chunk(filename, 0, 1024)
        os.remove(filepath)

        self.assertEquals(s, content)

def test_tail_small_file(self):
        _, filepath = tempfile.mkstemp(prefix='kobo-test-', suffix='.log', text=True)
        filename = os.path.basename(filepath)
        content = b'This is a log message.'

        with open(filepath, 'wb') as f:
            f.write(content)

        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        s, n = task_log.tail(filename, 1024, 1024)
        os.remove(filepath)

        self.assertEquals(s, content)
        self.assertEquals(n, len(content))

def test_get_item_non_existing_for_unsaved_task(self):
        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        filename = 'some-random-file.log'

        task_log = TaskLogs(task)
        content = task_log[filename]

        self.assertEquals(content, '')
        self.assertFalse(filename in task_log.cache)
        self.assertFalse(filename in task_log.changed)

def test_tail_invalid_file(self):
        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        with self.assertRaises(Exception):
            task_log.tail('invalid.log', 1024, 1024)

def test_tail_gz_file(self):
        _, filepath = tempfile.mkstemp(prefix='kobo-test-', suffix='.log.gz', text=True)
        filename = os.path.basename(filepath)[:-3]
        content = b'This is a log message.'

        f = gzip.open(filepath, mode='wb')
        f.write(content)
        f.close()

        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        s, n = task_log.tail(filename, 1024, 1024)
        os.remove(filepath)

        self.assertEquals(s, content)
        self.assertEquals(n, len(content))

def test_get_chunk_gz_file(self):
        _, filepath = tempfile.mkstemp(prefix='kobo-test-', suffix='.log.gz', text=True)
        filename = os.path.basename(filepath)[:-3]
        content = b'This is a log message.'

        f = gzip.open(filepath, mode='wb')
        f.write(content)
        f.close()

        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        s = task_log.get_chunk(filename, 0, 1024)
        os.remove(filepath)

        self.assertEquals(s, content)

filename = os.path.basename(filepath)
        lines = []

        for i in range(100):
            lines.append(('This is the line %d in the log message.\n' % (i + 1)).encode())

        with open(filepath, 'wb') as f:
            f.writelines(lines)

        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        s = task_log.get_chunk(filename, 0, len(lines[0]) + len(lines[1]))
        os.remove(filepath)

        self.assertEquals(s, lines[0] + lines[1])

filename = os.path.basename(filepath)
        lines = []

        for i in range(1000):
            lines.append(('This is the line %d in the log message.\n' % (i + 1)).encode())

        with open(filepath, 'wb') as f:
            f.writelines(lines)

        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        s, n = task_log.tail(filename, 50, 1024)
        os.remove(filepath)

        self.assertEquals(s, lines[-1])
        self.assertEquals(n, sum([len(line) for line in lines]))

def test_get_chunk_invalid_file(self):
        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        with self.assertRaises(Exception):
            task_log.get_chunk('invalid.log', 0, 1024)

def test_gzip_logs(self):
        task = PropertyMock(
            id=None,
            task_dir=Mock(return_value=tempfile.tempdir),
            spec=['id', 'task_dir'],
        )

        task_log = TaskLogs(task)

        with tempfile.NamedTemporaryFile(suffix='.log', dir=tempfile.tempdir, delete=True) as tf:
            filename = os.path.basename(tf.name)
            task_log[filename] = 'This is a log message.'

            with patch('kobo.hub.models.run') as mock_run:
                task_log.gzip_logs()

                mock_run.assert_called_once_with(
                    'gzip %s' % os.path.join(tempfile.tempdir, filename),
                    can_fail=True,
                    stdout=False
                )