streamsx.topology.tester module

class streamsx.topology.tester.Condition(name=None)

Bases: object

fail()
valid
class streamsx.topology.tester.Tester(topology)

Bases: object

Testing support for a Topology.

Allows testing of a Topology by creating conditions against the contents of its streams.

Conditions may be added to a topology at any time before submission.

If a topology is submitted directly to a context then the graph is not modified. This allows testing code to be inserted while the topology is being built, but not acted upon unless the topology is submitted in test mode.

If a topology is submitted through the test method then the topology may be modified to include operations to ensure the conditions are met.

Parameters:topology – Topology to be tested.
add_condition(stream, condition)
contents(stream, expected, ordered=True)

Test that a stream contains the expected tuples.

Parameters:
  • stream (Stream) – Stream to be tested.
  • expected (list) – Sequence of expected tuples.
  • ordered (bool) – True if the ordering of received tuples must match expected.

Returns:

static setup_distributed(test)

Setup a unittest.TestCase to run tests using IBM Streams distributed mode.

Requires a local IBM Streams install define by the STREAMS_INSTALL environment variable. If STREAMS_INSTALL is not set then the test is skipped.

The Steams instance to use is defined by the environment variables:
  • STREAMS_ZKCONNECT - Zookeeper connection string
  • STREAMS_DOMAIN_ID - Domain identifier
  • STREAMS_INSTANCE_ID - Instance identifier
Two attributes are set in the test case:
  • test_ctxtype - Context type the test will be run in.
  • test_config- Test configuration.
Parameters:test (unittest.TestCase) – Test case to be set up to run tests using Tester

Returns: None

static setup_standalone(test)

Setup a unittest.TestCase to run tests using IBM Streams standalone mode.

Requires a local IBM Streams install define by the STREAMS_INSTALL environment variable. If STREAMS_INSTALL is not set then the test is skipped.

Two attributes are set in the test case:
  • test_ctxtype - Context type the test will be run in.
  • test_config- Test configuration.
Parameters:test (unittest.TestCase) – Test case to be set up to run tests using Tester

Returns: None

setup_streaming_analytics(test, service_name=None, force_remote_build=False)

Setup a unittest.TestCase to run tests using Streaming Analytics service on IBM Bluemix cloud platform.

The service to use is defined by:
  • VCAP_SERVICES environment variable containing streaming_analytics entries.
  • service_name which defaults to the value of STREAMING_ANALYTICS_SERVICE_NAME environment variable.

If VCAP_SERVICES is not set or a service name is not defined then the test is skipped.

Two attributes are set in the test case:
  • test_ctxtype - Context type the test will be run in.
  • test_config- Test configuration.
Parameters:
  • test (unittest.TestCase) – Test case to be set up to run tests using Tester
  • service_name (str) – Name of Streaming Analytics service to use. Must exist as an entry in the VCAP services. Defaults to value of STREAMING_ANALYTICS_SERVICE_NAME environment variable.

Returns: None

test(ctxtype, config=None, assert_on_fail=True, username=None, password=None)

Test the topology.

Submits the topology for testing and verifies the test conditions are met.

Parameters:
  • ctxtype (str) – Context type for submission.
  • config – Configuration for submission.
  • assert_on_fail (bool) – True to raise an assertion if the test fails, False to return the passed status.
  • username (str) – username for distributed tests
  • password (str) – password for distributed tests
Returns:

True if test passed, False if test failed.
Return type:

bool
tuple_count(stream, count)

Test that that a stream returns an exact number of tuples.

Parameters:
  • stream (Stream) – Stream to be tested.
  • count – Number of tuples expected.

Returns: stream