Add DMTF mockup fixtures and test helpers
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.
This commit is contained in:
77 files changed
+4051
No files matched your search
@@ -0,0 +1,112 @@
|
||||
"""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")
|
||||
Reference in new issue
Block a user