MockPagerDuty#

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

Bases: object

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

Notes

To use:

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

# Or if you prefer explicit start/stop:

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

Known limitations:

At present this only supports event creation. We would probably want need add support for close and/or delete if we decide to use PagerDuty for the Watcher.
The events dict is never purged (though it would be if we add delete support). This is an explicit memory leak.

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

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

escalation_key#

The value of env var ESCALATION_KEY.

Type:: str

events#

Dict of event ID: event data in PagerDuty’s format. See ALLOWED_KEYS for the allowed keys; all values are str except “payload”, which is a dict of [str, str]. “payload” has the information about what went wrong; its keys are described in ALLOWED_PAYLOAD_KEYS.

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

Attributes Summary

`ALLOWED_KEYS`
`ALLOWED_PAYLOAD_KEYS`

Methods Summary

`assert_authorized`(routing_key)	Raise an error if self.reject_next_request true or not authorized.
`close`()	Stop the service, if running.
`make_app`()	Make an instance of the web application.
`send_event`(request)	Request handler for "send event".
`start`()	Start the service.

Attributes Documentation

ALLOWED_KEYS = frozenset({'client', 'client_url', 'dedup_key', 'event_action', 'images', 'links', 'payload', 'routing_key'})#

ALLOWED_PAYLOAD_KEYS = frozenset({'class', 'component', 'custom_details', 'group', 'severity', 'source', 'summary', 'timestamp'})#

Methods Documentation

assert_authorized(routing_key)#

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

Always reset self.reject_next_request to False.

Parameters:

routing_key (str) – Routing key.

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

make_app()#

Make an instance of the web application.

Return type:: Application

async send_event(request)#

Request handler for “send event”.

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

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