Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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
)