Skip to content

Commit

Permalink
add unit tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@nmgaston
nmgaston committed Dec 4, 2024
1 parent 419381d commit c41b439
Show file tree
Hide file tree
Showing 3 changed files with 280 additions and 21 deletions.
4 changes: 4 additions & 0 deletions inbm/cloudadapter-agent/cloudadapter/cloud/client/inbs_cloud_client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,10 @@ def publish_attribute(self, key: str, value: str) -> None:

logger.debug(f"Received telemetry: key={key}, value={value}")

if self._grpc_channel is None:
raise PublishError("gRPC channel not set up before calling InbsCloudClient.publish_node_update")


static_telemetry=common_pb2.StaticTelemetry(
node_id=self._client_id,
)
Expand Down
275 changes: 271 additions & 4 deletions inbm/cloudadapter-agent/tests/unit/cloud/client/test_inbs_cloud_client.py
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import uuid
import pytest
from mock import MagicMock, Mock, patch
import queue
Expand All @@ -11,6 +12,15 @@
from cloudadapter.pb.common.v1 import common_pb2
from cloudadapter.cloud.client.inbs_cloud_client import InbsCloudClient

# @pytest.fixture
# def client():
# client = InbsCloudClient()
# client._client_id = "test_client_id"
# client._grpc_channel = MagicMock()
# client._metadata = [("key", "value")]
# return client



@pytest.fixture
def inbs_client() -> Generator[InbsCloudClient, None, None]:
Expand All @@ -20,6 +30,7 @@ def inbs_client() -> Generator[InbsCloudClient, None, None]:
tls_enabled = False
tls_cert = None
token = None


with patch(
"cloudadapter.cloud.client.inbs_cloud_client.grpc.insecure_channel"
Expand All @@ -40,6 +51,266 @@ def test_constructor_initializes_values(self, inbs_client: InbsCloudClient) -> N
assert inbs_client._grpc_port == "50051"
assert inbs_client._client_id == "node_id"
assert inbs_client._metadata == [("node-id", "node_id")]

def test_publish_attribute_bios_release_date(self, inbs_client: InbsCloudClient) -> None:
key = "biosReleaseDate"
value = "2021-01-01T00:00:00"

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
bios_release_date=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.bios_release_date == expected_request.static_telemetry.bios_release_date

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_bios_version(self, inbs_client: InbsCloudClient) -> None:
key = "biosVersion"
value = "BNKBL357.86A.0080.2019.0725.1139"

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
bios_version=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.bios_version == expected_request.static_telemetry.bios_version

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_bios_vendor(self, inbs_client: InbsCloudClient) -> None:
key = "biosVendor"
value = "Intel Corporation"

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
bios_vendor=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.bios_vendor == expected_request.static_telemetry.bios_vendor

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_system_product_name(self, inbs_client: InbsCloudClient) -> None:
key = "systemProductName"
value = "ABC"

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
system_product_name=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.system_product_name == expected_request.static_telemetry.system_product_name

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_total_physical_memory(self, inbs_client: InbsCloudClient) -> None:
key = "totalPhysicalMemory"
value = "8203132928"

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
total_physical_memory_bytes=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.total_physical_memory_bytes == expected_request.static_telemetry.total_physical_memory_bytes

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_system_manufacturer(self, inbs_client: InbsCloudClient) -> None:
key = "systemManufacturer"
value = "Intel Corporation"

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
system_manufacturer=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.system_manufacturer == expected_request.static_telemetry.system_manufacturer

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_power_capabilities(self, inbs_client: InbsCloudClient) -> None:
key = "powerCapabilities"
value = '{"shutdown": true, "reboot": true, "suspend": true, "hibernate": true}'

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
power_capabilities=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.power_capabilities == expected_request.static_telemetry.power_capabilities

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_os_info(self, inbs_client: InbsCloudClient) -> None:
key = "osInformation"
value = 'Linux nat2-desktop 6.8.0-49-generic #49~22.04.1-Ubuntu SMP PREEMPT_DYNAMIC Wed Nov 6 17:42:15 UTC 2 x86_64 x86_64'

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
os_information=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.os_information == expected_request.static_telemetry.os_information

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_cpu_id(self, inbs_client: InbsCloudClient) -> None:
key = "cpuId"
value = "Intel(R) Core(TM) i7-7567U CPU @ 3.50GHz"

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
cpu_id=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.cpu_id == expected_request.static_telemetry.cpu_id

assert kwargs['metadata'] == inbs_client._metadata

def test_publish_attribute_os_info(self, inbs_client: InbsCloudClient) -> None:
key = "diskInformation"
value = '[{\"NAME\": \"loop0\", \"SIZE\": \"4096\", \"SSD\": \"True\"}, {\"NAME\": \"loop1\", \"SIZE\": \"58363904\", \"SSD\": \"True\"}, {\"NAME\": \"loop2\", \"SIZE\": \"58052608\", \"SSD\": \"True\"}, {\"NAME\": \"loop3\", \"SIZE\": \"67080192\", \"SSD\": \"True\"}, {\"NAME\": \"loop4\", \"SIZE\": \"66789376\", \"SSD\": \"True\"}, {\"NAME\": \"loop5\", \"SIZE\": \"77463552\", \"SSD\": \"True\"}, {\"NAME\": \"loop6\", \"SIZE\": \"76771328\", \"SSD\": \"True\"}, {\"NAME\": \"loop7\", \"SIZE\": \"46448640\", \"SSD\": \"True\"}, {\"NAME\": \"loop8\", \"SIZE\": \"286236672\", \"SSD\": \"True\"}, {\"NAME\": \"loop9\", \"SIZE\": \"366678016\", \"SSD\": \"True\"}, {\"NAME\": \"loop10\", \"SIZE\": \"366682112\", \"SSD\": \"True\"}, {\"NAME\": \"loop11\", \"SIZE\": \"528642048\", \"SSD\": \"True\"}, {\"NAME\": \"loop12\", \"SIZE\": \"529625088\", \"SSD\": \"True\"}, {\"NAME\": \"loop13\", \"SIZE\": \"96141312\", \"SSD\": \"True\"}, {\"NAME\": \"loop14\", \"SIZE\": \"181411840\", \"SSD\": \"True\"}, {\"NAME\": \"loop15\", \"SIZE\": \"181428224\", \"SSD\": \"True\"}, {\"NAME\": \"loop16\", \"SIZE\": \"790515712\", \"SSD\": \"True\"}, {\"NAME\": \"loop17\", \"SIZE\": \"787800064\", \"SSD\": \"True\"}, {\"NAME\": \"loop18\", \"SIZE\": \"13553664\", \"SSD\": \"True\"}, {\"NAME\": \"loop19\", \"SIZE\": \"12791808\", \"SSD\": \"True\"}, {\"NAME\": \"loop21\", \"SIZE\": \"40714240\", \"SSD\": \"True\"}, {\"NAME\": \"loop22\", \"SIZE\": \"577536\", \"SSD\": \"True\"}, {\"NAME\": \"loop23\", \"SIZE\": \"581632\", \"SSD\": \"True\"}, {\"NAME\": \"loop24\", \"SIZE\": \"25190400\", \"SSD\": \"True\"}, {\"NAME\": \"loop25\", \"SIZE\": \"25354240\", \"SSD\": \"True\"}, {\"NAME\": \"loop26\", \"SIZE\": \"33554432\", \"SSD\": \"True\"}, {\"NAME\": \"loop27\", \"SIZE\": \"287358976\", \"SSD\": \"True\"}, {\"NAME\": \"sda\", \"SIZE\": \"512110190592\", \"SSD\": \"True\"}]'

inbs_client._grpc_channel = MagicMock()
inbs_client.publish_attribute(key, value)

expected_static_telemetry = common_pb2.StaticTelemetry(
node_id=inbs_client._client_id,
disk_information=value
)
expected_request = inbs_sb_pb2.SendNodeUpdateRequest(
request_id=str(uuid.uuid4()),
job_update=None,
static_telemetry=expected_static_telemetry
)

inbs_client._grpc_channel.SendNodeUpdate.assert_called_once()
args, kwargs = inbs_client._grpc_channel.SendNodeUpdate.call_args
actual_request = args[0]

assert actual_request.static_telemetry.node_id == expected_request.static_telemetry.node_id
assert actual_request.static_telemetry.disk_information == expected_request.static_telemetry.disk_information

assert kwargs['metadata'] == inbs_client._metadata

def test_get_client_id(self, inbs_client: InbsCloudClient) -> None:
client_id = inbs_client.get_client_id()
Expand Down Expand Up @@ -83,10 +354,6 @@ def test_publish_event(self, inbs_client: InbsCloudClient) -> None:
# this is not expected to do anything yet
inbs_client.publish_event(key="example_event", value="event_value")

def test_publish_attribute(self, inbs_client: InbsCloudClient) -> None:
# this is not expected to do anything yet
inbs_client.publish_attribute(key="example_attribute", value="attribute_value")

@pytest.mark.parametrize(
"request_id, dispatcher_error_response, command_type, expected_response, expected_xml",
[
Expand Down
22 changes: 5 additions & 17 deletions inbm/telemetry-agent/telemetry/tests/unit/test_power_capabilities.py
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
from telemetry.power_capabilities import get_power_capabilities

# Assuming the get_power_capabilities function is defined in a module named power_capabilities
from telemetry.power_capabilities import get_power_capabilities
from telemetry.power_capabilities import PowerCapabilitiesLinux

def test_get_power_capabilities_all_supported(mocker):
mocker_runner = mocker.patch('inbm_common_lib.shell_runner.PseudoShellRunner.run')
Expand All @@ -10,14 +8,9 @@ def test_get_power_capabilities_all_supported(mocker):
["UNIT FILE STATE VENDOR PRESET\nsuspend.target static\nhibernate.target static\n\n1 unit files listed.", None, 0]
]

expected_result = {
"shutdown": True,
"reboot": True,
"suspend": True,
"hibernate": True
}
expected_result = "{\"shutdown\": true, \"reboot\": true, \"suspend\": true, \"hibernate\": true}"

result = get_power_capabilities()
result = PowerCapabilitiesLinux.get_power_capabilities()
assert result == expected_result

def test_get_power_capabilities_suspend_and_hibernate_not_supported(mocker):
Expand All @@ -27,12 +20,7 @@ def test_get_power_capabilities_suspend_and_hibernate_not_supported(mocker):
["UNIT FILE STATE VENDOR PRESET\nsuspend.target static\n\n0 unit files listed.", None, 0]
]

expected_result = {
"shutdown": True,
"reboot": True,
"suspend": False,
"hibernate": False
}
expected_result = "{\"shutdown\": true, \"reboot\": true, \"suspend\": false, \"hibernate\": false}"

result = get_power_capabilities()
result = PowerCapabilitiesLinux.get_power_capabilities()
assert result == expected_result

0 comments on commit c41b439

Please sign in to comment.