MockSquadCast¶

class lsst.ts.watcher.MockSquadCast(port: int = 80)¶

Bases: object

A mock SquadCast service to support Watcher escalation in unit tests.

Parameters:

portint: The TCP/IP port for the service. 0 to choose a free port. The default is 80 because that is what SquadCast uses.

Raises:

RuntimeError: If env var ESCALATION_KEY not set.

Notes

To use:

async with MockSquadCast(port=...) as server:
    # ... use the server

# Or if you prefer explicit start/stop:

server = MockSquadCast(port=...)
await server.start()
# ... use the server
await server.close()

Known limitations:

SquadCast supports having multiple incidents with the same event_id. This mock does not – it will reject creation of a new incident with a matching event ID.
SquadCast does not require the event_id field. This mock does, because it stores events by event_id.
SquadCast does not require the status field, based on tests, but the manual says nothing about it so I prefer to require it.
SquadCast does not require the “description” field when triggering an incident, but the manual says it is required. This emulates the manual.
The incidents dict is never purged, which is a memory leak, but emulates SquadCast. Thus this mock is intended for short-term unit tests, not long term emulation.

Attributes:

portint

The port. The port argument, except if that was 0 then this is updated to the chosen port when start is run.

urlstr

The root URL of the service. “” until start has run.

endpoint_urlstr

The URL of the endpoint (including the secret API key). “” until start has run.

escalation_keystr

The value of env var ESCALATION_KEY.

incidentsdict [str, dict]

Dict of event_id: incident data in SquadCast’s format, which has these keys:

event_id (str): unique identifier
status (str): one of “trigger” or “resolve”
message (str): summary of the problem
details (str): additional information
tags (dict[str, dict[str, str]]): a dict of tag_name: value.

These are keys that SquadCast uses. The first four are required and the mock does not check the rest of the data. Values for SquadCast “tags” can be str or dicts with a specific format, but Watcher only uses str values.

reject_next_requestbool

If the user sets this true then the mock will reject the next request with web.HTTPInternalServerError and reset this flag. For unit tests.

Methods Summary

`close`()	Stop the service, if running.
`incident_webhook`(request)	Request handler to trigger or resolve an incident.
`make_app`()	Make an instance of the web application.
`start`()	Start the service.

Methods Documentation

async close() → None¶: Stop the service, if running.

async incident_webhook(request: Request) → json_response¶

Request handler to trigger or resolve an incident.

Parameters:

requestaiohttp.web.Request: Request.

Raises:

web.HTTPInternalServerError: If self.reject_next_request True.
web.HTTPRequestEntityTooLarge: If the request is too large.

make_app() → Application¶: Make an instance of the web application.

async start() → None¶

Start the service.

Raises:

RuntimeError: If port = 0 and serving on more than one socket (in which case the served port cannot be determined), or if this method has already been called.

Navigation

MockSquadCast¶