Christian Manivong
c8bcd5ace1
Unmodified files from DMTF Redfish-Publications (public-rackmount1, public-localstorage, public-telemetry), only those the tests use, each set with DMTF's BSD-3-Clause license and a COMMANDS.txt naming the source commit and request paths. conftest.py loads them by request path and serves a mockup over mocked HTTP, with a SessionService that hands out a token, for the transport tests.
113 lines
4.0 KiB
Python
113 lines
4.0 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")
|