Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_visualize_raises_informative_error_without_sys_graphviz(self, monkeypatch):
f = Flow(name="test")
f.add_task(Task())
import graphviz as gviz
err = gviz.backend.ExecutableNotFound
graphviz = MagicMock(
Digraph=lambda: MagicMock(
render=MagicMock(side_effect=err("Can't find dot!"))
)
)
graphviz.backend.ExecutableNotFound = err
with patch.dict("sys.modules", graphviz=graphviz):
with pytest.raises(err, match="Please install Graphviz"):
f.visualize()
def test_missing_executable(func, args):
with pytest.raises(ExecutableNotFound, match=r'execute'):
func(*args)
def execute(self):
"""Execute the network through the fastr.network.execute command."""
# Draw and execute nwtwork
try:
self.network.draw(file_path=self.network.id + '.svg', draw_dimensions=True)
except graphviz.backend.ExecutableNotFound:
print('[WORC WARNING] Graphviz executable not found: not drawing network diagram. Make sure the Graphviz executables are on your systems PATH.')
if DebugDetector().do_detection():
print("Source Data:")
for k in self.source_data.keys():
print(f"\t {k}: {self.source_data[k]}.")
print("\n Sink Data:")
for k in self.sink_data.keys():
print(f"\t {k}: {self.sink_data[k]}.")
# When debugging, set the tempdir to the default of fastr + name
self.fastr_tmpdir = os.path.join(fastr.config.mounts['tmp'],
self.name)
self.network.execute(self.source_data, self.sink_data, execution_plugin=self.fastr_plugin, tmpdir=self.fastr_tmpdir)
label += column
label += ' '
label += ' >'
graph.node(name=node_name, label=label,
_attributes={'fontname': 'Helvetica, Arial, sans-serif', 'fontsize': '10',
'fontcolor': '#555555', 'shape': 'none'})
for (schema_name, table_name), (referenced_schema_name, referenced_table_name) in fk_constraints:
graph.edge(schema_name + '.' + table_name, referenced_schema_name + '.' + referenced_table_name,
_attributes={'color': '#888888'})
try:
svg = graph.pipe('svg').decode('utf-8')
except graphviz.backend.ExecutableNotFound as e:
return str(_.tt(style='color:red')[str(e)])
response = flask.Response(svg)
response.headers[
'Content-Disposition'] = f'attachment; filename="{datetime.date.today().isoformat()}-{db_alias}.svg"'
return response
"""Run the command described by cmd and return its (stdout, stderr) tuple."""
log.debug('run %r', cmd)
if input is not None:
kwargs['stdin'] = subprocess.PIPE
if encoding is not None:
input = input.encode(encoding)
if capture_output:
kwargs['stdout'] = kwargs['stderr'] = subprocess.PIPE
try:
proc = subprocess.Popen(cmd, startupinfo=get_startupinfo(), **kwargs)
except OSError as e:
if e.errno == errno.ENOENT:
raise ExecutableNotFound(cmd)
else:
raise
out, err = proc.communicate(input)
if not quiet and err:
_compat.stderr_write_bytes(err, flush=True)
if encoding is not None:
if out is not None:
out = out.decode(encoding)
if err is not None:
err = err.decode(encoding)
if check and proc.returncode:
raise CalledProcessError(proc.returncode, cmd,
def __init__(self, args):
super(ExecutableNotFound, self).__init__(self._msg % args)
def make_report(self):
""" Makes a html report and saves it into the same folder where logs are stored.
"""
if self.eoexecutor.execution_stats is None:
raise RuntimeError('Cannot produce a report without running the executor first, check EOExecutor.run '
'method')
if os.environ.get('DISPLAY', '') == '':
plt.switch_backend('Agg')
try:
dependency_graph = self._create_dependency_graph()
except graphviz.backend.ExecutableNotFound as ex:
dependency_graph = None
warnings.warn("{}.\nPlease install the system package 'graphviz' (in addition "
"to the python package) to have the dependency graph in the final report!".format(ex),
Warning, stacklevel=2)
task_descriptions = self._get_task_descriptions()
formatter = HtmlFormatter(linenos=True)
task_source = self._render_task_source(formatter)
execution_stats = self._render_execution_errors(formatter)
template = self._get_template()
html = template.render(dependency_graph=dependency_graph,
task_descriptions=task_descriptions,
task_source=task_source,
def execute(self):
"""Execute the network through the fastr.network.execute command."""
# Draw and execute nwtwork
try:
self.network.draw(file_path=self.network.id + '.svg',
draw_dimensions=True)
except graphviz.backend.ExecutableNotFound:
print('[WORC WARNING] Graphviz executable not found: not drawing network diagram. MAke sure the Graphviz executables are on your systems PATH.')
self.network.execute(self.source_data, self.sink_data,
execution_plugin=self.fastr_plugin,
tmpdir=self.fastr_tmpdir)
if edgeKey in edges:
return
edges.add(edgeKey)
edgeType = networkEdgeTypes[diagramNetwork]["all-types"]
mdl.edge(parentDocName, docName,
dir=edgeType.get("dir"), arrowhead=edgeType.get("arrowhead"), arrowtail=edgeType.get("arrowtail"))
for referencedDoc in modelDoc.referencesDocument.keys():
if referencedDoc.basename != parentDocName: # skip reverse linkbase ref
viewDtsDoc(referencedDoc, docName, parentDocName)
viewDtsDoc(modelXbrl.modelDocument, None, None)
mdl.format = "pdf"
try:
mdl.render(diagramFile.replace(".pdf", ".gv"), view=viewDiagram)
except backend.ExecutableNotFound as ex:
modelXbrl.warning("objectmaker:graphvizExecutable",
"Diagram saving requires installation of graphviz, error: %(error)s:",
modelXbrl=modelXbrl, error=ex)
style=style,
)
if filename:
graph.render(filename, view=False)
else:
try:
from IPython import get_ipython
assert get_ipython().config.get("IPKernelApp") is not None
except Exception:
with tempfile.NamedTemporaryFile(delete=False) as tmp:
tmp.close()
try:
graph.render(tmp.name, view=True)
except graphviz.backend.ExecutableNotFound:
msg = "It appears you do not have Graphviz installed, or it is not on your PATH.\n"
msg += "Please install Graphviz from http://www.graphviz.org/download/\n"
msg += "And note: just installing the `graphviz` python package is not sufficient!"
raise graphviz.backend.ExecutableNotFound(msg)
finally:
os.unlink(tmp.name)
return graph