Christian Manivong
09029ca0a4
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). Mapping and Iterable come from collections.abc, imports are sorted. What ruff cannot fix comes in the next commit.
132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
"""Shared helpers for loading the recorded Redfish responses under tests/fixtures/."""
|
|
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
import pytest
|
|
import responses
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
PREFIX = "/redfish/v1"
|
|
|
|
|
|
def fixture_file(mockup: str, path: str) -> Path:
|
|
"""Map a request path to its file: /redfish/v1/X/Y -> <mockup>/X/Y/index.json."""
|
|
path = path.split("#", 1)[0].split("?", 1)[0].rstrip("/")
|
|
if not path.startswith(PREFIX):
|
|
raise ValueError(f"not a Redfish path: {path}")
|
|
relative = path[len(PREFIX) :].lstrip("/")
|
|
return FIXTURES / mockup / relative / "index.json"
|
|
|
|
|
|
def load(mockup: str, path: str) -> dict:
|
|
return json.loads(fixture_file(mockup, path).read_text())
|
|
|
|
|
|
@pytest.fixture
|
|
def rackmount():
|
|
"""Loader for the DMTF public-rackmount1 mockup."""
|
|
return lambda path: load("dmtf-public-rackmount1", path)
|
|
|
|
|
|
@pytest.fixture
|
|
def localstorage():
|
|
"""Loader for the DMTF public-localstorage mockup."""
|
|
return lambda path: load("dmtf-public-localstorage", path)
|
|
|
|
|
|
@pytest.fixture
|
|
def telemetry():
|
|
"""Loader for the DMTF public-telemetry mockup."""
|
|
return lambda path: load("dmtf-public-telemetry", path)
|
|
|
|
|
|
BASE_URL = "https://bmc.example.net"
|
|
SESSIONS = "/redfish/v1/SessionService/Sessions"
|
|
SESSION_PATH = SESSIONS + "/0123456789abcdef"
|
|
TOKEN = "secret-session-token"
|
|
|
|
|
|
class MockupServer:
|
|
"""Serve a fixture mockup over mocked HTTP, plus a SessionService that hands out one token.
|
|
|
|
Unknown paths return a Redfish error body with 404, like a real service. ``override``
|
|
replaces the answer for one method and path (the path may include a query string).
|
|
"""
|
|
|
|
def __init__(self, rsps, mockup: str):
|
|
self.rsps = rsps
|
|
self.mockup = mockup
|
|
self.overrides: dict[tuple[str, str], tuple[int, dict, dict]] = {}
|
|
anything = re.compile(re.escape(BASE_URL) + r"/.*")
|
|
for method in (responses.GET, responses.POST, responses.DELETE):
|
|
rsps.add_callback(method, anything, callback=self._handle)
|
|
|
|
def override(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
status: int,
|
|
body: dict | str | None = None,
|
|
headers: dict | None = None,
|
|
):
|
|
self.overrides[(method, path)] = (status, body if body is not None else {}, headers or {})
|
|
|
|
@staticmethod
|
|
def _reply(status, body, headers=None):
|
|
if isinstance(body, str):
|
|
text = body
|
|
else:
|
|
text = json.dumps(body) if body != {} or status != 204 else ""
|
|
return status, {"Content-Type": "application/json", **(headers or {})}, text
|
|
|
|
def _handle(self, request):
|
|
parts = urlsplit(request.url)
|
|
for key in ((request.method, f"{parts.path}?{parts.query}"), (request.method, parts.path)):
|
|
if key in self.overrides:
|
|
return self._reply(*self.overrides[key])
|
|
if request.method == "POST" and parts.path == SESSIONS:
|
|
return self._reply(
|
|
201,
|
|
{"@odata.id": SESSION_PATH, "Id": "0123456789abcdef"},
|
|
{"X-Auth-Token": TOKEN, "Location": BASE_URL + SESSION_PATH},
|
|
)
|
|
if request.method == "DELETE" and parts.path == SESSION_PATH:
|
|
return self._reply(204, {})
|
|
if request.method == "GET":
|
|
try:
|
|
return self._reply(200, load(self.mockup, parts.path))
|
|
except (FileNotFoundError, ValueError):
|
|
pass
|
|
return self._reply(
|
|
404,
|
|
{
|
|
"error": {
|
|
"code": "Base.1.0.ResourceMissingAtURI",
|
|
"message": f"{parts.path} not found",
|
|
}
|
|
},
|
|
)
|
|
|
|
def calls(self, method: str | None = None, path: str | None = None):
|
|
return [
|
|
c
|
|
for c in self.rsps.calls
|
|
if (method is None or c.request.method == method)
|
|
and (path is None or urlsplit(c.request.url).path == path)
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def http():
|
|
with responses.RequestsMock(assert_all_requests_are_fired=False) as rsps:
|
|
yield rsps
|
|
|
|
|
|
@pytest.fixture
|
|
def server(http):
|
|
"""The public-rackmount1 mockup served on BASE_URL."""
|
|
return MockupServer(http, "dmtf-public-rackmount1")
|