redfish-tests/redfish_requests.py

import enum
import math
import re

import requests


class RedfishHttpStatus(enum.IntEnum):
    ok = 200
    created = 201
    no_content = 204
    bad_request = 400
    not_found = 404
    internal_server_error = 500


class RedfishRequest:
    telemetry_service_path = "/redfish/v1/TelemetryService"
    metric_definition_path = f"{telemetry_service_path}/MetricDefinitions"
    metric_report_definition_path = (
        f"{telemetry_service_path}/MetricReportDefinitions"
    )
    metric_report_path = f"{telemetry_service_path}/MetricReports"

    def __init__(self, host_addr, username, password):
        self.host_addr = host_addr
        self.username = username
        self.password = password

    def get(self, path, code=RedfishHttpStatus.ok):
        u = self.host_addr + path
        r = requests.get(u, auth=(self.username, self.password), verify=False)
        assert (
            r.status_code == code
        ), f"{r.status_code} == {code} on path {u}\n{r.text}"
        print(r.text)
        return r.json()

    def post(self, path, body, code=RedfishHttpStatus.created):
        u = self.host_addr + path
        r = requests.post(
            u, auth=(self.username, self.password), verify=False, json=body
        )
        assert (
            r.status_code == code
        ), f"{r.status_code} == {code} on path {u}\n{r.text}"
        print(r.text)
        return r.json()

    def delete(self, path, code=RedfishHttpStatus.no_content):
        u = self.host_addr + path
        r = requests.delete(
            u, auth=(self.username, self.password), verify=False
        )
        assert (
            r.status_code == code
        ), f"{r.status_code} == {code} on path {u}\n{r.text}"


class TelemetryService:
    def __init__(self, redfish, metric_limit):
        r = redfish.get(redfish.telemetry_service_path)
        self.min_interval = Duration.to_seconds(r["MinCollectionInterval"])
        self.max_reports = r["MaxReports"]
        self.metrics = []
        r = redfish.get(redfish.metric_definition_path)
        for m in r["Members"]:
            path = m["@odata.id"]
            metricDef = redfish.get(path)
            self.metrics += [x for x in metricDef["MetricProperties"]]
        self.metrics = self.metrics[:metric_limit]


class ReportDef:
    def __init__(self, redfish):
        self.redfish = redfish

    def get_collection(self):
        r = self.redfish.get(self.redfish.metric_report_definition_path)
        return [x["@odata.id"] for x in r["Members"]]

    def add_report(
        self,
        id,
        metrics=None,
        type="OnRequest",
        actions=None,
        interval=None,
        code=RedfishHttpStatus.created,
    ):
        body = {
            "Id": id,
            "Metrics": [],
            "MetricReportDefinitionType": type,
            "ReportActions": ["RedfishEvent", "LogToMetricReportsCollection"],
        }
        if metrics is not None:
            body["Metrics"] = metrics
        if actions is not None:
            body["ReportActions"] = actions
        if interval is not None:
            body["Schedule"] = {"RecurrenceInterval": interval}
        return self.redfish.post(
            self.redfish.metric_report_definition_path, body, code
        )

    def delete_report(self, path):
        self.redfish.delete(f"{path}")


class Duration:
    def __init__(self):
        pass

    def to_iso8061(time):
        assert time >= 0, "Invalid argument, time is negative"
        days = int(time / (24 * 60 * 60))
        time = math.fmod(time, (24 * 60 * 60))
        hours = int(time / (60 * 60))
        time = math.fmod(time, (60 * 60))
        minutes = int(time / 60)
        time = round(math.fmod(time, 60), 3)
        return f"P{str(days)}DT{str(hours)}H{str(minutes)}M{str(time)}S"

    def to_seconds(duration):
        r = re.fullmatch(
            r"-?P(\d+D)?(T(\d+H)?(\d+M)?(\d+(.\d+)?S)?)?", duration
        )
        assert r, "Invalid argument, not match with regex"
        days = r.group(1)
        hours = r.group(3)
        minutes = r.group(4)
        seconds = r.group(5)
        result = 0
        if days is not None:
            result += int(days[:-1]) * 60 * 60 * 24
        if hours is not None:
            result += int(hours[:-1]) * 60 * 60
        if minutes is not None:
            result += int(minutes[:-1]) * 60
        if seconds is not None:
            result += float(seconds[:-1])
        return result