MockOpsGenie#

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

Bases: object

A mock OpsGenie 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 OpsGenie uses.

Notes

To use:

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

# Or if you prefer explicit start/stop:

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

Raises:: RuntimeError – If env var ESCALATION_KEY not set.

port#

If the specified port is 0, then the chosen port (after start has run). Otherwise the specified port. The default is 80 because that is what OpsGenie uses.

Type:: int

url#

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

Type:: str | None

escalation_key#

The value of env var ESCALATION_KEY.

Type:: str

alerts#

Dict of alert ID: alert data in OpsGenie’s format. The following keys should be in the sent data:

message (str): a summary of the problem
description (str): a detailed description of the problem
responders (list[dict[str, str]]): a list of responders, each of which is a dict with two keys:,
- type (str): one of “user” or “team” (or some other value Watcher will not use).
- id (str): user email address; only present if type is “user”
- name (str): team name; only present if type is “team”

The OpsGenie service the following two keys:

id (str): a unique alert ID
status (str): one of “open” or “closed” (the real OpsGenie service has additional statuses).

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

`assert_authorized`(request)	Raise an error if self.reject_next_request true or not authorized.
`close`()	Stop the service, if running.
`close_alert`(request)	Request handler to set the status of an alert to "closed".
`create_alert`(request)	Request handler to create a new alert.
`delete_alert`(request)	Request handler to delete an alert.
`make_app`()	Make an instance of the web application.
`start`()	Start the service.

Methods Documentation

assert_authorized(request)#

Raise an error if self.reject_next_request true or not authorized.

Always reset self.reject_next_request to False.

Parameters:

request (aiohttp.web.Request) – Request.

Raises:

web.HTTPInternalServerError – If self.reject_next_request True.
web.HTTPForbidden – If the request is not authorized.

Return type:

None

async close()#

Stop the service, if running.

Return type:: None

async close_alert(request)#

Request handler to set the status of an alert to “closed”.

Parameters:: request (aiohttp.web.Request) – Request.
Return type:: json_response

async create_alert(request)#

Request handler to create a new alert.

Parameters:: request (aiohttp.web.Request) – Request.
Return type:: json_response

async delete_alert(request)#

Request handler to delete an alert.

Parameters:: request (aiohttp.web.Request) – Request.
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