/usr/lib/python3/dist-packages/plainbox/impl/test_runner.py is in python3-plainbox 0.5.3-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | # This file is part of Checkbox.
#
# Copyright 2013 Canonical Ltd.
# Written by:
# Sylvain Pineau <sylvain.pineau@canonical.com>
# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
"""
plainbox.impl.test_runner
=========================
Test definitions for plainbox.impl.runner module
"""
from tempfile import TemporaryDirectory
from unittest import TestCase
import os
from plainbox.impl.job import JobDefinition
from plainbox.impl.runner import CommandOutputWriter
from plainbox.impl.runner import FallbackCommandOutputPrinter
from plainbox.impl.runner import IOLogRecordGenerator
from plainbox.impl.runner import JobRunner
from plainbox.impl.runner import slugify
from plainbox.testing_utils.io import TestIO
from plainbox.vendor.mock import Mock, patch
class SlugifyTests(TestCase):
def test_random_strings(self):
self.assertEqual(slugify("A "), "A_")
self.assertEqual(slugify("A-"), "A-")
self.assertEqual(slugify("A_"), "A_")
self.assertEqual(slugify(".b"), ".b")
self.assertEqual(slugify("\z"), "_z")
self.assertEqual(slugify("/z"), "_z")
self.assertEqual(slugify("1k"), "1k")
class IOLogGeneratorTests(TestCase):
def test_smoke(self):
builder = IOLogRecordGenerator()
# Calling on_begin() resets internal state
builder.on_begin(None, None)
builder.on_new_record.connect(
lambda record: setattr(self, 'last_record', record))
# Calling on_line generates records
builder.on_line('stdout', b'text\n')
self.assertEqual(self.last_record.stream_name, 'stdout')
self.assertEqual(self.last_record.data, b'text\n')
builder.on_line('stdout', b'different text\n')
self.assertEqual(self.last_record.stream_name, 'stdout')
self.assertEqual(self.last_record.data, b'different text\n')
builder.on_line('stderr', b'error message\n')
self.assertEqual(self.last_record.stream_name, 'stderr')
self.assertEqual(self.last_record.data, b'error message\n')
class FallbackCommandOutputPrinterTests(TestCase):
def test_smoke(self):
with TestIO(combined=False) as io:
obj = FallbackCommandOutputPrinter("example")
# Whatever gets printed by the job...
obj.on_line('stdout', b'line 1\n')
obj.on_line('stderr', b'line 1\n')
obj.on_line('stdout', b'line 2\n')
obj.on_line('stdout', b'line 3\n')
obj.on_line('stderr', b'line 2\n')
# Gets printed to stdout _only_, stderr is combined with stdout here
self.assertEqual(io.stdout, (
"(job example, <stdout:00001>) line 1\n"
"(job example, <stderr:00001>) line 1\n"
"(job example, <stdout:00002>) line 2\n"
"(job example, <stdout:00003>) line 3\n"
"(job example, <stderr:00002>) line 2\n"
))
class CommandOutputWriterTests(TestCase):
def assertFileContentsEqual(self, pathname, contents):
with open(pathname, 'rb') as stream:
self.assertEqual(stream.read(), contents)
def test_smoke(self):
with TemporaryDirectory() as scratch_dir:
stdout = os.path.join(scratch_dir, "stdout")
stderr = os.path.join(scratch_dir, "stderr")
writer = CommandOutputWriter(stdout, stderr)
# Initially nothing is created
self.assertFalse(os.path.exists(stdout))
self.assertFalse(os.path.exists(stderr))
# Logs are created when the command is first started
writer.on_begin(None, None)
self.assertTrue(os.path.exists(stdout))
self.assertTrue(os.path.exists(stderr))
# Each line simply gets saved
writer.on_line('stdout', b'text\n')
writer.on_line('stderr', b'error\n')
# (but it may not be on disk yet because of buffering)
# After the command is done the logs are left on disk
writer.on_end(None)
self.assertFileContentsEqual(stdout, b'text\n')
self.assertFileContentsEqual(stderr, b'error\n')
|