+nondebug_tests = (
+ test_debug_disabled,
+)
+
+
+def start_service(config):
+ # Clean up any previously run daemon which didn't terminate
+ if os.path.exists(pidfile):
+ with open(pidfile) as pidfd:
+ pid = int(pidfd.read())
+ try:
+ # Stop the running service
+ os.kill(pid, 15)
+ except ProcessLookupError:
+ # If there was no process, just remove the stale PID file
+ os.remove(pidfile)
+ # If there's a preexisting hung service, we can't proceed
+ assert not os.path.exists(pidfile)
+
+ # Clean up any previous test output
+ for f in pathlib.Path(".").glob("capture_*.log"):
+ os.remove(f)
+ for d in ("data", "var"):
+ shutil.rmtree(d, ignore_errors=True)
+
+ # Start the service and wait for it to be ready for connections
+ service = subprocess.Popen(("mudpy", config),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
+ return service
+
+
+def stop_service(service):
+ success = True
+
+ # The no-op case when no service was started
+ if service is None:
+ return success
+
+ # This handles when the service is running as a direct child process
+ service.terminate()
+ returncode = service.wait(10)
+ if returncode != 0:
+ tlog("\nERROR: Service exited with code %s." % returncode)
+ success = False
+
+ # This cleans up a daemonized and disassociated service
+ if os.path.exists(pidfile):
+ with open(pidfile) as pidfd:
+ pid = int(pidfd.read())
+ try:
+ # Stop the running service
+ os.kill(pid, 15)
+ time.sleep(1)
+ except ProcessLookupError:
+ # If there was no process, just remove the stale PID file
+ os.remove(pidfile)
+ # The PID file didn't disappear, so we have a hung service
+ if os.path.exists(pidfile):
+ tlog("\nERROR: Hung daemon with PID %s." % pid)
+ success = False
+
+ # Log the contents of stdout and stderr, if any
+ stdout, stderr = service.communicate()
+ tlog("\nRecording stdout as capture_stdout.log.")
+ with open("capture_stdout.log", "w") as serviceout:
+ serviceout.write(stdout.decode("utf-8"))
+ tlog("\nRecording stderr as capture_stderr.log.")
+ with open("capture_stderr.log", "w") as serviceerr:
+ serviceerr.write(stderr.decode("utf-8"))
+
+ # Error if anything was written on stderr as this may indicate ignored
+ # exceptions (e.g. ResourceWarning during garbage collection)
+ if stderr:
+ tlog("\nERROR: something was written to stderr, see "
+ "capture_stderr.log for details.")
+ success = False
+
+ return success
+
+
+def tlog(message, quiet=False):
+ logfile = "capture_tests.log"
+ with open(logfile, "a") as logfd:
+ logfd.write(message + "\n")
+ if not quiet:
+ sys.stdout.write(message)
+ return True
+
+
+def option_callback(telnet_socket, command, option):
+ if option == b'\x7f':
+ # We use this unassigned option value as a canary, so short-circuit
+ # any response to avoid endlessly looping
+ pass
+ elif command in (telnetlib.DO, telnetlib.DONT):
+ telnet_socket.send(telnetlib.IAC + telnetlib.WONT + option)
+ elif command in (telnetlib.WILL, telnetlib.WONT):
+ telnet_socket.send(telnetlib.IAC + telnetlib.DONT + option)
+
+
+def check_debug():
+ if len(sys.argv) > 1:
+ with open(sys.argv[1]) as config_fd:
+ config = yaml.safe_load(config_fd)
+ return config.get(".mudpy.limit.debug", False)
+ return False
+
+
+def connect_client(luser, service):
+ # Try multiple times to connect, with an exponential backoff
+ for retry in range(5):
+ try:
+ # Skipping the retry=0 case gives an immediate first attempt
+ if retry:
+ time.sleep((2 ** retry) / 10)
+ luser.open("::1", 4000)
+ # Attempt to poll the connection, closing if unusable
+ try:
+ luser.fill_rawq()
+ except ConnectionResetError:
+ luser.close()
+ continue
+ # Short-circuit if we get this far, connection is safe to use
+ return luser
+ except ConnectionRefusedError:
+ continue
+ else:
+ # Connection retries have been exhausted, so give up
+ tlog("\nERROR: Client could not connect.\n")
+ stop_service(service)
+ sys.exit(1)
+