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.

Sign up for GitHub

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

Handle output from call_cmd better in maas_storage #283

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Handle output from call_cmd better in maas_storage #283

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
25 changes: 17 additions & 8 deletions device-connectors/src/testflinger_device_connectors/devices/maas2/maas_storage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,14 +32,18 @@ class MaasStorageError(ProvisioningError):
class MaasStorage:
"""Maas device connector storage module."""

def __init__(self, maas_user, node_id):
def __init__(self, maas_user, node_id, node_info=None):
self.maas_user = maas_user
self.node_id = node_id
self.device_list = None
self.init_data = None
self.node_info = self._node_read()
self.block_ids = {}
self.partition_sizes = {}
# Better support for testing
if node_info is None:
self.node_info = self._node_read()
else:
self.node_info = node_info
Comment on lines +42 to +46
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can be a one-line assignment, but you can keep it as is since it's a personal preference

Suggested change
# Better support for testing
if node_info is None:
self.node_info = self._node_read()
else:
self.node_info = node_info
# Better support for testing
self.node_info = self._node_read() if node_info is None else node_info


def _logger_debug(self, message):
logger.debug("MAAS: {}".format(message))
Expand Down Expand Up @@ -71,19 +75,24 @@ def call_cmd(cmd, output_json=False):
stderr=subprocess.STDOUT,
check=False,
)
except FileNotFoundError as err:
except (FileNotFoundError, TypeError) as err:
raise MaasStorageError(err) from err

if proc.returncode != 0:
raise MaasStorageError(proc.stdout.decode())
raise MaasStorageError(proc.stdout)

if not proc.stdout:
return {} if output_json else None

if proc.stdout:
output = proc.stdout.decode()
output = proc.stdout.decode()

if output_json:
if output_json:
try:
return json.loads(output)
except json.JSONDecodeError as err:
raise MaasStorageError(output) from err

return output
return output

@staticmethod
def convert_size_to_bytes(size_str):
Expand Down
136 changes: 136 additions & 0 deletions device-connectors/src/testflinger_device_connectors/devices/maas2/tests/test_maas_storage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

import pytest
import json
import subprocess
from unittest.mock import Mock, MagicMock, call
from testflinger_device_connectors.devices.maas2.maas_storage import (
MaasStorage,
Expand Down Expand Up @@ -533,3 +534,138 @@ def test_process_mount(self, maas_storage):
f"mount_point={device['path']}",
]
)


class MockMaasStorageWithCallCmd(MaasStorage):
"""Don't mock call_cmd, must provide node_info."""

def __init__(self, maas_user, node_id, node_info):
super().__init__(maas_user, node_id, node_info)


class TestMaasStorageCallCmd:
"""Test maas device connector storage module."""

node_info = json.dumps(
[
{
"id": 1,
"name": "sda",
"type": "physical",
"size": "300000000000",
"path": "/dev/disk/by-dname/sda",
"filesystem": None,
"partitions": [
{
"id": 10,
"type": "partition",
"size": "1000000000",
"parent_disk": 1,
"bootable": "true",
"filesystem": {
"mount_point": "/boot",
"fstype": "ext4",
},
}
],
},
{
"id": 2,
"name": "sdb",
"type": "physical",
"size": "400000000000",
"path": "/dev/disk/by-dname/sdb",
"filesystem": None,
"partitions": [
{
"id": 20,
"type": "partition",
"size": "20000000000",
"parent_disk": 2,
"filesystem": {
"mount_point": "/data",
"fstype": "ext4",
},
}
],
},
{
"id": 3,
"name": "sdc",
"type": "physical",
"size": "900000000000",
"path": "/dev/disk/by-dname/sdc",
"filesystem": {"mount_point": "/backup", "fstype": "ext4"},
"partitions": [],
},
{
"id": 4,
"type": "virtual",
},
]
)

@pytest.fixture
def maas_storage(self):
"""Provides a MockMaasStorage instance for testing."""
maas_user = "maas_user"
node_id = "node_id"
yield MockMaasStorageWithCallCmd(maas_user, node_id, self.node_info)

def test_call_cmd_type_error(self, maas_storage):
"""Checks if 'call_cmd' raises MaasStorageError for TypeError."""
with pytest.raises(MaasStorageError):
maas_storage.call_cmd(None)

def test_call_cmd_return_nonzero(self, maas_storage):
"""Checks if 'call_cmd' raises MaasStorageError for non-zero return."""
subprocess.run = MagicMock()
subprocess.run.return_value = subprocess.CompletedProcess(
args=["foo"],
returncode=1,
)
with pytest.raises(MaasStorageError):
maas_storage.call_cmd(["foo"])

def test_call_cmd_return_empty_json(self, maas_storage):
"""Checks if 'call_cmd' converts empty stdout to json when needed."""
subprocess.run = MagicMock()
subprocess.run.return_value = subprocess.CompletedProcess(
args=["foo"],
returncode=0,
stdout=b"",
)
assert maas_storage.call_cmd(["foo"], output_json=True) == {}

def test_call_cmd_json_decode_error(self, maas_storage):
"""Checks if 'call_cmd' raises MaasStorageError on JSONDecodeError."""
subprocess.run = MagicMock()
subprocess.run.return_value = subprocess.CompletedProcess(
args=["foo"],
returncode=0,
stdout=b"foo",
)
with pytest.raises(MaasStorageError):
maas_storage.call_cmd(["foo"], output_json=True)

def test_call_cmd_good_text_output(self, maas_storage):
"""Checks if 'call_cmd' returns text output when requested."""
subprocess.run = MagicMock()
subprocess.run.return_value = subprocess.CompletedProcess(
args=["foo"],
returncode=0,
stdout=b"foo",
)
assert maas_storage.call_cmd(["foo"], output_json=False) == "foo"

def test_call_cmd_good_json_output(self, maas_storage):
"""Checks if 'call_cmd' returns json output when requested."""
subprocess.run = MagicMock()
subprocess.run.return_value = subprocess.CompletedProcess(
args=["foo"],
returncode=0,
stdout=b'{"foo": "bar"}',
)
assert maas_storage.call_cmd(["foo"], output_json=True) == {
"foo": "bar"
}