Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added new exception class to reveal more information to the user. Fix… #35

Merged
merged 4 commits into from
Oct 17, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions hpcpy/client/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,6 @@ def status(self, job_id):
job_id : str
Job ID.
"""
# raise NotImplementedError()
cmd = self._tmp_status.format(job_id=job_id)
result = self._shell(cmd)
return result
Expand Down Expand Up @@ -171,11 +170,15 @@ def _shell(self, cmd, decode=True):
Command to run.
decode : bool
Automatically decode response with utf-8, defaults to True
Raises
------
hpcpy.exceptions.ShellException :
When the underlying shell call fails.

Returns
-------
_type_
_description_
str
Result from the underlying called command.
"""
result = shell(cmd)

Expand Down
31 changes: 31 additions & 0 deletions hpcpy/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,36 @@
import subprocess as sp


class NoClientException(Exception):
def __init__(self):
super().__init__(
"Unable to detect scheduler type, cannot determine client type."
)


class ShellException(Exception):
"""Exception class to improve readability of the subprocess.CalledProcessError"""

def __init__(self, called_process_error: sp.CalledProcessError):
"""Constructor

Parameters
----------
called_process_error : sp.CalledProcessError
Source subprocess.CalledError
"""
self.returncode = called_process_error.returncode
self.cmd = called_process_error.cmd[0]
self.stdout = called_process_error.stdout.decode()
self.stderr = called_process_error.stderr.decode()
self.output = called_process_error.output

def __str__(self):
"""Improved string representation of the error message.

Returns
-------
str
Improved error messaging
"""
return f"Error {self.returncode}: {self.stderr}"
17 changes: 13 additions & 4 deletions hpcpy/utilities.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import jinja2.meta as j2m
from pathlib import Path
from importlib import resources
from hpcpy.exceptions import ShellException
import shlex


Expand All @@ -27,11 +28,19 @@ def shell(cmd, check=True, capture_output=True, **kwargs) -> sp.CompletedProcess

Raises
------
subprocess.CalledProcessError
hpcpy.exceptions.ShellException :
When the shell call fails.
"""
return sp.run(
shlex.split(cmd), check=check, capture_output=capture_output, **kwargs
)
try:
return sp.run(
shlex.split(cmd),
shell=True,
check=check,
capture_output=capture_output,
**kwargs,
)
except sp.CalledProcessError as ex:
raise ShellException(ex)


def interpolate_string_template(template, **kwargs) -> str:
Expand Down
11 changes: 11 additions & 0 deletions tests/test_utilities.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,19 @@
"""Tests for utilities.py"""

import hpcpy.utilities as hu
import hpcpy.exceptions as hx


def test_interpolate_string_template():
"""Test interpolating a string template."""
assert hu.interpolate_string_template("hello {{arg}}", arg="world") == "hello world"


def test_shell_exception():
"""Test that error messaging is being raised to the user."""
try:
hu.shell("blah")
except hx.ShellException as ex:

expected = f"Error {ex.returncode}: {ex.stderr}"
assert ex.__str__() == expected
Loading