MockOpsGenie¶

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

Bases: object

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

Raises:

RuntimeError: If env var ESCALATION_KEY not set.

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

Attributes:

portint

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.

urlstr | None

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

escalation_keystr

The value of env var ESCALATION_KEY.

alertsdict [str, dict]

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).

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

`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: Request) → None¶

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

Always reset self.reject_next_request to False.

Parameters:

requestaiohttp.web.Request: Request.

Raises:

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

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

async close_alert(request: Request) → json_response¶

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

Parameters:

requestaiohttp.web.Request: Request.

async create_alert(request: Request) → json_response¶

Request handler to create a new alert.

Parameters:

requestaiohttp.web.Request: Request.

async delete_alert(request: Request) → json_response¶

Request handler to delete an alert.

Parameters:

requestaiohttp.web.Request: Request.

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

MockOpsGenie¶