BaseScriptTestCase

class lsst.ts.standardscripts.BaseScriptTestCase

Bases: object

Base class for Script tests.

Subclasses must:

• Inherit both from this and unittest.IsolatedAsyncioTestCase.

• Override basic_make_script 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_script():

await self.configure_script(…) # … test the results of configuring the script (self.script)

await self.run_script() # (unless only testing configuration) # … test the results of running the script

Methods Summary

asyncTearDown()	Runs when the test is done.
basic_make_script(index)	Make a script as self.script.
check_executable(script_path)	Check that an executable script can be launched.
close()	Optional cleanup before closing the scripts and etc.
configure_script(**kwargs)	Configure the script and set the group ID (if using ts_salobj 4.5 or later).
make_script([log_level, timeout, verbose, ...])	Create a Script.
next_index()
run_script([expected_final_state])	Run the script.
wait_for(coro, timeout, description, verbose)	A wrapper around asyncio.wait_for that prints timing information.

Methods Documentation

async asyncTearDown() → None

Runs when the test is done.

This will delete all the topics and schema from the kafka cluster.

abstract async basic_make_script(index)

Make a script as self.script.

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

Parameters:

indexint

The SAL index of the script.

Returns:

itemsList [any]

Controllers, Remotes and Script, 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 check_executable(script_path)

Check that an executable script can be launched.

async close()

Optional cleanup before closing the scripts and etc.

async configure_script(**kwargs)

Configure the script and set the group ID (if using ts_salobj 4.5 or later).

Sets the script state to UNCONFIGURED. This allows you to call configure_script multiple times.

Parameters:

kwargsdict

Keyword arguments for configuration.

Returns:

configtypes.SimpleNamespace

kwargs expressed as a SimpleNamespace. This is provided as a convenience, to avoid boilerplate and duplication in your unit tests. The data is strictly based on the input arguments; it has nothing to do with the script.

make_script(log_level=20, timeout=90, verbose=False, randomize_topic_subname=False)

Create a Script.

The script is accessed as self.script.

Parameters:

namestr

Name of SAL component.

log_levelint (optional)

Logging level, such as logging.INFO.

timeoutfloat

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

verbosebool

Log data? This can be helpful for setting timeout.

randomize_topic_subnamebool

Randomize topic subname?

next_index()

async run_script(expected_final_state=ScriptState.DONE)

Run the script.

Requires that the script be configured and the group ID set (if using ts_salobj 4.5 or later).

async wait_for(coro, timeout, description, verbose)

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.