MockPagerDuty¶

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

Bases: object

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

Raises:

RuntimeError: If env var ESCALATION_KEY not set.

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.

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.
escalation_keystr: The value of env var ESCALATION_KEY.
eventsdict [str, dict]: 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.
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.

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

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

Always reset self.reject_next_request to False.

Parameters:

routing_keystr: Routing key.

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.

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

async send_event(request: Request) → json_response¶

Request handler for “send event”.

Parameters:

requestaiohttp.web.Request: Request.

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

MockPagerDuty¶