Source code for toil.test.lib.dockerTest

# Copyright (C) 2015-2021 Regents of the University of California
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import os
import signal
import time
import uuid
from threading import Thread

from docker.errors import ContainerError
from toil.common import Toil
from toil.exceptions import FailedJobsException
from toil.job import Job
from toil.lib.docker import (FORGO,
from toil.test import ToilTest, needs_docker, slow

logger = logging.getLogger(__name__)

[docs] @needs_docker class DockerTest(ToilTest): """ Tests dockerCall and ensures no containers are left around. When running tests you may optionally set the TOIL_TEST_TEMP environment variable to the path of a directory where you want temporary test files be placed. The directory will be created if it doesn't exist. The path may be relative in which case it will be assumed to be relative to the project root. If TOIL_TEST_TEMP is not defined, temporary files and directories will be created in the system's default location for such files and any temporary files or directories left over from tests will be removed automatically removed during tear down. Otherwise, left-over files will not be removed. """
[docs] def setUp(self): self.tempDir = self._createTempDir(purpose='tempDir') self.dockerTestLogLevel = 'INFO'
[docs] def testDockerClean(self, caching=False, detached=True, rm=True, deferParam=None): """ Run the test container that creates a file in the work dir, and sleeps for 5 minutes. Ensure that the calling job gets SIGKILLed after a minute, leaving behind the spooky/ghost/zombie container. Ensure that the container is killed on batch system shutdown (through the deferParam mechanism). """ # We need to test the behaviour of `deferParam` with `rm` and # `detached`. We do not look at the case where `rm` and `detached` are # both True. This is the truth table for the different combinations at # the end of the test. R = Running, X = Does not exist, E = Exists but # not running. # None FORGO STOP RM # rm X R X X # detached X R E X # Neither X R E X data_dir = os.path.join(self.tempDir, 'data') working_dir = os.path.join(self.tempDir, 'working') test_file = os.path.join(working_dir, 'test.txt') os.makedirs(data_dir, exist_ok=True) os.makedirs(working_dir, exist_ok=True) options = Job.Runner.getDefaultOptions(os.path.join(self.tempDir, 'jobstore')) options.logLevel = self.dockerTestLogLevel options.workDir = working_dir options.clean = 'always' options.retryCount = 0 # we're expecting the job to fail so don't retry! options.caching = caching # No base64 logic since it might create a name starting with a `-`. container_name = uuid.uuid4().hex A = Job.wrapJobFn(_testDockerCleanFn, working_dir, detached, rm, deferParam, container_name) try: with Toil(options) as toil: toil.start(A) except FailedJobsException: # The file created by spooky_container would remain in the directory # and since it was created inside the container, it would have had # uid and gid == 0 (root) which may cause problems when docker # attempts to clean up the jobstore. file_stats = os.stat(test_file) assert file_stats.st_gid != 0 assert file_stats.st_uid != 0 if (rm and (deferParam != FORGO)) or deferParam == RM or deferParam is None: # These containers should not exist assert containerIsRunning(container_name) is None, \ 'Container was not removed.' elif deferParam == STOP: # These containers should exist but be non-running assert containerIsRunning(container_name) == False, \ 'Container was not stopped.' else: # These containers will be running assert containerIsRunning(container_name) == True, \ 'Container was not running.' finally: # Clean up try: dockerKill(container_name, remove=True) os.remove(test_file) except: pass
[docs] def testDockerClean_CRx_FORGO(self): self.testDockerClean(caching=False, detached=False, rm=True, deferParam=FORGO)
[docs] def testDockerClean_CRx_STOP(self): self.testDockerClean(caching=False, detached=False, rm=True, deferParam=STOP)
[docs] def testDockerClean_CRx_RM(self): self.testDockerClean(caching=False, detached=False, rm=True, deferParam=RM)
[docs] @slow def testDockerClean_CRx_None(self): self.testDockerClean(caching=False, detached=False, rm=True, deferParam=None)
[docs] @slow def testDockerClean_CxD_FORGO(self): self.testDockerClean(caching=False, detached=True, rm=False, deferParam=FORGO)
[docs] @slow def testDockerClean_CxD_STOP(self): self.testDockerClean(caching=False, detached=True, rm=False, deferParam=STOP)
[docs] @slow def testDockerClean_CxD_RM(self): self.testDockerClean(caching=False, detached=True, rm=False, deferParam=RM)
[docs] @slow def testDockerClean_CxD_None(self): self.testDockerClean(caching=False, detached=True, rm=False, deferParam=None)
[docs] @slow def testDockerClean_Cxx_FORGO(self): self.testDockerClean(caching=False, detached=False, rm=False, deferParam=FORGO)
[docs] @slow def testDockerClean_Cxx_STOP(self): self.testDockerClean(caching=False, detached=False, rm=False, deferParam=STOP)
[docs] @slow def testDockerClean_Cxx_RM(self): self.testDockerClean(caching=False, detached=False, rm=False, deferParam=RM)
[docs] @slow def testDockerClean_Cxx_None(self): self.testDockerClean(caching=False, detached=False, rm=False, deferParam=None)
[docs] @slow def testDockerClean_xRx_FORGO(self): self.testDockerClean(caching=True, detached=False, rm=True, deferParam=FORGO)
[docs] @slow def testDockerClean_xRx_STOP(self): self.testDockerClean(caching=True, detached=False, rm=True, deferParam=STOP)
[docs] @slow def testDockerClean_xRx_RM(self): self.testDockerClean(caching=True, detached=False, rm=True, deferParam=RM)
[docs] @slow def testDockerClean_xRx_None(self): self.testDockerClean(caching=True, detached=False, rm=True, deferParam=None)
[docs] @slow def testDockerClean_xxD_FORGO(self): self.testDockerClean(caching=True, detached=True, rm=False, deferParam=FORGO)
[docs] @slow def testDockerClean_xxD_STOP(self): self.testDockerClean(caching=True, detached=True, rm=False, deferParam=STOP)
[docs] @slow def testDockerClean_xxD_RM(self): self.testDockerClean(caching=True, detached=True, rm=False, deferParam=RM)
[docs] @slow def testDockerClean_xxD_None(self): self.testDockerClean(caching=True, detached=True, rm=False, deferParam=None)
[docs] @slow def testDockerClean_xxx_FORGO(self): self.testDockerClean(caching=True, detached=False, rm=False, deferParam=FORGO)
[docs] @slow def testDockerClean_xxx_STOP(self): self.testDockerClean(caching=True, detached=False, rm=False, deferParam=STOP)
[docs] @slow def testDockerClean_xxx_RM(self): self.testDockerClean(caching=True, detached=False, rm=False, deferParam=RM)
[docs] @slow def testDockerClean_xxx_None(self): self.testDockerClean(caching=True, detached=False, rm=False, deferParam=None)
[docs] def testDockerPipeChain(self, caching=False): """ Test for piping API for dockerCall(). Using this API (activated when list of argument lists is given as parameters), commands a piped together into a chain. ex: parameters=[ ['printf', 'x\n y\n'], ['wc', '-l'] ] should execute: printf 'x\n y\n' | wc -l """ options = Job.Runner.getDefaultOptions(os.path.join(self.tempDir, 'jobstore')) options.logLevel = self.dockerTestLogLevel options.workDir = self.tempDir options.clean = 'always' options.caching = caching A = Job.wrapJobFn(_testDockerPipeChainFn) rv = Job.Runner.startToil(A, options)'Container pipeline result: %s', repr(rv)) rv = rv.decode('utf-8') assert rv.strip() == '2'
[docs] def testDockerPipeChainErrorDetection(self, caching=False): """ By default, executing cmd1 | cmd2 | ... | cmdN, will only return an error if cmdN fails. This can lead to all manor of errors being silently missed. This tests to make sure that the piping API for dockerCall() throws an exception if non-last commands in the chain fail. """ options = Job.Runner.getDefaultOptions(os.path.join(self.tempDir, 'jobstore')) options.logLevel = self.dockerTestLogLevel options.workDir = self.tempDir options.clean = 'always' options.caching = caching A = Job.wrapJobFn(_testDockerPipeChainErrorFn) rv = Job.Runner.startToil(A, options) assert rv is True
[docs] def testNonCachingDockerChain(self): self.testDockerPipeChain(caching=True)
[docs] def testNonCachingDockerChainErrorDetection(self): self.testDockerPipeChainErrorDetection(caching=True)
[docs] def testDockerLogs(self, stream=False, demux=False): """Test for the different log outputs when deatch=False.""" working_dir = os.path.join(self.tempDir, 'working') script_file = os.path.join(working_dir, '') os.makedirs(working_dir, exist_ok=True) options = Job.Runner.getDefaultOptions(os.path.join(self.tempDir, 'jobstore')) options.logLevel = self.dockerTestLogLevel options.workDir = working_dir options.clean = 'always' A = Job.wrapJobFn(_testDockerLogsFn, working_dir=working_dir, script_file=script_file, stream=stream, demux=demux) try: rv = Job.Runner.startToil(A, options) assert rv is True finally: try: os.remove(script_file) except: pass
[docs] def testDockerLogs_Stream(self): self.testDockerLogs(stream=True, demux=False)
[docs] def testDockerLogs_Demux(self): self.testDockerLogs(stream=False, demux=True)
[docs] def testDockerLogs_Demux_Stream(self): self.testDockerLogs(stream=True, demux=True)
def _testDockerCleanFn(job, working_dir, detached=None, rm=None, deferParam=None, containerName=None): """ Test function for test docker_clean. Runs a container with given flags and then dies leaving behind a zombie container. :param toil.job.Job job: job :param working_dir: See `work_dir=` in :func:`dockerCall` :param bool rm: See `rm=` in :func:`dockerCall` :param bool detached: See `detached=` in :func:`dockerCall` :param int deferParam: See `deferParam=` in :func:`dockerCall` :param str containerName: See `container_name=` in :func:`dockerCall` """ def killSelf(): test_file = os.path.join(working_dir, 'test.txt') # Kill the worker once we are sure the docker container is started while not os.path.exists(test_file): logger.debug('Waiting on the file created by spooky_container.') time.sleep(1) # By the time we reach here, we are sure the container is running. time.sleep(1) os.kill(os.getpid(), signal.SIGKILL) t = Thread(target=killSelf) # Make it a daemon thread so that thread failure doesn't hang tests. t.daemon = True t.start() apiDockerCall(job, image='', working_dir=working_dir, deferParam=deferParam, containerName=containerName, detach=detached, remove=rm, privileged=True) def _testDockerPipeChainFn(job): """Return the result of a simple pipe chain. Should be 2.""" parameters = [['printf', 'x\n y\n'], ['wc', '-l']] return apiDockerCall(job, image='', parameters=parameters, privileged=True) def _testDockerPipeChainErrorFn(job): """Return True if the command exit 1 | wc -l raises a ContainerError.""" parameters = [['exit', '1'], ['wc', '-l']] try: apiDockerCall(job, image='', parameters=parameters) except ContainerError: return True return False def _testDockerLogsFn(job, working_dir, script_file, stream=False, demux=False): """Return True if the test succeeds. Otherwise Exception is raised.""" # we write a script file because the redirection operator, '>&2', is wrapped # in quotes when passed as parameters. import textwrap bash_script = textwrap.dedent(''' #!/bin/bash echo hello stdout ; echo hello stderr >&2 ; echo hello stdout ; echo hello stderr >&2 ; echo hello stdout ; echo hello stdout ; ''') with open(script_file, 'w') as file: file.write(bash_script) out = apiDockerCall(job, image='', working_dir=working_dir, parameters=[script_file], volumes={working_dir: {'bind': working_dir, 'mode': 'rw'}}, entrypoint="/bin/bash", stdout=True, stderr=True, stream=stream, demux=demux) # we check the output length because order is not guaranteed. if stream: if demux: # a generator with tuples of (stdout, stderr) assert hasattr(out, '__iter__') for _ in range(6): stdout, stderr = next(out) if stdout: # len('hello stdout\n') == 13 assert len(stdout) == 13 elif stderr: assert len(stderr) == 13 else: assert False else: # a generator with bytes assert hasattr(out, '__iter__') for _ in range(6): assert len(next(out)) == 13 else: if demux: # a tuple of (stdout, stderr) stdout, stderr = out # len('hello stdout\n' * 4) == 52 assert len(stdout) == 52 # len('hello stderr\n' * 2) == 26 assert len(stderr) == 26 else: # a bytes object # len('hello stdout\n' * 4 + 'hello stderr\n' * 2) == 78 assert len(out) == 78 return True