Browse code Refactored shell commands and tests into a subpackage. master
1 parent 814d427 commit b55389d4bb10bb5a41a4475de40647e77cbd739f Nigel Stanger authored on 24 Sep 2018

Showing 15 changed files

• shell_command.py
• shell_command/__init__.py
• shell_command/shared_tests.py
• shell_command/shell_command.py
• shell_command/test_convert_command.py
• shell_command/test_ffmpeg_command.py
• shell_command/test_ffmpeg_concat_command.py
• shell_command/test_ffprobe_command.py
• shell_command/test_shell_command.py
• test_convert_command.py
• test_ffmpeg_command.py
• test_ffmpeg_concat_command.py
• test_ffprobe_command.py
• test_shared.py
• test_shell_command.py

Show notes View shell_command.py 100755 → 0

#!/usr/bin/env python3 import datetime import json import tempfile from pathlib import Path import re import shutil import pexpect from progress_bar import (ProgressBar) class ShellCommand(object): """A shell command. _executable contains the full path to the relevant executable. _base_options is a list of standard general options for this command. """ _executable = "" _base_options = [] _expect_patterns = [] @staticmethod def shellquote(s): """Quote a string so it can be safely pasted into the shell.""" # Note: pipes/shlex.quote() only wraps '' around things, # it doesn't do things like \( \), which we also need. # We double-quote everything because that makes it easier # to deal with single quotes. if s: regexes = [ # grouping parentheses for convert re.compile(r"^(\(|\))$"), # double quote (in middle of string) re.compile(r'^(.+".*)$'), # quotes around entire string re.compile(r"""^((?P<quote>['"]).*(?P=quote))$"""), # command line switch (starts with one or two "-") re.compile(r"^(--?.+)$"), # command path (starts with "/") re.compile(r"^(/[\w/]+)$")] substitutions = [r"\\\1", r"'\1'", r"\1", r"\1", r"\1"] num_matches = 0 for subst in zip(regexes, substitutions): (s, n) = subst[0].subn(subst[1], s) num_matches += n if num_matches == 0: # catch-all for non-word characters when none of the other # substitutions apply s = re.compile(r"^(.*\W+.*)$").sub(r'"\1"', s) return s def __init__(self, input_options=[], output_options=[], quiet=False): self.input_options = input_options self.output_options = output_options self.progress = None self.process = None def __repr__(self): return "<{cls}: {cmd}>".format( cls=self.__class__.__name__, cmd=self.command_string(quote=True)) def append_input_options(self, items=[]): """Add a list of items to the end of the input options.""" self.input_options += items def prepend_input_options(self, items=[]): """Add a list of items at the front of the input options.""" self.input_options = items + self.input_options def append_output_options(self, items=[]): """Add a list of items to the end of the output options.""" self.output_options += items def prepend_output_options(self, items=[]): """Add a list of items at the front of the output options.""" self.output_options = items + self.output_options def executable_string(self, quote=False): """Return the executable as a string.""" if quote: return ShellCommand.shellquote(self._executable) else: return self._executable def argument_string(self, quote=False): """Return the list of arguments as a string.""" args = self._base_options + self.input_options + self.output_options if quote: return " ".join([ShellCommand.shellquote(a) for a in args]) else: return " ".join(args) def argument_list(self): """Return a combined list of all arguments.""" return self._base_options + self.input_options + self.output_options def command_string(self, quote=False): """Return the entire command as a string.""" return "{exe} {arg}".format(exe=self.executable_string(quote), arg=self.argument_string(quote)) def process_pattern(self, pat): """Respond to a pexpect pattern. Return True on EOF.""" return (pat == 0) def run(self): """Execute the command in a subprocess.""" self.process = pexpect.spawn(self.executable_string(), self.argument_list()) # EOF is *always* the first pattern. patterns = self.process.compile_pattern_list( [pexpect.EOF] + self._expect_patterns) try: while True: i = self.process.expect_list(patterns, timeout=None) if self.process_pattern(i): break except (KeyboardInterrupt): pass finally: if self.progress: self.progress.finish() self.process.close() return self.process.exitstatus def get_output(self): """Execute the command in a subprocess and return the output.""" return pexpect.run(self.command_string(quote=True)) class ConvertCommand(ShellCommand): """An ImageMagick convert command.""" _executable = shutil.which("convert") _base_options = ["-density", "600", "xc:dimgrey", "null:", # dark grey background "("] def __init__(self, input_options=[], output_options=[], width=2048, height=1536): super().__init__(input_options, output_options) self._base_options = (["-size", "{w}x{h}".format(w=width, h=height)] + self._base_options) self.append_input_options( ["-resize", "{w}x{h}".format(w=width, h=height), "-background", "white", "-alpha", "remove", "-type", "truecolor", # force RGB (this and next line) "-define", "colorspace:auto-grayscale=off"]) self.prepend_output_options([")", "-gravity", "center", "-layers", "composite", "-flatten"]) class FFprobeCommand(ShellCommand): """An ffprobe shell command.""" _executable = shutil.which("ffprobe") _base_options = ["-loglevel", "error", "-show_entries", "format:stream", "-print_format", "json"] def __init__(self, input_options=[], output_options=[]): super().__init__(input_options, output_options) self.entries = None # The input file should be the last input option. assert(Path(self.input_options[-1]).exists()) self.last_modified = Path(self.input_options[-1]).stat().st_mtime def get_entries(self, section="stream", find_list=[]): """Fetch specified attributes from the input file.""" # Re-fetch if the file's changed since we last looked. modified = Path(self.input_options[-1]).stat().st_mtime if (not self.entries) or (modified > self.last_modified): js = json.loads(self.get_output()) self.entries = {"format": js["format"], "stream": js["streams"][0]} return [self.entries[section][f] for f in find_list] class FFmpegCommand(ShellCommand): """A "simple" ffmpeg shell command.""" _executable = shutil.which("ffmpeg") _base_options = ["-y", "-nostdin"] class FFmpegConcatCommand(FFmpegCommand): """An ffmpeg shell command with a complex concat filter.""" _expect_patterns = [r"time=(\d\d):(\d\d):(\d\d\.\d\d)"] def __init__(self, input_options=[], output_options=[], quiet=False, max_progress=100, has_audio=False, has_video=False, process_audio=True, process_video=True, audio_codec="pcm_s16le", video_codec="h264"): super().__init__(input_options, output_options) self.progress = ProgressBar(max_value=max_progress, quiet=quiet) self.has_video = has_video self.process_video = process_video self.video_codec = video_codec if (self.has_video): self.prepend_output_options(["-map", "[vconc]"]) if (self.process_video): self.prepend_output_options(["-pix_fmt", "yuv420p"]) self.prepend_output_options(["-codec:v", self.video_codec]) self.has_audio = has_audio self.process_audio = process_audio self.audio_codec = audio_codec if (self.has_audio): if (self.process_audio): self.prepend_output_options(["-ac", "1", "-map", "[anorm]"]) else: self.prepend_output_options(["-map", "[aconc]"]) self.prepend_output_options(["-codec:a", self.audio_codec]) self.filters = [] def append_filter(self, filter): """Append a filter to the filters list.""" if (filter): self.filters.append(filter) def append_normalisation_filter(self): """Append a normalisation audio filter to the complex filter.""" if (self.has_audio): self.append_filter("[aconc] dynaudnorm=r=0.25:f=10:b=y [anorm]") def append_concat_filter(self, frame_type, segments=[]): """Append a concat filter to the filters list""" # Ignore frame segments. if frame_type in ["a", "v"]: if (len(segments) > 1): self.append_filter( "{inspecs} concat=n={n}:v={v}:a={a} [{t}conc]".format( inspecs=" ".join([s.output_stream_specifier() for s in segments]), n=len(segments), v=int(frame_type == "v"), a=int(frame_type == "a"), t=frame_type)) elif (len(segments) == 1): self.append_filter( "{inspec} {a}null [{t}conc]".format( inspec=segments[0].output_stream_specifier(), a=frame_type if frame_type == "a" else "", t=frame_type)) def build_complex_filter(self): """Build the complete complex filter. Filters in the filtergraph are separated by ";". """ return "{f}".format(f=";".join(self.filters)) def argument_string(self, quote=False): """Return the list of arguments as a string.""" args = (self._base_options + self.input_options + ["-filter_complex", self.build_complex_filter()] + self.output_options) if quote: return " ".join([ShellCommand.shellquote(a) for a in args]) else: return " ".join(args) def argument_list(self): """Return a combined list of all arguments.""" return (self._base_options + self.input_options + ["-filter_complex", self.build_complex_filter()] + self.output_options) def process_pattern(self, pat): """Respond to a pexpect pattern. Return True on EOF.""" if (pat == 1): elapsed = datetime.timedelta( hours=int(self.process.match.group(1)), minutes=int(self.process.match.group(2)), seconds=float(self.process.match.group(3))) self.progress.update(elapsed.total_seconds()) return (pat == 0) if (__name__ == "__main__"): print(ShellCommand()) print(ConvertCommand(input_options=["in.pdf[12]"], output_options=["out.png"])) # FFprobeCommand expects the input file to exist. f = tempfile.NamedTemporaryFile() print(FFprobeCommand(input_options=["-i", f.name])) f.close() print(FFmpegCommand(input_options=["-i", "in.mov"], output_options=["out.mov"])) concat = FFmpegConcatCommand(input_options=["-i", "in.mov"], output_options=["out.mov"], has_audio=True) concat.append_normalisation_filter() print(concat) print("Quoting:") for s in ["foobar", "foo bar baz", "foo(bar)", "[foobar]", "foo[bar", "foo 'bar'", '"foobar"', "'foobar'", 'foo"bar', "foo.bar", "(", ")", "'", "/foo/bar/baz/quux", "-f", "--foobar"]: print(" {s} => echo {sub}".format(s=s, sub=ShellCommand.shellquote(s)))

Ignore Space Show notes View shell_command/__init__.py 0 → 100644
from shell_command.shell_command import ( ShellCommand, ConvertCommand, FFprobeCommand, FFmpegCommand, FFmpegConcatCommand, )

Ignore Space Show notes View shell_command/shared_tests.py 0 → 100644

import unittest from shell_command import ShellCommand class ShellCommandSharedTestCase(unittest.TestCase): """Shared tests for ShellCommand and its subclasses.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set to # [] otherwise they hang around from previous tests. self.command = ShellCommand(input_options=[], output_options=[]) self.expected_executable = "" self.expected_base_options = [] self.expected_input_options = [] self.expected_filter_options = [] self.expected_output_options = [] def tearDown(self): """Clean up after test.""" self.command = None def test_base_options(self): """Test that base options match expected.""" self.assertEqual( self.command._base_options, self.expected_base_options) def test_input_options(self): """Test that input options match expected.""" self.assertEqual( self.command.input_options, self.expected_input_options) def test_output_options(self): """Test that output options match expected.""" self.assertEqual( self.command.output_options, self.expected_output_options) def test_executable_string(self): """Test that executable path matches expected.""" with self.subTest(msg="unquoted paths match"): self.assertEqual( self.command.executable_string(quote=False), self.expected_executable) with self.subTest(msg="unquoted paths match"): # Note: don't explicitly specify quoted value, because # the executable path will vary across different systems. self.assertEqual( self.command.executable_string(quote=True), ShellCommand.shellquote(self.expected_executable)) def test_argument_string(self): """Test that argument string matches expected.""" args = (self.expected_base_options + self.expected_input_options + self.expected_filter_options + self.expected_output_options) with self.subTest(msg="unquoted agruments match"): self.assertEqual( self.command.argument_string(quote=False), " ".join(args)) with self.subTest(msg="quoted agruments match"): self.assertEqual( self.command.argument_string(quote=True), " ".join([ShellCommand.shellquote(a) for a in args])) def test_argument_list(self): """Test that argument list matches expected.""" self.assertEqual(self.command.argument_list(), self.expected_base_options + self.expected_input_options + self.expected_filter_options + self.expected_output_options) def test_command_string(self): """Test that command string matches expected.""" args = (self.expected_base_options + self.expected_input_options + self.expected_filter_options + self.expected_output_options) expected_cmd_unquoted = ( "{exe} {arg}".format(exe=self.expected_executable, arg=" ".join(args)) ) expected_cmd_quoted = ( '{exe} {arg}'.format( exe=ShellCommand.shellquote(self.expected_executable), arg=" ".join([ShellCommand.shellquote(a) for a in args])) ) with self.subTest(msg="unquoted command string matches"): self.assertEqual( self.command.command_string(quote=False), expected_cmd_unquoted) with self.subTest(msg="quoted command string matches"): self.assertEqual( self.command.command_string(quote=True), expected_cmd_quoted)

Ignore Space Show notes View shell_command/shell_command.py 0 → 100755

#!/usr/bin/env python3 import datetime import json import tempfile from pathlib import Path import re import shutil import pexpect from progress_bar import (ProgressBar) class ShellCommand(object): """A shell command. _executable contains the full path to the relevant executable. _base_options is a list of standard general options for this command. """ _executable = "" _base_options = [] _expect_patterns = [] @staticmethod def shellquote(s): """Quote a string so it can be safely pasted into the shell.""" # Note: pipes/shlex.quote() only wraps '' around things, # it doesn't do things like \( \), which we also need. # We double-quote everything because that makes it easier # to deal with single quotes. if s: regexes = [ # grouping parentheses for convert re.compile(r"^(\(|\))$"), # double quote (in middle of string) re.compile(r'^(.+".*)$'), # quotes around entire string re.compile(r"""^((?P<quote>['"]).*(?P=quote))$"""), # command line switch (starts with one or two "-") re.compile(r"^(--?.+)$"), # command path (starts with "/") re.compile(r"^(/[\w/]+)$")] substitutions = [r"\\\1", r"'\1'", r"\1", r"\1", r"\1"] num_matches = 0 for subst in zip(regexes, substitutions): (s, n) = subst[0].subn(subst[1], s) num_matches += n if num_matches == 0: # catch-all for non-word characters when none of the other # substitutions apply s = re.compile(r"^(.*\W+.*)$").sub(r'"\1"', s) return s def __init__(self, input_options=[], output_options=[], quiet=False): self.input_options = input_options self.output_options = output_options self.progress = None self.process = None def __repr__(self): return "<{cls}: {cmd}>".format( cls=self.__class__.__name__, cmd=self.command_string(quote=True)) def append_input_options(self, items=[]): """Add a list of items to the end of the input options.""" self.input_options += items def prepend_input_options(self, items=[]): """Add a list of items at the front of the input options.""" self.input_options = items + self.input_options def append_output_options(self, items=[]): """Add a list of items to the end of the output options.""" self.output_options += items def prepend_output_options(self, items=[]): """Add a list of items at the front of the output options.""" self.output_options = items + self.output_options def executable_string(self, quote=False): """Return the executable as a string.""" if quote: return ShellCommand.shellquote(self._executable) else: return self._executable def argument_string(self, quote=False): """Return the list of arguments as a string.""" args = self._base_options + self.input_options + self.output_options if quote: return " ".join([ShellCommand.shellquote(a) for a in args]) else: return " ".join(args) def argument_list(self): """Return a combined list of all arguments.""" return self._base_options + self.input_options + self.output_options def command_string(self, quote=False): """Return the entire command as a string.""" return "{exe} {arg}".format(exe=self.executable_string(quote), arg=self.argument_string(quote)) def process_pattern(self, pat): """Respond to a pexpect pattern. Return True on EOF.""" return (pat == 0) def run(self): """Execute the command in a subprocess.""" self.process = pexpect.spawn(self.executable_string(), self.argument_list()) # EOF is *always* the first pattern. patterns = self.process.compile_pattern_list( [pexpect.EOF] + self._expect_patterns) try: while True: i = self.process.expect_list(patterns, timeout=None) if self.process_pattern(i): break except (KeyboardInterrupt): pass finally: if self.progress: self.progress.finish() self.process.close() return self.process.exitstatus def get_output(self): """Execute the command in a subprocess and return the output.""" return pexpect.run(self.command_string(quote=True)) class ConvertCommand(ShellCommand): """An ImageMagick convert command.""" _executable = shutil.which("convert") _base_options = ["-density", "600", "xc:dimgrey", "null:", # dark grey background "("] def __init__(self, input_options=[], output_options=[], width=2048, height=1536): super().__init__(input_options, output_options) self._base_options = (["-size", "{w}x{h}".format(w=width, h=height)] + self._base_options) self.append_input_options( ["-resize", "{w}x{h}".format(w=width, h=height), "-background", "white", "-alpha", "remove", "-type", "truecolor", # force RGB (this and next line) "-define", "colorspace:auto-grayscale=off"]) self.prepend_output_options([")", "-gravity", "center", "-layers", "composite", "-flatten"]) class FFprobeCommand(ShellCommand): """An ffprobe shell command.""" _executable = shutil.which("ffprobe") _base_options = ["-loglevel", "error", "-show_entries", "format:stream", "-print_format", "json"] def __init__(self, input_options=[], output_options=[]): super().__init__(input_options, output_options) self.entries = None # The input file should be the last input option. assert(Path(self.input_options[-1]).exists()) self.last_modified = Path(self.input_options[-1]).stat().st_mtime def get_entries(self, section="stream", find_list=[]): """Fetch specified attributes from the input file.""" # Re-fetch if the file's changed since we last looked. modified = Path(self.input_options[-1]).stat().st_mtime if (not self.entries) or (modified > self.last_modified): js = json.loads(self.get_output()) self.entries = {"format": js["format"], "stream": js["streams"][0]} return [self.entries[section][f] for f in find_list] class FFmpegCommand(ShellCommand): """A "simple" ffmpeg shell command.""" _executable = shutil.which("ffmpeg") _base_options = ["-y", "-nostdin"] class FFmpegConcatCommand(FFmpegCommand): """An ffmpeg shell command with a complex concat filter.""" _expect_patterns = [r"time=(\d\d):(\d\d):(\d\d\.\d\d)"] def __init__(self, input_options=[], output_options=[], quiet=False, max_progress=100, has_audio=False, has_video=False, process_audio=True, process_video=True, audio_codec="pcm_s16le", video_codec="h264"): super().__init__(input_options, output_options) self.progress = ProgressBar(max_value=max_progress, quiet=quiet) self.has_video = has_video self.process_video = process_video self.video_codec = video_codec if (self.has_video): self.prepend_output_options(["-map", "[vconc]"]) if (self.process_video): self.prepend_output_options(["-pix_fmt", "yuv420p"]) self.prepend_output_options(["-codec:v", self.video_codec]) self.has_audio = has_audio self.process_audio = process_audio self.audio_codec = audio_codec if (self.has_audio): if (self.process_audio): self.prepend_output_options(["-ac", "1", "-map", "[anorm]"]) else: self.prepend_output_options(["-map", "[aconc]"]) self.prepend_output_options(["-codec:a", self.audio_codec]) self.filters = [] def append_filter(self, filter): """Append a filter to the filters list.""" if (filter): self.filters.append(filter) def append_normalisation_filter(self): """Append a normalisation audio filter to the complex filter.""" if (self.has_audio): self.append_filter("[aconc] dynaudnorm=r=0.25:f=10:b=y [anorm]") def append_concat_filter(self, frame_type, segments=[]): """Append a concat filter to the filters list""" # Ignore frame segments. if frame_type in ["a", "v"]: if (len(segments) > 1): self.append_filter( "{inspecs} concat=n={n}:v={v}:a={a} [{t}conc]".format( inspecs=" ".join([s.output_stream_specifier() for s in segments]), n=len(segments), v=int(frame_type == "v"), a=int(frame_type == "a"), t=frame_type)) elif (len(segments) == 1): self.append_filter( "{inspec} {a}null [{t}conc]".format( inspec=segments[0].output_stream_specifier(), a=frame_type if frame_type == "a" else "", t=frame_type)) def build_complex_filter(self): """Build the complete complex filter. Filters in the filtergraph are separated by ";". """ return "{f}".format(f=";".join(self.filters)) def argument_string(self, quote=False): """Return the list of arguments as a string.""" args = (self._base_options + self.input_options + ["-filter_complex", self.build_complex_filter()] + self.output_options) if quote: return " ".join([ShellCommand.shellquote(a) for a in args]) else: return " ".join(args) def argument_list(self): """Return a combined list of all arguments.""" return (self._base_options + self.input_options + ["-filter_complex", self.build_complex_filter()] + self.output_options) def process_pattern(self, pat): """Respond to a pexpect pattern. Return True on EOF.""" if (pat == 1): elapsed = datetime.timedelta( hours=int(self.process.match.group(1)), minutes=int(self.process.match.group(2)), seconds=float(self.process.match.group(3))) self.progress.update(elapsed.total_seconds()) return (pat == 0) if (__name__ == "__main__"): print(ShellCommand()) print(ConvertCommand(input_options=["in.pdf[12]"], output_options=["out.png"])) # FFprobeCommand expects the input file to exist. f = tempfile.NamedTemporaryFile() print(FFprobeCommand(input_options=["-i", f.name])) f.close() print(FFmpegCommand(input_options=["-i", "in.mov"], output_options=["out.mov"])) concat = FFmpegConcatCommand(input_options=["-i", "in.mov"], output_options=["out.mov"], has_audio=True) concat.append_normalisation_filter() print(concat) print("Quoting:") for s in ["foobar", "foo bar baz", "foo(bar)", "[foobar]", "foo[bar", "foo 'bar'", '"foobar"', "'foobar'", 'foo"bar', "foo.bar", "(", ")", "'", "/foo/bar/baz/quux", "-f", "--foobar"]: print(" {s} => echo {sub}".format(s=s, sub=ShellCommand.shellquote(s)))

Ignore Space Show notes View shell_command/test_convert_command.py 0 → 100644

import shutil import unittest from shell_command import ConvertCommand from shell_command.shared_tests import ShellCommandSharedTestCase class ConvertCommandTestCase(ShellCommandSharedTestCase): """Test the ConvertCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. self.command = ConvertCommand( input_options=["in.pdf[12]"], output_options=["out.png"]) self.expected_executable = shutil.which("convert") self.expected_base_options = [ "-size", "2048x1536", "-density", "600", "xc:dimgrey", "null:", "(" ] self.expected_input_options = [ "in.pdf[12]", "-resize", "2048x1536", "-background", "white", "-alpha", "remove", "-type", "truecolor", "-define", "colorspace:auto-grayscale=off", ] self.expected_filter_options = [] self.expected_output_options = [ ")", "-gravity", "center", "-layers", "composite", "-flatten", "out.png", ] def tearDown(self): """Clean up after test.""" self.command = None # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Ignore Space Show notes View shell_command/test_ffmpeg_command.py 0 → 100644

import shutil from pathlib import Path import tempfile import unittest from shell_command import FFmpegCommand from shell_command.shared_tests import ShellCommandSharedTestCase class FFmpegCommandTestCase(ShellCommandSharedTestCase): """Test the FFmpegCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. self.command = FFmpegCommand( input_options=["-i", "in.mov"], output_options=["out.mov"]) self.expected_executable = shutil.which("ffmpeg") self.expected_base_options = ["-y", "-nostdin",] self.expected_input_options = ["-i", "in.mov"] self.expected_filter_options = [] self.expected_output_options = ["out.mov"] def tearDown(self): """Clean up after test.""" self.command = None # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Ignore Space Show notes View shell_command/test_ffmpeg_concat_command.py 0 → 100644

from datetime import timedelta from pathlib import Path import shutil import tempfile import unittest from unittest.mock import MagicMock from segment import Segment, AudioSegment, VideoSegment from shell_command import FFmpegConcatCommand from shell_command.shared_tests import ShellCommandSharedTestCase class FFmpegConcatCommandTestCase(ShellCommandSharedTestCase): """Test the FFmpegConcatCommand class.""" NORMALISATION_FILTER = "[aconc] dynaudnorm=r=0.25:f=10:b=y [anorm]" # What about testing different permutations of audio and video? def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. self.command = FFmpegConcatCommand( input_options=["-i", "in.mov"], output_options=["out.mov"], has_video=True, has_audio=True, quiet=True) self.expected_executable = shutil.which("ffmpeg") self.expected_base_options = ["-y", "-nostdin",] self.expected_input_options = ["-i", "in.mov"] self.expected_filter_options = ["-filter_complex", ""] self.expected_output_options = [ "-codec:a", self.command.audio_codec, "-ac", "1", "-map", "[anorm]", "-codec:v", self.command.video_codec, "-pix_fmt", "yuv420p", "-map", "[vconc]", "out.mov" ] def tearDown(self): """Clean up after test.""" self.command = None def test_append_filter(self): """Test appending to the filter list.""" test_data = ( ("", [], "append empty filter"), ("normal_filter", ["normal_filter"], "append normal filter"), ) for appended, expected, description in test_data: with self.subTest(msg=description): self.command.append_filter(appended) self.assertEqual(self.command.filters, expected) def test_append_normalisation_filter(self): """Test appending a normalisation filter.""" filters = [self.NORMALISATION_FILTER] self.command.append_normalisation_filter() self.assertEqual(self.command.filters, filters) def test_append_concat_filter(self): """Test appending various concat filters.""" test_data = ( ("a", "audio"), ("v", "video"), ("f", "frame (should be ignored)"), ) expected = [] for frame_type, description in test_data: for num_segments in range(0, 3): segments = None if frame_type == "a": segments = num_segments * [AudioSegment( file="file.in", punch_in=timedelta(), punch_out=timedelta(seconds=20), input_stream=0)] elif frame_type == "v": segments = num_segments * [VideoSegment( file="file.in", punch_in=timedelta(), punch_out=timedelta(seconds=20), input_stream=0)] elif frame_type == "f": # Frame segments should be ignored by # append_concat_filter() regardless. segments = [] else: raise TypeError self.command.append_concat_filter( frame_type=frame_type, segments=segments) if frame_type not in ["a", "v"]: # Ignore frame segments. pass elif num_segments > 1: expected.append( "{inspecs} concat=n={n}:v={v}:a={a} [{t}conc]".format( inspecs=" ".join([s.output_stream_specifier() for s in segments]), n=num_segments, v=int(frame_type == "v"), a=int(frame_type == "a"), t=frame_type) ) elif num_segments == 1: expected.append( "{inspec} {a}null [{t}conc]".format( inspec=segments[0].output_stream_specifier(), a=frame_type if frame_type == "a" else "", t=frame_type) ) with self.subTest( msg="{d}: {n}".format(d=description, n=num_segments)): self.assertEqual(self.command.filters, expected) def test_build_complex_filter(self): self.command.append_normalisation_filter() self.assertEqual( self.command.build_complex_filter(), self.NORMALISATION_FILTER) test_data = ( ("", self.NORMALISATION_FILTER, "append empty filter"), ("normal_filter", "{n};normal_filter".format(n=self.NORMALISATION_FILTER), "append normal filter"), ) for appended, expected, description in test_data: with self.subTest(msg=description): self.command.append_filter(appended) self.assertEqual(self.command.build_complex_filter(), expected) # process_pattern() updates the internal ProgressBar object. # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Ignore Space Show notes View shell_command/test_ffprobe_command.py 0 → 100644

import shutil from pathlib import Path import tempfile import unittest from shell_command import FFprobeCommand from shell_command.shared_tests import ShellCommandSharedTestCase class FFprobeCommandTestCase(ShellCommandSharedTestCase): """Test the FFprobeCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. A fresh # temporary file is created for each test. self.tmpfile = tempfile.NamedTemporaryFile() self.command = FFprobeCommand( input_options=["-i", self.tmpfile.name], output_options=[]) self.expected_executable = shutil.which("ffprobe") self.expected_base_options = [ "-loglevel", "error", "-show_entries", "format:stream", "-print_format", "json", ] self.expected_input_options = ["-i", self.tmpfile.name] self.expected_filter_options = [] self.expected_output_options = [] def tearDown(self): """Clean up after test.""" self.tmpfile.close() self.command = None def test_last_modified(self): """Test that last modified time of temp file matches.""" self.assertEqual( self.command.last_modified, Path(self.tmpfile.name).stat().st_mtime) # get_entries()? # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Ignore Space Show notes View shell_command/test_shell_command.py 0 → 100644

import unittest from shell_command import ShellCommand from shell_command.shared_tests import ShellCommandSharedTestCase class ShellCommandTestCase(ShellCommandSharedTestCase): """Test the ShellCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set to # [] otherwise they hang around from previous tests. self.command = ShellCommand(input_options=[], output_options=[]) self.expected_executable = "" self.expected_base_options = [] self.expected_input_options = [] self.expected_filter_options = [] self.expected_output_options = [] def tearDown(self): """Clean up after test.""" self.command = None def test_shellquote(self): """Test shell quoting (static method).""" test_data = ( (None, None, "None ⇒ None"), ("", "", "(empty string) ⇒ (empty string)"), (" ", '" "', '␢ ⇒ " "'), (" ", '" "', '␢␢␢␢␢ ⇒ " "'), ("foobar", "foobar", "foobar ⇒ foobar"), ("foo bar", '"foo bar"', 'foo bar ⇒ "foo bar"'), ('"foobar"', '\'"foobar"\'', '"foobar" ⇒ \'"foobar"\''), ("'foobar'", "'foobar'", "'foobar' ⇒ 'foobar'"), ("foo 'bar'", '"foo \'bar\'"', "foo 'bar' ⇒ \"foo 'bar'\""), ('foo"bar', '\'foo"bar\'', 'foo"bar ⇒ \'foo"bar\''), ("foo.bar", '"foo.bar"', 'foo.bar ⇒ "foo.bar"'), ("foo(bar)", '"foo(bar)"', 'foo(bar) ⇒ "foo(bar)"'), ("[foobar]", '"[foobar]"', '[foobar] ⇒ "[foobar]"'), ("foo[bar", '"foo[bar"', 'foo[bar ⇒ "foo[bar"'), ("/foo/bar", "/foo/bar", "/foo/bar ⇒ /foo/bar"), ("-f", "-f", "-f ⇒ -f"), ("--foobar", "--foobar", "--foobar ⇒ --foobar"), ("(", r"\(", r"( ⇒ \("), (")", r"\)", r"( ⇒ \)"), ("'", '"\'"', '\' ⇒ "\'"'), ) for original, expected, description in test_data: with self.subTest(msg=description): self.assertEqual(ShellCommand.shellquote(original), expected) def test_append_input_options(self): """Test appending to input options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.input_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["foo", "bar"], '["bar"] ⇒ ["foo", "bar"]'), (["baz", 42], ["foo", "bar", "baz", 42], '["baz", 42] ⇒ ["foo", "bar", "baz", 42]'), ) for appended, expected, description in test_data: with self.subTest(msg="appending {}".format(description)): self.command.append_input_options(appended) self.assertEqual(self.command.input_options, expected) def test_prepend_input_options(self): """Test prepending to input options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.input_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["bar", "foo"], '["bar"] ⇒ ["bar", "foo"]'), (["baz", 42], ["baz", 42, "bar", "foo"], '["baz", 42] ⇒ ["baz", 42, "bar", "foo"]'), ) for prepended, expected, description in test_data: with self.subTest(msg="prepending {}".format(description)): self.command.prepend_input_options(prepended) self.assertEqual(self.command.input_options, expected) def test_append_output_options(self): """Test appending to output options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.output_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["foo", "bar"], '["bar"] ⇒ ["foo", "bar"]'), (["baz", 42], ["foo", "bar", "baz", 42], '["baz", 42] ⇒ ["foo", "bar", "baz", 42]'), ) for appended, expected, description in test_data: with self.subTest(msg="appending {}".format(description)): self.command.append_output_options(appended) self.assertEqual(self.command.output_options, expected) def test_prepend_output_options(self): """Test prepending to output options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.output_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["bar", "foo"], '["bar"] ⇒ ["bar", "foo"]'), (["baz", 42], ["baz", 42, "bar", "foo"], '["baz", 42] ⇒ ["baz", 42, "bar", "foo"]'), ) for prepended, expected, description in test_data: with self.subTest(msg="prepending {}".format(description)): self.command.prepend_output_options(prepended) self.assertEqual(self.command.output_options, expected) def test_process_pattern(self): """ Test pattern processing.""" # True on EOF (0) with self.subTest(msg="returns True on EOF"): self.assertTrue(self.command.process_pattern(0)) # False on anything else for i in (x for x in range(-1000, 1001) if x is not 0): with self.subTest(msg="returns False on {}".format(i)): self.assertFalse(self.command.process_pattern(i)) # The following two will require mocking of pexpect? def test_run(self): """Test running of subprocess.""" pass def test_get_output(self): """Test getting output from subprocess.""" pass # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Show notes View test_convert_command.py 100644 → 0

import shutil import unittest from shell_command import ConvertCommand from test_shared import ShellCommandSharedTestCase class ConvertCommandTestCase(ShellCommandSharedTestCase): """Test the ConvertCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. self.command = ConvertCommand( input_options=["in.pdf[12]"], output_options=["out.png"]) self.expected_executable = shutil.which("convert") self.expected_base_options = [ "-size", "2048x1536", "-density", "600", "xc:dimgrey", "null:", "(" ] self.expected_input_options = [ "in.pdf[12]", "-resize", "2048x1536", "-background", "white", "-alpha", "remove", "-type", "truecolor", "-define", "colorspace:auto-grayscale=off", ] self.expected_filter_options = [] self.expected_output_options = [ ")", "-gravity", "center", "-layers", "composite", "-flatten", "out.png", ] def tearDown(self): """Clean up after test.""" self.command = None # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Show notes View test_ffmpeg_command.py 100644 → 0

import shutil from pathlib import Path import tempfile import unittest from shell_command import FFmpegCommand from test_shared import ShellCommandSharedTestCase class FFmpegCommandTestCase(ShellCommandSharedTestCase): """Test the FFmpegCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. self.command = FFmpegCommand( input_options=["-i", "in.mov"], output_options=["out.mov"]) self.expected_executable = shutil.which("ffmpeg") self.expected_base_options = ["-y", "-nostdin",] self.expected_input_options = ["-i", "in.mov"] self.expected_filter_options = [] self.expected_output_options = ["out.mov"] def tearDown(self): """Clean up after test.""" self.command = None # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Show notes View test_ffmpeg_concat_command.py 100644 → 0

from datetime import timedelta from pathlib import Path import shutil import tempfile import unittest from unittest.mock import MagicMock from segment import Segment, AudioSegment, VideoSegment from shell_command import FFmpegConcatCommand from test_shared import ShellCommandSharedTestCase class FFmpegConcatCommandTestCase(ShellCommandSharedTestCase): """Test the FFmpegConcatCommand class.""" NORMALISATION_FILTER = "[aconc] dynaudnorm=r=0.25:f=10:b=y [anorm]" # What about testing different permutations of audio and video? def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. self.command = FFmpegConcatCommand( input_options=["-i", "in.mov"], output_options=["out.mov"], has_video=True, has_audio=True, quiet=True) self.expected_executable = shutil.which("ffmpeg") self.expected_base_options = ["-y", "-nostdin",] self.expected_input_options = ["-i", "in.mov"] self.expected_filter_options = ["-filter_complex", ""] self.expected_output_options = [ "-codec:a", self.command.audio_codec, "-ac", "1", "-map", "[anorm]", "-codec:v", self.command.video_codec, "-pix_fmt", "yuv420p", "-map", "[vconc]", "out.mov" ] def tearDown(self): """Clean up after test.""" self.command = None def test_append_filter(self): """Test appending to the filter list.""" test_data = ( ("", [], "append empty filter"), ("normal_filter", ["normal_filter"], "append normal filter"), ) for appended, expected, description in test_data: with self.subTest(msg=description): self.command.append_filter(appended) self.assertEqual(self.command.filters, expected) def test_append_normalisation_filter(self): """Test appending a normalisation filter.""" filters = [self.NORMALISATION_FILTER] self.command.append_normalisation_filter() self.assertEqual(self.command.filters, filters) def test_append_concat_filter(self): """Test appending various concat filters.""" test_data = ( ("a", "audio"), ("v", "video"), ("f", "frame (should be ignored)"), ) expected = [] for frame_type, description in test_data: for num_segments in range(0, 3): segments = None if frame_type == "a": segments = num_segments * [AudioSegment( file="file.in", punch_in=timedelta(), punch_out=timedelta(seconds=20), input_stream=0)] elif frame_type == "v": segments = num_segments * [VideoSegment( file="file.in", punch_in=timedelta(), punch_out=timedelta(seconds=20), input_stream=0)] elif frame_type == "f": # Frame segments should be ignored by # append_concat_filter() regardless. segments = [] else: raise TypeError self.command.append_concat_filter( frame_type=frame_type, segments=segments) if frame_type not in ["a", "v"]: # Ignore frame segments. pass elif num_segments > 1: expected.append( "{inspecs} concat=n={n}:v={v}:a={a} [{t}conc]".format( inspecs=" ".join([s.output_stream_specifier() for s in segments]), n=num_segments, v=int(frame_type == "v"), a=int(frame_type == "a"), t=frame_type) ) elif num_segments == 1: expected.append( "{inspec} {a}null [{t}conc]".format( inspec=segments[0].output_stream_specifier(), a=frame_type if frame_type == "a" else "", t=frame_type) ) with self.subTest( msg="{d}: {n}".format(d=description, n=num_segments)): self.assertEqual(self.command.filters, expected) def test_build_complex_filter(self): self.command.append_normalisation_filter() self.assertEqual( self.command.build_complex_filter(), self.NORMALISATION_FILTER) test_data = ( ("", self.NORMALISATION_FILTER, "append empty filter"), ("normal_filter", "{n};normal_filter".format(n=self.NORMALISATION_FILTER), "append normal filter"), ) for appended, expected, description in test_data: with self.subTest(msg=description): self.command.append_filter(appended) self.assertEqual(self.command.build_complex_filter(), expected) # process_pattern() updates the internal ProgressBar object. # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Show notes View test_ffprobe_command.py 100644 → 0

import shutil from pathlib import Path import tempfile import unittest from shell_command import FFprobeCommand from test_shared import ShellCommandSharedTestCase class FFprobeCommandTestCase(ShellCommandSharedTestCase): """Test the FFprobeCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set, # otherwise they hang around from previous tests. A fresh # temporary file is created for each test. self.tmpfile = tempfile.NamedTemporaryFile() self.command = FFprobeCommand( input_options=["-i", self.tmpfile.name], output_options=[]) self.expected_executable = shutil.which("ffprobe") self.expected_base_options = [ "-loglevel", "error", "-show_entries", "format:stream", "-print_format", "json", ] self.expected_input_options = ["-i", self.tmpfile.name] self.expected_filter_options = [] self.expected_output_options = [] def tearDown(self): """Clean up after test.""" self.tmpfile.close() self.command = None def test_last_modified(self): """Test that last modified time of temp file matches.""" self.assertEqual( self.command.last_modified, Path(self.tmpfile.name).stat().st_mtime) # get_entries()? # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Show notes View test_shared.py 100644 → 0

import unittest from shell_command import ShellCommand class ShellCommandSharedTestCase(unittest.TestCase): """Shared tests for ShellCommand and its subclasses.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set to # [] otherwise they hang around from previous tests. self.command = ShellCommand(input_options=[], output_options=[]) self.expected_executable = "" self.expected_base_options = [] self.expected_input_options = [] self.expected_filter_options = [] self.expected_output_options = [] def tearDown(self): """Clean up after test.""" self.command = None def test_base_options(self): """Test that base options match expected.""" self.assertEqual( self.command._base_options, self.expected_base_options) def test_input_options(self): """Test that input options match expected.""" self.assertEqual( self.command.input_options, self.expected_input_options) def test_output_options(self): """Test that output options match expected.""" self.assertEqual( self.command.output_options, self.expected_output_options) def test_executable_string(self): """Test that executable path matches expected.""" with self.subTest(msg="unquoted paths match"): self.assertEqual( self.command.executable_string(quote=False), self.expected_executable) with self.subTest(msg="unquoted paths match"): # Note: don't explicitly specify quoted value, because # the executable path will vary across different systems. self.assertEqual( self.command.executable_string(quote=True), ShellCommand.shellquote(self.expected_executable)) def test_argument_string(self): """Test that argument string matches expected.""" args = (self.expected_base_options + self.expected_input_options + self.expected_filter_options + self.expected_output_options) with self.subTest(msg="unquoted agruments match"): self.assertEqual( self.command.argument_string(quote=False), " ".join(args)) with self.subTest(msg="quoted agruments match"): self.assertEqual( self.command.argument_string(quote=True), " ".join([ShellCommand.shellquote(a) for a in args])) def test_argument_list(self): """Test that argument list matches expected.""" self.assertEqual(self.command.argument_list(), self.expected_base_options + self.expected_input_options + self.expected_filter_options + self.expected_output_options) def test_command_string(self): """Test that command string matches expected.""" args = (self.expected_base_options + self.expected_input_options + self.expected_filter_options + self.expected_output_options) expected_cmd_unquoted = ( "{exe} {arg}".format(exe=self.expected_executable, arg=" ".join(args)) ) expected_cmd_quoted = ( '{exe} {arg}'.format( exe=ShellCommand.shellquote(self.expected_executable), arg=" ".join([ShellCommand.shellquote(a) for a in args])) ) with self.subTest(msg="unquoted command string matches"): self.assertEqual( self.command.command_string(quote=False), expected_cmd_unquoted) with self.subTest(msg="quoted command string matches"): self.assertEqual( self.command.command_string(quote=True), expected_cmd_quoted)

Show notes View test_shell_command.py 100644 → 0

import unittest from shell_command import ShellCommand from test_shared import ShellCommandSharedTestCase class ShellCommandTestCase(ShellCommandSharedTestCase): """Test the ShellCommand class.""" def setUp(self): """Set up for test.""" # Make sure the input and output options are explicitly set to # [] otherwise they hang around from previous tests. self.command = ShellCommand(input_options=[], output_options=[]) self.expected_executable = "" self.expected_base_options = [] self.expected_input_options = [] self.expected_filter_options = [] self.expected_output_options = [] def tearDown(self): """Clean up after test.""" self.command = None def test_shellquote(self): """Test shell quoting (static method).""" test_data = ( (None, None, "None ⇒ None"), ("", "", "(empty string) ⇒ (empty string)"), (" ", '" "', '␢ ⇒ " "'), (" ", '" "', '␢␢␢␢␢ ⇒ " "'), ("foobar", "foobar", "foobar ⇒ foobar"), ("foo bar", '"foo bar"', 'foo bar ⇒ "foo bar"'), ('"foobar"', '\'"foobar"\'', '"foobar" ⇒ \'"foobar"\''), ("'foobar'", "'foobar'", "'foobar' ⇒ 'foobar'"), ("foo 'bar'", '"foo \'bar\'"', "foo 'bar' ⇒ \"foo 'bar'\""), ('foo"bar', '\'foo"bar\'', 'foo"bar ⇒ \'foo"bar\''), ("foo.bar", '"foo.bar"', 'foo.bar ⇒ "foo.bar"'), ("foo(bar)", '"foo(bar)"', 'foo(bar) ⇒ "foo(bar)"'), ("[foobar]", '"[foobar]"', '[foobar] ⇒ "[foobar]"'), ("foo[bar", '"foo[bar"', 'foo[bar ⇒ "foo[bar"'), ("/foo/bar", "/foo/bar", "/foo/bar ⇒ /foo/bar"), ("-f", "-f", "-f ⇒ -f"), ("--foobar", "--foobar", "--foobar ⇒ --foobar"), ("(", r"\(", r"( ⇒ \("), (")", r"\)", r"( ⇒ \)"), ("'", '"\'"', '\' ⇒ "\'"'), ) for original, expected, description in test_data: with self.subTest(msg=description): self.assertEqual(ShellCommand.shellquote(original), expected) def test_append_input_options(self): """Test appending to input options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.input_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["foo", "bar"], '["bar"] ⇒ ["foo", "bar"]'), (["baz", 42], ["foo", "bar", "baz", 42], '["baz", 42] ⇒ ["foo", "bar", "baz", 42]'), ) for appended, expected, description in test_data: with self.subTest(msg="appending {}".format(description)): self.command.append_input_options(appended) self.assertEqual(self.command.input_options, expected) def test_prepend_input_options(self): """Test prepending to input options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.input_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["bar", "foo"], '["bar"] ⇒ ["bar", "foo"]'), (["baz", 42], ["baz", 42, "bar", "foo"], '["baz", 42] ⇒ ["baz", 42, "bar", "foo"]'), ) for prepended, expected, description in test_data: with self.subTest(msg="prepending {}".format(description)): self.command.prepend_input_options(prepended) self.assertEqual(self.command.input_options, expected) def test_append_output_options(self): """Test appending to output options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.output_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["foo", "bar"], '["bar"] ⇒ ["foo", "bar"]'), (["baz", 42], ["foo", "bar", "baz", 42], '["baz", 42] ⇒ ["foo", "bar", "baz", 42]'), ) for appended, expected, description in test_data: with self.subTest(msg="appending {}".format(description)): self.command.append_output_options(appended) self.assertEqual(self.command.output_options, expected) def test_prepend_output_options(self): """Test prepending to output options.""" with self.subTest(msg="should initially be []"): self.assertEqual(self.command.output_options, []) test_data = ( ([], [], "[] ⇒ []"), (["foo"], ["foo"], '["foo"] ⇒ ["foo"]'), (["bar"], ["bar", "foo"], '["bar"] ⇒ ["bar", "foo"]'), (["baz", 42], ["baz", 42, "bar", "foo"], '["baz", 42] ⇒ ["baz", 42, "bar", "foo"]'), ) for prepended, expected, description in test_data: with self.subTest(msg="prepending {}".format(description)): self.command.prepend_output_options(prepended) self.assertEqual(self.command.output_options, expected) def test_process_pattern(self): """ Test pattern processing.""" # True on EOF (0) with self.subTest(msg="returns True on EOF"): self.assertTrue(self.command.process_pattern(0)) # False on anything else for i in (x for x in range(-1000, 1001) if x is not 0): with self.subTest(msg="returns False on {}".format(i)): self.assertFalse(self.command.process_pattern(i)) # The following two will require mocking of pexpect? def test_run(self): """Test running of subprocess.""" pass def test_get_output(self): """Test getting output from subprocess.""" pass # Remove ShellCommandSharedTestCase from the namespace so we don't run # the shared tests twice. See <https://stackoverflow.com/a/22836015>. del(ShellCommandSharedTestCase)

Show line notes below