MockSquadCast#

class lsst.ts.watcher.MockSquadCast(port=80)#

Bases: object

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

Parameters:

port (int) – 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.

port#

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

Type:

int

url#

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

Type:

str

endpoint_url#

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

Type:

str

escalation_key#

The value of env var ESCALATION_KEY.

Type:

str

incidents#

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.

Type:

dict [str, dict]

reject_next_request#

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

Type:

bool

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()#

Stop the service, if running.

Return type:

None

async incident_webhook(request)#

Request handler to trigger or resolve an incident.

Parameters:

request (aiohttp.web.Request) – Request.

Raises:

  • web.HTTPInternalServerError – If self.reject_next_request True.

  • web.HTTPRequestEntityTooLarge – If the request is too large.

Return type:

json_response

make_app()#

Make an instance of the web application.

Return type:

Application

async start()#

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.

Return type:

None