"""
This module provides functions to support hosts-file and remote-command
configuration in the Configuration GUI.
"""
import json
import os
import subprocess
SCHRODINGER = os.environ["SCHRODINGER"]
############################################################################
class _ErrorLevel:
"""
An ErrorLevel is an ordered type that represents the severity
of a host-check error.
"""
def __init__(self, level, tag=""):
if not hasattr(self, "_level"):
self._level = level
self._tag = tag
def __repr__(self):
return self._tag
def __index__(self):
return self._level
OK = _ErrorLevel(0, "OK") # no error
WARNING = _ErrorLevel(1, "WARNING") # an error that may not be serious
ERROR = _ErrorLevel(2, "ERROR") # an error preventing use of the license
############################################################################
ERR_LEVEL = {"ok": OK, "warning": WARNING, "error": ERROR}
[docs]class Error:
"""
An Error represents an system-level issue that prevents remote
commands executed via ssh from succeeding.
"""
[docs] def __init__(self,
message,
level=None,
description="",
advice="",
output=""):
self.message = message
self.level = level
self.description = description
self.advice = advice
self.output = output
[docs]def ssh_check(host, user=""):
"""
Check whether remote commands to the given host succeed or not.
An alternate remote account name can be specified.
Results are returned in a list of Error objects reporting any
serious problems uncovered with remote commands to the specified
host.
The current implementation involves executing a separate test
program to diagnose remote-command issues, remote_command.pl.
"""
errors = []
cmd = [os.path.join(SCHRODINGER, "run"), "remote_command.pl", "-check"]
if user:
cmd.append(f"{user}@{host}")
outfile = f"{host}-{user}-check.json"
else:
cmd.append(host)
outfile = "%s-check.json" % host
cmd.extend(("-output", outfile))
cmdline = subprocess.list2cmdline(cmd)
try:
if os.path.exists(outfile):
os.unlink(outfile)
output = subprocess.check_output(cmd,
stderr=subprocess.STDOUT,
universal_newlines=True)
returncode = 0
except subprocess.CalledProcessError as exc:
output = exc.output
returncode = exc.returncode
except OSError as exc:
output = "\n".join(("% " + cmdline, str(exc)))
try:
with open(outfile, 'rb') as fp:
result = json.load(fp)
except OSError:
if not output:
output = "There was no output. The exit code was %d." % returncode
else:
output = "\n".join(
("The output from the failed program was:", "", output))
output = output.rstrip()
advice = "\n".join([
"The external program used to test ssh commands failed.", output,
"",
"Please contact support at https://www.schrodinger.com/supportcenter"
])
errors.append(
Error("ssh test failed",
level=WARNING,
description=
"The external process used to test ssh commands failed.",
advice=advice))
return errors
for item in result:
if item["severity"] != "ok":
errors.append(
Error(item["message"],
level=ERR_LEVEL.get(item["severity"], "ERROR"),
description=item.get("description", ""),
advice=item.get("advice", "")))
return errors
[docs]def host_check(entry, hostfile="", cwd=None):
"""
Check whether jobs can be run successfully using the given host
entry. A non-default hosts file can be specified; otherwise the
usual hosts file is used.
Results are returned in a list of Error objects reporting any
serious problems uncovered with jobs launched to the specified
host entry.
The current implementation involves executing a separate test
program, "$SCHRODINGER/installation_check", to manage the test
job and diagnose the problem, if it fails.
:param cwd: Use to set the current working directory.
:type cwd: str
"""
errors = []
outfile = "hostcheck-%s.json" % entry
if cwd:
outfile = os.path.join(cwd, outfile)
cmd = [
os.path.join(SCHRODINGER, "installation_check"), "-noverbose",
"-testonly", entry, "-ident", "hostcheck"
]
cmd.extend(("-output", outfile))
if hostfile:
cmd.extend(("-file", hostfile))
cmdline = subprocess.list2cmdline(cmd)
try:
if os.path.exists(outfile):
os.unlink(outfile)
output = subprocess.check_output(cmd,
stderr=subprocess.STDOUT,
cwd=cwd,
universal_newlines=True)
returncode = 0
except subprocess.CalledProcessError as exc:
output = exc.output
returncode = exc.returncode
except OSError as exc:
output = "\n".join(("% " + cmdline, str(exc)))
try:
with open(outfile, 'rb') as fp:
instchek_results = json.load(fp)
except OSError:
if not output:
output = "There was no output. The exit code was %d." % returncode
else:
output = "\n".join(
("The output from the failed program was:", "", output))
output = output.rstrip()
advice = "\n".join([
"The external program used to run test jobs failed.", output, "",
"Please contact support at https://www.schrodinger.com/supportcenter"
])
errors.append(
Error(
"job test failed",
level=WARNING,
description="The external process used to run test jobs failed.",
advice=advice))
return errors
for entry_result in instchek_results:
for item in entry_result["checks"]:
level = ERR_LEVEL.get(item["severity"], "")
if level != OK:
advice = item.get("advice", "")
if advice:
# strip first two lines, which repeat the message.
lines = advice.splitlines()
advice = "\n".join(lines[2:])
errors.append(
Error(item["message"],
level=level,
description=item.get("description", ""),
advice=advice))
return errors