/usr/lib/python3/dist-packages/systemfixtures/executable.py is in python3-systemfixtures 0.6.4-1.
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 121 122 123 124 125 126 127 | import os
import socket
import six
from testtools.content import Content
from testtools.content_type import UTF8_TEXT
from fixtures import (
Fixture,
TempDir,
)
if six.PY2:
import subprocess32 as subprocess
if six.PY3:
import subprocess
class FakeExecutable(Fixture):
"""Create Python scripts that mimic the behavior of real executables."""
def _setUp(self):
self.path = self.useFixture(TempDir()).join("executable")
self.line("#!/usr/bin/env python")
self.line("import logging")
self.line("logging.basicConfig("
"format='%(asctime)s %(message)s', level=logging.DEBUG)")
os.chmod(self.path, 0o0755)
self._process = None
self.addDetail("fake-process", Content(UTF8_TEXT, self._process_info))
def spawn(self):
"""Spawn the fake executable using subprocess.Popen."""
self._process = subprocess.Popen(
[self.path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
self.addCleanup(self._process_kill)
def out(self, text):
self.line("import sys")
self.line("sys.stdout.write('{}\\n')".format(text))
self.line("sys.stdout.flush()")
def log(self, message):
self.line("logging.info('{}')".format(message))
def sleep(self, seconds):
self.line("import time")
self.line("time.sleep({})".format(seconds))
def hang(self):
self.line("import time")
self.line("import signal")
self.line("signal.signal(signal.SIGTERM, lambda *args: None)")
self.out("hanging")
self.line("while True: time.sleep(1)")
def listen(self, port=None):
"""Make the fake executable listen to the specified port.
Possible values for 'port' are:
- None: Allocate immediately a free port and instruct the fake
executable to use it when it's invoked. This is subject to
a race condition, if that port that was free when listen() was
invoked later becomes used before the fake executable had chance
to bind to it. However it has the advantage of exposing the
free port as FakeExecutable.port instance variable, that can easily
be consumed by tests.
- An integer: Listen to this specific port.
"""
if port is None:
port = allocate_port()
self.port = port
self.line("import socket")
self.line("sock = socket.socket()")
self.line("sock.bind(('localhost', {}))".format(self.port))
self.log("listening: %d" % self.port)
self.line("sock.listen(0)")
def line(self, line):
with open(self.path, "a") as fd:
fd.write("{}\n".format(line))
def _process_kill(self):
"""Kill the fake executable process if it's still running."""
if self._process.poll() is None: # pragma: no cover
self._process.kill()
self._process.wait(timeout=5)
def _process_info(self):
"""Return details about the fake process."""
if not self._process:
return []
output, error = self._process.communicate(timeout=5)
if error is None:
error = b""
output = output.decode("utf-8").strip()
error = error.decode("utf-8").strip()
info = (u"returncode: %r\n"
u"output:\n%s\n"
u"error:\n%s\n" % (self._process.returncode, output, error))
return [info.encode("utf-8")]
def get_port(socket):
"""Return the port to which a socket is bound."""
addr, port = socket.getsockname()
return port
def allocate_port():
"""Allocate an unused port.
There is a small race condition here (between the time we allocate the
port, and the time it actually gets used), but for the purposes for which
this function gets used it isn't a problem in practice.
"""
sock = socket.socket()
try:
sock.bind(("localhost", 0))
return get_port(sock)
finally:
sock.close()
|