Cloud Integration Testing Framework (citest) is a python package to facilitate writing integration tests against the REST style APIs typically used by cloud services.

The gist is to allow tests to be written using a literary programming style where the core framework accommodates for asynchronous calls and even retries if appropriate.

pip install -r requirements.txt

A test case might look something like this:

  def create_google_load_balancer(self):
    bindings = self._bindings
    load_balancer_name = bindings['TEST_APP_COMPONENT_NAME']

    spec = {
        'checkIntervalSec': 9,
        'healthyThreshold': 3,
        'unhealthyThreshold': 5,
        'timeoutSec': 2,
        'port': 80
    }

    target_pool_name = '{0}/targetPools/{1}-tp'.format(
        bindings['TEST_REGION'], load_balancer_name)

        job=[{
            'cloudProvider': 'gce',
            'provider': 'gce',
            'stack': bindings['TEST_STACK'],
            'detail': bindings['TEST_COMPONENT_DETAIL'],
            'credentials': bindings['GCE_CREDENTIALS'],
            'region': bindings['TEST_GCE_REGION'],
            'ipProtocol': 'TCP',
            'portRange': spec['port'],
            'loadBalancerName': load_balancer_name,
            'healthCheck': {
                'port': spec['port'],
                'timeoutSec': spec['timeoutSec'],
                'checkIntervalSec': spec['checkIntervalSec'],
                'healthyThreshold': spec['healthyThreshold'],
                'unhealthyThreshold': spec['unhealthyThreshold'],
            },
            'type': 'upsertLoadBalancer',
            'availabilityZones': {bindings['TEST_GCE_REGION']: []},
            'user': '[anonymous]'
        }],
        description='Create Load Balancer: ' + load_balancer_name,
        application=bindings['TEST_APP'])

    builder = gcp_testing.GceContractBuilder(self.gce_observer)
    (builder.new_clause_builder('Health Check Added',
                                retryable_for_secs=30)
     .list_resources('http-health-checks')
     .contains_pred_list(
         [json_predicate.PathContainsPredicate(
              'name', '%s-hc' % load_balancer_name),
          json_predicate.DICT_SUBSET(spec)]))
    (builder.new_clause_builder('Target Pool Added',
                                retryable_for_secs=30)
     .list_resources('target-pools')
     .contains_path_value('name', '%s-tp' % load_balancer_name))
    (builder.new_clause_builder('Forwarding Rules Added',
                                retryable_for_secs=30)
     .list_resources('forwarding-rules')
     .contains_pred_list([
          json_predicate.PathContainsPredicate('name', load_balancer_name),
          json_predicate.PathContainsPredicate('target', target_pool_name)]))

    return service_testing.OperationContract(
        self.new_post_operation(
            title='upsert_load_balancer', data=payload, path='tasks'),
        contract=builder.build())

Where the test can be executed like this:

  self.run_test_case(test_case)

Integration tests are written against services by sending an operation to the service then observing the effects of that operation and verifying them against expectations.

citest introduces a BaseAgent class for adapting external services to the framework where the agent is responsible for understanding the transport and protocol for exchanging messages with the service. It introduces some basic concrete types such as HttpAgent and CliAgent where the primary means is HTTP messaging or running a command-line program. Specific systems may need to further specialize these to understand any additional application protocols added, such as status responses for asynchronous HTTP messaging.

BaseAgent also acts as an AgentOperation factory where the operations provide a means to wrap these service calls as first-class objects understood by the core citest framework. This allows citest to invoke (or reinvoke) the operation when it is appropriate to do so, rather than when the static code is specifying what the test will be. When executed, the AgentOperation will create an OperationStatus allowing citest to track its progress and eventual result.

In order to verify the operation, citest uses Contract objects. A contract is a collection of ContractClause where each clause can look for different effects. A ContractClause is typically composed of an observation on the effects and an assertion about what is expected to be observed. The observation is made by an Observer that collects data in an Observation by working with a BaseAgent to collect the data (e.g. an HTTP GET on some JSON resource). The assertion is made by looking for expected values and patterns in the collected resources. Each clause can collect different resources.

The assertions are written using specialized ValuePredicate objects, which are python callable classes that take the object value to be validate and return a PredicateResult containing the conclusion and justification for it.

When a test is run, it will provide a trace of the operations performed, data collected and justifications as to why it thinks the collected data meets or does not meet expectations, ultimately passing or failing the test.

For more examples, see:

• The examples subdirectory.
• Spinnaker Citests.

The Usage Overview Document provides some instructions and examples to guide basic usage.

See the CONTRIBUTING file for more information.

See the LICENSE file for more information.

The package is composed of several subpackages of individual modules.

For more information, problems, or interest, contact [email protected].