MockJupyter#

class rubin.nublado.client.MockJupyter(base_url, *, use_subdomains=True)#

Bases: object

A mock Jupyter state machine.

This should be invoked via mocked HTTP calls so that tests can simulate making REST calls to the real JupyterHub and Lab. It simulates the process of spawning a lab, creating a session, and running code within that session.

All calls must contain an Authorization header of the form Bearer <token>, where the token was created by create_mock_token.

Parameters:

  • base_url (str) – Base URL at which to install the Jupyter mocks.

  • use_subdomains (bool, default: True) – If True, simulate per-user subdomains. JupyterHub will use the URL nb.hostname where the hostname is taken from base_url, and JupyterLab will use username.nb.hostname.

Methods Summary

build_code_result(code, variables)

Get the execution results for a piece of code.

create_mock_token(username)

Create a mock Gafaelfawr token for the given user.

create_session(request, user)

delete_lab(request, user)

delete_session(request, user)

fail_on(username, actions)

Configure the given action to fail for the given user.

get_last_spawn_form(username)

Get the contents of the last spawn form submitted for a user.

get_session(username)

Retrieve the currently active mock JupyterLab session for a user.

lab(request, user)

lab_callback(request, user)

Simulate returning to JupyterLab after authentication.

login(request, user)

missing_lab(request, user)

progress(request, user)

register_notebook_result(notebook, result)

Register the result of full notebook execution.

register_python_result(code, result)

Register the expected cell output for a given source input.

run_notebook(request, user)

Simulate the /rubin/execution endpoint.

set_delete_delay(delay)

Set whether to delete labs immediately.

set_redirect_loop(*, enabled)

Set whether to return an infinite redirect loop.

set_spawn_delay(delay)

Set whether to time out during lab spawn.

spawn(request, user)

spawn_pending(request, user)

user(request, user)

Methods Documentation

build_code_result(code, variables)#

Get the execution results for a piece of code.

Normally, this is only used by the JupyterLab WebSocket mock, but test suites can call it directly if they want to see what the mock would return for a given piece of code.

Warning

Code for which no result has been registered via register_python_result will be executed via exec. This mock therefore supports arbitrary code execution via its handlers and must never be exposed to untrusted messages.

Parameters:

  • code (str) – Code for which to retrieve or generate results.

  • variables (dict[str, Any]) – Dictionary holding global and local variables. This may be written to by the code, and the caller should keep passing the same dictionary if variable continuity across cell execution (emulating a real notebook) is desired.

Returns:

Corresponding results.

Return type:

str

Raises:

Exception – Raised if there is no registered result for a given piece of code and the code produces an exception when run with exec, or if the registered code result is a BaseException object (which is then raised).

static create_mock_token(username)#

Create a mock Gafaelfawr token for the given user.

Parameters:

username (str) – Username for which to create a token.

Returns:

Mock token usable only with MockJupyter that will be considered a valid token for the given username.

Return type:

str

async create_session(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async delete_lab(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async delete_session(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

fail_on(username, actions)#

Configure the given action to fail for the given user.

This can be used by test suites to test handling of Nublado failures at various calls in the process of executing code in a JupyterLab.

Parameters:
Return type:

None

get_last_spawn_form(username)#

Get the contents of the last spawn form submitted for a user.

Parameters:

username (str) – Username of the user.

Returns:

Key and value pairs submitted to the Nublado spawn form, or None if that user hasn’t submitted a spawn form. Note that although a native Python parsing of a form submission will return a list of values for each key, this method checks that only one value is present for each key and then removes the list wrapper.

Return type:

dict of str or None

get_session(username)#

Retrieve the currently active mock JupyterLab session for a user.

Parameters:

username (str) – Username of the user.

Returns:

Current open JupyterLab session for that user, or None if none is open.

Return type:

MockJupyterLabSession or None

async lab(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async lab_callback(request, user)#

Simulate returning to JupyterLab after authentication.

Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async login(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async missing_lab(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async progress(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

register_notebook_result(notebook, result)#

Register the result of full notebook execution.

Parameters:

  • code – Full notebook contents as a JSON-formatted string.

  • result (NotebookExecutionResult) – Results to return when that notebook is executed via the mock.

  • notebook (str)

Return type:

None

register_python_result(code, result)#

Register the expected cell output for a given source input.

Whenever the given code is executed inside a JupyterLab session in the mock, the given result will be returned or, if it is an exception type, raised.

Parameters:

  • code (str) – Expected code block.

  • result (str | BaseException) – Result (standard output) of execution, or a BaseException object to raise that exception.

Return type:

None

async run_notebook(request, user)#

Simulate the /rubin/execution endpoint.

Notes

This does not use the nbconvert/nbformat method of the actual endpoint, because installing kernels into what are already-running pythons in virtual evironments in the testing environment is nasty.

First, we will try using the input notebook text as a key into a cache of registered responses (this is analogous to doing the same with registered responses to python snippets in the Session mock): if the key is present, then we will return the response that corresponds to that key.

If not, we’re just going to return the input notebook as if it ran without errors, but without updating any of its outputs or resources, or throwing an error. This is not a a very good simulation. But since the whole point of this is to run a notebook in a particular kernel context, and for us that usually means the “LSST” kernel with the DM Pipelines Stack in it, that would be incredibly awful to use in a unit test context. If you want to know if your notebook will really work, you’re going to have to run it in the correct kernel, and the client unit tests are not the place for that.

Much more likely is that you have a test notebook that should produce certain results in the wild. In that case, you would register those results, and then the correct output would be delivered by the cache.

Parameters:

  • request (Request)

  • user (str)

Return type:

Response

set_delete_delay(delay)#

Set whether to delete labs immediately.

By default, the mock deletes user labs immediately. Sometimes the caller may want to test handling of the lab shutdown. It can do that by calling this method on the mock and set a delay for how long it should take to delete a lab.

Parameters:

delay (timedelta | None) – How long to wait before deleting the lab or None to not wait. The lab deletion will actually happen after this delay has passed and the client calls the mock route that returns the list of a user’s current labs (called by is_lab_stopped).

Return type:

None

set_redirect_loop(*, enabled)#

Set whether to return an infinite redirect loop.

If enabled, the endpoints for getting the JupyterHub and JupyterLab top-level pages and for watching spawn progress will instead return an infinite redirect loop to the same URL.

Parameters:

enabled (bool) – Whether to enable a redirect loop.

Return type:

None

set_spawn_delay(delay)#

Set whether to time out during lab spawn.

If enabled, the endpoint for watching spawn progress will wait for this long before returning success.

Parameters:

delay (timedelta | None) – How long to delay spawn.

Return type:

None

async spawn(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async spawn_pending(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response

async user(request, user)#
Parameters:

  • request (Request)

  • user (str)

Return type:

Response