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=None)#

Make a group as self.group.

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

Returns:

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

Return type:

List [any]

Notes

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

Parameters:

usage (Optional[int], default: None)

async close()#

Optional cleanup before closing.

Return type:

None

make_group(timeout=30, usage=None, verbose=False)#

Create a Group.

The group is accessed as self.group.

Parameters:
  • timeout (float) – 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.

  • verbose (bool) – Log data? This can be helpful for setting timeout.

Return type:

AsyncGenerator

run(result=None)#

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

https://stackoverflow.com/a/11180583

Parameters:

result (Any, default: None)

Return type:

None

async wait_for(coro, timeout, description, verbose)#

A wrapper around asyncio.wait_for that prints timing information.

Parameters:
  • coro (awaitable) – Coroutine or task to await.

  • timeout (float) – Timeout (seconds)

  • description (str) – Description of what is being awaited.

  • verbose (bool) – 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.

Return type:

Any