RemoteGroupTestCase

class lsst.ts.observatory.control.RemoteGroupTestCase

Bases: object

Base class for testing groups of CSCs.

Subclasses must:

• Inherit both from this and asynctest.TestCase.

• Override basic_make_group to make the script and any other controllers, remotes and such, and return a list of scripts, controllers and remotes that you made.

A typical test will look like this:

async def test_something(self):

async with make_group():

# … test something

Methods Summary

basic_make_group([usage])	Make a group as self.group.
close()	Optional cleanup before closing.
make_group([timeout, usage, verbose])	Create a Group.
run([result])	Override run to set a random LSST_DDS_PARTITION_PREFIX and set LSST_SITE=test for every test.
wait_for(coro, timeout, description, verbose)	A wrapper around asyncio.wait_for that prints timing information.

Methods Documentation

abstract async basic_make_group(usage: Optional[int] = None) → None

Make a group as self.group.

Make all other controllers and remotes, as well and return a list of the items made.

Returns:

itemsList [any]

Controllers, Remotes and Group, or any other items for which to initially wait for item.start_task and finally wait for item.close().

Notes

This is a coroutine in the unlikely case that you might want to wait for something.

async close() → None

Optional cleanup before closing.

make_group(timeout: float = 30, usage: Optional[int] = None, verbose: bool = False) → AsyncGenerator

Create a Group.

The group is accessed as self.group.

Parameters:

timeoutfloat

Timeout (sec) for waiting for item.start_task and item.close() for each item returned by basic_make_script, and self.close.

usage: `int`

Combined enumeration with intended usage.

verbosebool

Log data? This can be helpful for setting timeout.

run(result: Any = None) → None

Override run to set a random LSST_DDS_PARTITION_PREFIX and set LSST_SITE=test for every test.

https://stackoverflow.com/a/11180583

async wait_for(coro: Awaitable, timeout: float, description: str, verbose: bool) → Any

A wrapper around asyncio.wait_for that prints timing information.

Parameters:

coroawaitable

Coroutine or task to await.

timeoutfloat

Timeout (seconds)

descriptionstr

Description of what is being awaited.

verbosebool

If True then print a message before waiting and another after that includes how long it waited. If False only print a message if the wait times out.