# integ-tests
<!--BEGIN STABILITY BANNER-->---
> The APIs of higher level constructs in this module are experimental and under active development.
> They are subject to non-backward compatible changes or removal in any future version. These are
> not subject to the [Semantic Versioning](https://semver.org/) model and breaking changes will be
> announced in the release notes. This means that while you may use them, you may need to update
> your source code when upgrading to a newer version of this package.
---
<!--END STABILITY BANNER-->
## Overview
This library is meant to be used in combination with the [integ-runner](https://github.com/aws/aws-cdk/tree/main/packages/%40aws-cdk/integ-runner) CLI
to enable users to write and execute integration tests for AWS CDK Constructs.
An integration test should be defined as a CDK application, and
there should be a 1:1 relationship between an integration test and a CDK application.
So for example, in order to create an integration test called `my-function`
we would need to create a file to contain our integration test application.
*test/integ.my-function.ts*
```python
app = App()
stack = Stack()
lambda_.Function(stack, "MyFunction",
runtime=lambda_.Runtime.NODEJS_LATEST,
handler="index.handler",
code=lambda_.Code.from_asset(path.join(__dirname, "lambda-handler"))
)
```
This is a self contained CDK application which we could deploy by running
```bash
cdk deploy --app 'node test/integ.my-function.js'
```
In order to turn this into an integration test, all that is needed is to
use the `IntegTest` construct.
```python
# app: App
# stack: Stack
IntegTest(app, "Integ", test_cases=[stack])
```
You will notice that the `stack` is registered to the `IntegTest` as a test case.
Each integration test can contain multiple test cases, which are just instances
of a stack. See the [Usage](#usage) section for more details.
## Usage
### IntegTest
Suppose you have a simple stack, that only encapsulates a Lambda function with a
certain handler:
```python
class StackUnderTest(Stack):
def __init__(self, scope, id, *, architecture=None, description=None, env=None, stackName=None, tags=None, notificationArns=None, synthesizer=None, terminationProtection=None, analyticsReporting=None, crossRegionReferences=None, permissionsBoundary=None, suppressTemplateIndentation=None):
super().__init__(scope, id, architecture=architecture, description=description, env=env, stackName=stackName, tags=tags, notificationArns=notificationArns, synthesizer=synthesizer, terminationProtection=terminationProtection, analyticsReporting=analyticsReporting, crossRegionReferences=crossRegionReferences, permissionsBoundary=permissionsBoundary, suppressTemplateIndentation=suppressTemplateIndentation)
lambda_.Function(self, "Handler",
runtime=lambda_.Runtime.NODEJS_LATEST,
handler="index.handler",
code=lambda_.Code.from_asset(path.join(__dirname, "lambda-handler")),
architecture=architecture
)
```
You may want to test this stack under different conditions. For example, we want
this stack to be deployed correctly, regardless of the architecture we choose
for the Lambda function. In particular, it should work for both `ARM_64` and
`X86_64`. So you can create an `IntegTestCase` that exercises both scenarios:
```python
class StackUnderTest(Stack):
def __init__(self, scope, id, *, architecture=None, description=None, env=None, stackName=None, tags=None, notificationArns=None, synthesizer=None, terminationProtection=None, analyticsReporting=None, crossRegionReferences=None, permissionsBoundary=None, suppressTemplateIndentation=None):
super().__init__(scope, id, architecture=architecture, description=description, env=env, stackName=stackName, tags=tags, notificationArns=notificationArns, synthesizer=synthesizer, terminationProtection=terminationProtection, analyticsReporting=analyticsReporting, crossRegionReferences=crossRegionReferences, permissionsBoundary=permissionsBoundary, suppressTemplateIndentation=suppressTemplateIndentation)
lambda_.Function(self, "Handler",
runtime=lambda_.Runtime.NODEJS_LATEST,
handler="index.handler",
code=lambda_.Code.from_asset(path.join(__dirname, "lambda-handler")),
architecture=architecture
)
# Beginning of the test suite
app = App()
IntegTest(app, "DifferentArchitectures",
test_cases=[
StackUnderTest(app, "Stack1",
architecture=lambda_.Architecture.ARM_64
),
StackUnderTest(app, "Stack2",
architecture=lambda_.Architecture.X86_64
)
]
)
```
This is all the instruction you need for the integration test runner to know
which stacks to synthesize, deploy and destroy. But you may also need to
customize the behavior of the runner by changing its parameters. For example:
```python
app = App()
stack_under_test = Stack(app, "StackUnderTest")
stack = Stack(app, "stack")
test_case = IntegTest(app, "CustomizedDeploymentWorkflow",
test_cases=[stack_under_test],
diff_assets=True,
stack_update_workflow=True,
cdk_command_options=CdkCommands(
deploy=DeployCommand(
args=DeployOptions(
require_approval=RequireApproval.NEVER,
json=True
)
),
destroy=DestroyCommand(
args=DestroyOptions(
force=True
)
)
)
)
```
### IntegTestCaseStack
In the majority of cases an integration test will contain a single `IntegTestCase`.
By default when you create an `IntegTest` an `IntegTestCase` is created for you
and all of your test cases are registered to this `IntegTestCase`. The `IntegTestCase`
and `IntegTestCaseStack` constructs are only needed when it is necessary to
defined different options for individual test cases.
For example, you might want to have one test case where `diffAssets` is enabled.
```python
# app: App
# stack_under_test: Stack
test_case_with_assets = IntegTestCaseStack(app, "TestCaseAssets",
diff_assets=True
)
IntegTest(app, "Integ", test_cases=[stack_under_test, test_case_with_assets])
```
## Assertions
This library also provides a utility to make assertions against the infrastructure that the integration test deploys.
There are two main scenarios in which assertions are created.
* Part of an integration test using `integ-runner`
In this case you would create an integration test using the `IntegTest` construct and then make assertions using the `assert` property.
You should **not** utilize the assertion constructs directly, but should instead use the `methods` on `IntegTest.assertions`.
```python
# app: App
# stack: Stack
integ = IntegTest(app, "Integ", test_cases=[stack])
integ.assertions.aws_api_call("S3", "getObject")
```
By default an assertions stack is automatically generated for you. You may however provide your own stack to use.
```python
# app: App
# stack: Stack
# assertion_stack: Stack
integ = IntegTest(app, "Integ", test_cases=[stack], assertion_stack=assertion_stack)
integ.assertions.aws_api_call("S3", "getObject")
```
* Part of a normal CDK deployment
In this case you may be using assertions as part of a normal CDK deployment in order to make an assertion on the infrastructure
before the deployment is considered successful. In this case you can utilize the assertions constructs directly.
```python
# my_app_stack: Stack
AwsApiCall(my_app_stack, "GetObject",
service="S3",
api="getObject"
)
```
### DeployAssert
Assertions are created by using the `DeployAssert` construct. This construct creates it's own `Stack` separate from
any stacks that you create as part of your integration tests. This `Stack` is treated differently from other stacks
by the `integ-runner` tool. For example, this stack will not be diffed by the `integ-runner`.
`DeployAssert` also provides utilities to register your own assertions.
```python
# my_custom_resource: CustomResource
# stack: Stack
# app: App
integ = IntegTest(app, "Integ", test_cases=[stack])
integ.assertions.expect("CustomAssertion",
ExpectedResult.object_like({"foo": "bar"}),
ActualResult.from_custom_resource(my_custom_resource, "data"))
```
In the above example an assertion is created that will trigger a user defined `CustomResource`
and assert that the `data` attribute is equal to `{ foo: 'bar' }`.
### API Calls
A common method to retrieve the "actual" results to compare with what is expected is to make an
API call to receive some data. This library does this by utilizing CloudFormation custom resources
which means that CloudFormation will call out to a Lambda Function which will
make the API call.
#### HttpApiCall
Using the `HttpApiCall` will use the
[node-fetch](https://github.com/node-fetch/node-fetch) JavaScript library to
make the HTTP call.
This can be done by using the class directory (in the case of a normal deployment):
```python
# stack: Stack
HttpApiCall(stack, "MyAsssertion",
url="https://example-api.com/abc"
)
```
Or by using the `httpApiCall` method on `DeployAssert` (when writing integration tests):
```python
# app: App
# stack: Stack
integ = IntegTest(app, "Integ",
test_cases=[stack]
)
integ.assertions.http_api_call("https://example-api.com/abc")
```
#### AwsApiCall
Using the `AwsApiCall` construct will use the AWS JavaScript SDK to make the API call.
This can be done by using the class directory (in the case of a normal deployment):
```python
# stack: Stack
AwsApiCall(stack, "MyAssertion",
service="SQS",
api="receiveMessage",
parameters={
"QueueUrl": "url"
}
)
```
Or by using the `awsApiCall` method on `DeployAssert` (when writing integration tests):
```python
# app: App
# stack: Stack
integ = IntegTest(app, "Integ",
test_cases=[stack]
)
integ.assertions.aws_api_call("SQS", "receiveMessage", {
"QueueUrl": "url"
})
```
You must specify the `service` and the `api` when using The `AwsApiCall` construct.
The `service` is the name of an AWS service, in one of the following forms:
* An AWS SDK for JavaScript v3 package name (`@aws-sdk/client-api-gateway`)
* An AWS SDK for JavaScript v3 client name (`api-gateway`)
* An AWS SDK for JavaScript v2 constructor name (`APIGateway`)
* A lowercase AWS SDK for JavaScript v2 constructor name (`apigateway`)
The `api` is the name of an AWS API call, in one of the following forms:
* An API call name as found in the API Reference documentation (`GetObject`)
* The API call name starting with a lowercase letter (`getObject`)
* The AWS SDK for JavaScript v3 command class name (`GetObjectCommand`)
By default, the `AwsApiCall` construct will automatically add the correct IAM policies
to allow the Lambda function to make the API call. It does this based on the `service`
and `api` that is provided. In the above example the service is `SQS` and the api is
`receiveMessage` so it will create a policy with `Action: 'sqs:ReceiveMessage`.
There are some cases where the permissions do not exactly match the service/api call, for
example the S3 `listObjectsV2` api. In these cases it is possible to add the correct policy
by accessing the `provider` object.
```python
# app: App
# stack: Stack
# integ: IntegTest
api_call = integ.assertions.aws_api_call("S3", "listObjectsV2", {
"Bucket": "mybucket"
})
api_call.provider.add_to_role_policy({
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:ListBucket"],
"Resource": ["*"]
})
```
When executing `waitForAssertion()`, it is necessary to add an IAM policy using `waiterProvider.addToRolePolicy()`.
Because `IApiCall` does not have a `waiterProvider` property, you need to cast it to `AwsApiCall`.
```python
# integ: IntegTest
api_call = integ.assertions.aws_api_call("S3", "listObjectsV2", {
"Bucket": "mybucket"
}).wait_for_assertions()
api_call.waiter_provider.add_to_role_policy({
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:ListBucket"],
"Resource": ["*"]
})
```
Note that addToRolePolicy() uses direct IAM JSON policy blobs, not a iam.PolicyStatement
object like you will see in the rest of the CDK.
### EqualsAssertion
This library currently provides the ability to assert that two values are equal
to one another by utilizing the `EqualsAssertion` class. This utilizes a Lambda
backed `CustomResource` which in tern uses the [Match](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.assertions.Match.html) utility from the
[@aws-cdk/assertions](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.assertions-readme.html) library.
```python
# app: App
# stack: Stack
# queue: sqs.Queue
# fn: lambda.IFunction
integ = IntegTest(app, "Integ",
test_cases=[stack]
)
integ.assertions.invoke_function(
function_name=fn.function_name,
invocation_type=InvocationType.EVENT,
payload=JSON.stringify({"status": "OK"})
)
message = integ.assertions.aws_api_call("SQS", "receiveMessage", {
"QueueUrl": queue.queue_url,
"WaitTimeSeconds": 20
})
message.assert_at_path("Messages.0.Body", ExpectedResult.object_like({
"request_context": {
"condition": "Success"
},
"request_payload": {
"status": "OK"
},
"response_context": {
"status_code": 200
},
"response_payload": "success"
}))
```
#### Match
`integ-tests` also provides a `Match` utility similar to the `@aws-cdk/assertions` module. `Match`
can be used to construct the `ExpectedResult`. While the utility is similar, only a subset of methods are currently available on the `Match` utility of this module: `arrayWith`, `objectLike`, `stringLikeRegexp` and `serializedJson`.
```python
# message: AwsApiCall
message.expect(ExpectedResult.object_like({
"Messages": Match.array_with([{
"Payload": Match.serialized_json({"key": "value"})
}, {
"Body": {
"Values": Match.array_with([{"Asdf": 3}]),
"Message": Match.string_like_regexp("message")
}
}
])
}))
```
### Examples
#### Invoke a Lambda Function
In this example there is a Lambda Function that is invoked and
we assert that the payload that is returned is equal to '200'.
```python
# lambda_function: lambda.IFunction
# app: App
stack = Stack(app, "cdk-integ-lambda-bundling")
integ = IntegTest(app, "IntegTest",
test_cases=[stack]
)
invoke = integ.assertions.invoke_function(
function_name=lambda_function.function_name
)
invoke.expect(ExpectedResult.object_like({
"Payload": "200"
}))
```
The above example will by default create a CloudWatch log group that's never
expired. If you want to configure it with custom log retention days, you need
to specify the `logRetention` property.
```python
import aws_cdk.aws_logs as logs
# lambda_function: lambda.IFunction
# app: App
stack = Stack(app, "cdk-integ-lambda-bundling")
integ = IntegTest(app, "IntegTest",
test_cases=[stack]
)
invoke = integ.assertions.invoke_function(
function_name=lambda_function.function_name,
log_retention=logs.RetentionDays.ONE_WEEK
)
```
#### Make an AWS API Call
In this example there is a StepFunctions state machine that is executed
and then we assert that the result of the execution is successful.
```python
# app: App
# stack: Stack
# sm: IStateMachine
test_case = IntegTest(app, "IntegTest",
test_cases=[stack]
)
# Start an execution
start = test_case.assertions.aws_api_call("StepFunctions", "startExecution", {
"state_machine_arn": sm.state_machine_arn
})
# describe the results of the execution
describe = test_case.assertions.aws_api_call("StepFunctions", "describeExecution", {
"execution_arn": start.get_att_string("executionArn")
})
# assert the results
describe.expect(ExpectedResult.object_like({
"status": "SUCCEEDED"
}))
```
#### Chain ApiCalls
Sometimes it may be necessary to chain API Calls. Since each API call is its own resource, all you
need to do is add a dependency between the calls. There is an helper method `next` that can be used.
```python
# integ: IntegTest
integ.assertions.aws_api_call("S3", "putObject", {
"Bucket": "amzn-s3-demo-bucket",
"Key": "my-key",
"Body": "helloWorld"
}).next(integ.assertions.aws_api_call("S3", "getObject", {
"Bucket": "amzn-s3-demo-bucket",
"Key": "my-key"
}))
```
#### Wait for results
A common use case when performing assertions is to wait for a condition to pass. Sometimes the thing
that you are asserting against is not done provisioning by the time the assertion runs. In these
cases it is possible to run the assertion asynchronously by calling the `waitForAssertions()` method.
Taking the example above of executing a StepFunctions state machine, depending on the complexity of
the state machine, it might take a while for it to complete.
```python
# app: App
# stack: Stack
# sm: IStateMachine
test_case = IntegTest(app, "IntegTest",
test_cases=[stack]
)
# Start an execution
start = test_case.assertions.aws_api_call("StepFunctions", "startExecution", {
"state_machine_arn": sm.state_machine_arn
})
# describe the results of the execution
describe = test_case.assertions.aws_api_call("StepFunctions", "describeExecution", {
"execution_arn": start.get_att_string("executionArn")
}).expect(ExpectedResult.object_like({
"status": "SUCCEEDED"
})).wait_for_assertions()
```
When you call `waitForAssertions()` the assertion provider will continuously make the `awsApiCall` until the
`ExpectedResult` is met. You can also control the parameters for waiting, for example:
```python
# test_case: IntegTest
# start: IApiCall
describe = test_case.assertions.aws_api_call("StepFunctions", "describeExecution", {
"execution_arn": start.get_att_string("executionArn")
}).expect(ExpectedResult.object_like({
"status": "SUCCEEDED"
})).wait_for_assertions(
total_timeout=Duration.minutes(5),
interval=Duration.seconds(15),
backoff_rate=3
)
```
Raw data
{
"_id": null,
"home_page": "https://github.com/aws/aws-cdk",
"name": "aws-cdk.integ-tests-alpha",
"maintainer": null,
"docs_url": null,
"requires_python": "~=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Amazon Web Services",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/50/a5/72e70db5ad4e3a64769480bd1c7020b68c86104a1f1470b40e02f22983ba/aws_cdk_integ_tests_alpha-2.170.0a0.tar.gz",
"platform": null,
"description": "# integ-tests\n\n<!--BEGIN STABILITY BANNER-->---\n\n\n\n\n> The APIs of higher level constructs in this module are experimental and under active development.\n> They are subject to non-backward compatible changes or removal in any future version. These are\n> not subject to the [Semantic Versioning](https://semver.org/) model and breaking changes will be\n> announced in the release notes. This means that while you may use them, you may need to update\n> your source code when upgrading to a newer version of this package.\n\n---\n<!--END STABILITY BANNER-->\n\n## Overview\n\nThis library is meant to be used in combination with the [integ-runner](https://github.com/aws/aws-cdk/tree/main/packages/%40aws-cdk/integ-runner) CLI\nto enable users to write and execute integration tests for AWS CDK Constructs.\n\nAn integration test should be defined as a CDK application, and\nthere should be a 1:1 relationship between an integration test and a CDK application.\n\nSo for example, in order to create an integration test called `my-function`\nwe would need to create a file to contain our integration test application.\n\n*test/integ.my-function.ts*\n\n```python\napp = App()\nstack = Stack()\nlambda_.Function(stack, \"MyFunction\",\n runtime=lambda_.Runtime.NODEJS_LATEST,\n handler=\"index.handler\",\n code=lambda_.Code.from_asset(path.join(__dirname, \"lambda-handler\"))\n)\n```\n\nThis is a self contained CDK application which we could deploy by running\n\n```bash\ncdk deploy --app 'node test/integ.my-function.js'\n```\n\nIn order to turn this into an integration test, all that is needed is to\nuse the `IntegTest` construct.\n\n```python\n# app: App\n# stack: Stack\n\nIntegTest(app, \"Integ\", test_cases=[stack])\n```\n\nYou will notice that the `stack` is registered to the `IntegTest` as a test case.\nEach integration test can contain multiple test cases, which are just instances\nof a stack. See the [Usage](#usage) section for more details.\n\n## Usage\n\n### IntegTest\n\nSuppose you have a simple stack, that only encapsulates a Lambda function with a\ncertain handler:\n\n```python\nclass StackUnderTest(Stack):\n def __init__(self, scope, id, *, architecture=None, description=None, env=None, stackName=None, tags=None, notificationArns=None, synthesizer=None, terminationProtection=None, analyticsReporting=None, crossRegionReferences=None, permissionsBoundary=None, suppressTemplateIndentation=None):\n super().__init__(scope, id, architecture=architecture, description=description, env=env, stackName=stackName, tags=tags, notificationArns=notificationArns, synthesizer=synthesizer, terminationProtection=terminationProtection, analyticsReporting=analyticsReporting, crossRegionReferences=crossRegionReferences, permissionsBoundary=permissionsBoundary, suppressTemplateIndentation=suppressTemplateIndentation)\n\n lambda_.Function(self, \"Handler\",\n runtime=lambda_.Runtime.NODEJS_LATEST,\n handler=\"index.handler\",\n code=lambda_.Code.from_asset(path.join(__dirname, \"lambda-handler\")),\n architecture=architecture\n )\n```\n\nYou may want to test this stack under different conditions. For example, we want\nthis stack to be deployed correctly, regardless of the architecture we choose\nfor the Lambda function. In particular, it should work for both `ARM_64` and\n`X86_64`. So you can create an `IntegTestCase` that exercises both scenarios:\n\n```python\nclass StackUnderTest(Stack):\n def __init__(self, scope, id, *, architecture=None, description=None, env=None, stackName=None, tags=None, notificationArns=None, synthesizer=None, terminationProtection=None, analyticsReporting=None, crossRegionReferences=None, permissionsBoundary=None, suppressTemplateIndentation=None):\n super().__init__(scope, id, architecture=architecture, description=description, env=env, stackName=stackName, tags=tags, notificationArns=notificationArns, synthesizer=synthesizer, terminationProtection=terminationProtection, analyticsReporting=analyticsReporting, crossRegionReferences=crossRegionReferences, permissionsBoundary=permissionsBoundary, suppressTemplateIndentation=suppressTemplateIndentation)\n\n lambda_.Function(self, \"Handler\",\n runtime=lambda_.Runtime.NODEJS_LATEST,\n handler=\"index.handler\",\n code=lambda_.Code.from_asset(path.join(__dirname, \"lambda-handler\")),\n architecture=architecture\n )\n\n# Beginning of the test suite\napp = App()\n\nIntegTest(app, \"DifferentArchitectures\",\n test_cases=[\n StackUnderTest(app, \"Stack1\",\n architecture=lambda_.Architecture.ARM_64\n ),\n StackUnderTest(app, \"Stack2\",\n architecture=lambda_.Architecture.X86_64\n )\n ]\n)\n```\n\nThis is all the instruction you need for the integration test runner to know\nwhich stacks to synthesize, deploy and destroy. But you may also need to\ncustomize the behavior of the runner by changing its parameters. For example:\n\n```python\napp = App()\n\nstack_under_test = Stack(app, \"StackUnderTest\")\n\nstack = Stack(app, \"stack\")\n\ntest_case = IntegTest(app, \"CustomizedDeploymentWorkflow\",\n test_cases=[stack_under_test],\n diff_assets=True,\n stack_update_workflow=True,\n cdk_command_options=CdkCommands(\n deploy=DeployCommand(\n args=DeployOptions(\n require_approval=RequireApproval.NEVER,\n json=True\n )\n ),\n destroy=DestroyCommand(\n args=DestroyOptions(\n force=True\n )\n )\n )\n)\n```\n\n### IntegTestCaseStack\n\nIn the majority of cases an integration test will contain a single `IntegTestCase`.\nBy default when you create an `IntegTest` an `IntegTestCase` is created for you\nand all of your test cases are registered to this `IntegTestCase`. The `IntegTestCase`\nand `IntegTestCaseStack` constructs are only needed when it is necessary to\ndefined different options for individual test cases.\n\nFor example, you might want to have one test case where `diffAssets` is enabled.\n\n```python\n# app: App\n# stack_under_test: Stack\n\ntest_case_with_assets = IntegTestCaseStack(app, \"TestCaseAssets\",\n diff_assets=True\n)\n\nIntegTest(app, \"Integ\", test_cases=[stack_under_test, test_case_with_assets])\n```\n\n## Assertions\n\nThis library also provides a utility to make assertions against the infrastructure that the integration test deploys.\n\nThere are two main scenarios in which assertions are created.\n\n* Part of an integration test using `integ-runner`\n\nIn this case you would create an integration test using the `IntegTest` construct and then make assertions using the `assert` property.\nYou should **not** utilize the assertion constructs directly, but should instead use the `methods` on `IntegTest.assertions`.\n\n```python\n# app: App\n# stack: Stack\n\n\ninteg = IntegTest(app, \"Integ\", test_cases=[stack])\ninteg.assertions.aws_api_call(\"S3\", \"getObject\")\n```\n\nBy default an assertions stack is automatically generated for you. You may however provide your own stack to use.\n\n```python\n# app: App\n# stack: Stack\n# assertion_stack: Stack\n\n\ninteg = IntegTest(app, \"Integ\", test_cases=[stack], assertion_stack=assertion_stack)\ninteg.assertions.aws_api_call(\"S3\", \"getObject\")\n```\n\n* Part of a normal CDK deployment\n\nIn this case you may be using assertions as part of a normal CDK deployment in order to make an assertion on the infrastructure\nbefore the deployment is considered successful. In this case you can utilize the assertions constructs directly.\n\n```python\n# my_app_stack: Stack\n\n\nAwsApiCall(my_app_stack, \"GetObject\",\n service=\"S3\",\n api=\"getObject\"\n)\n```\n\n### DeployAssert\n\nAssertions are created by using the `DeployAssert` construct. This construct creates it's own `Stack` separate from\nany stacks that you create as part of your integration tests. This `Stack` is treated differently from other stacks\nby the `integ-runner` tool. For example, this stack will not be diffed by the `integ-runner`.\n\n`DeployAssert` also provides utilities to register your own assertions.\n\n```python\n# my_custom_resource: CustomResource\n# stack: Stack\n# app: App\n\n\ninteg = IntegTest(app, \"Integ\", test_cases=[stack])\ninteg.assertions.expect(\"CustomAssertion\",\n ExpectedResult.object_like({\"foo\": \"bar\"}),\n ActualResult.from_custom_resource(my_custom_resource, \"data\"))\n```\n\nIn the above example an assertion is created that will trigger a user defined `CustomResource`\nand assert that the `data` attribute is equal to `{ foo: 'bar' }`.\n\n### API Calls\n\nA common method to retrieve the \"actual\" results to compare with what is expected is to make an\nAPI call to receive some data. This library does this by utilizing CloudFormation custom resources\nwhich means that CloudFormation will call out to a Lambda Function which will\nmake the API call.\n\n#### HttpApiCall\n\nUsing the `HttpApiCall` will use the\n[node-fetch](https://github.com/node-fetch/node-fetch) JavaScript library to\nmake the HTTP call.\n\nThis can be done by using the class directory (in the case of a normal deployment):\n\n```python\n# stack: Stack\n\n\nHttpApiCall(stack, \"MyAsssertion\",\n url=\"https://example-api.com/abc\"\n)\n```\n\nOr by using the `httpApiCall` method on `DeployAssert` (when writing integration tests):\n\n```python\n# app: App\n# stack: Stack\n\ninteg = IntegTest(app, \"Integ\",\n test_cases=[stack]\n)\ninteg.assertions.http_api_call(\"https://example-api.com/abc\")\n```\n\n#### AwsApiCall\n\nUsing the `AwsApiCall` construct will use the AWS JavaScript SDK to make the API call.\n\nThis can be done by using the class directory (in the case of a normal deployment):\n\n```python\n# stack: Stack\n\n\nAwsApiCall(stack, \"MyAssertion\",\n service=\"SQS\",\n api=\"receiveMessage\",\n parameters={\n \"QueueUrl\": \"url\"\n }\n)\n```\n\nOr by using the `awsApiCall` method on `DeployAssert` (when writing integration tests):\n\n```python\n# app: App\n# stack: Stack\n\ninteg = IntegTest(app, \"Integ\",\n test_cases=[stack]\n)\ninteg.assertions.aws_api_call(\"SQS\", \"receiveMessage\", {\n \"QueueUrl\": \"url\"\n})\n```\n\nYou must specify the `service` and the `api` when using The `AwsApiCall` construct.\nThe `service` is the name of an AWS service, in one of the following forms:\n\n* An AWS SDK for JavaScript v3 package name (`@aws-sdk/client-api-gateway`)\n* An AWS SDK for JavaScript v3 client name (`api-gateway`)\n* An AWS SDK for JavaScript v2 constructor name (`APIGateway`)\n* A lowercase AWS SDK for JavaScript v2 constructor name (`apigateway`)\n\nThe `api` is the name of an AWS API call, in one of the following forms:\n\n* An API call name as found in the API Reference documentation (`GetObject`)\n* The API call name starting with a lowercase letter (`getObject`)\n* The AWS SDK for JavaScript v3 command class name (`GetObjectCommand`)\n\nBy default, the `AwsApiCall` construct will automatically add the correct IAM policies\nto allow the Lambda function to make the API call. It does this based on the `service`\nand `api` that is provided. In the above example the service is `SQS` and the api is\n`receiveMessage` so it will create a policy with `Action: 'sqs:ReceiveMessage`.\n\nThere are some cases where the permissions do not exactly match the service/api call, for\nexample the S3 `listObjectsV2` api. In these cases it is possible to add the correct policy\nby accessing the `provider` object.\n\n```python\n# app: App\n# stack: Stack\n# integ: IntegTest\n\n\napi_call = integ.assertions.aws_api_call(\"S3\", \"listObjectsV2\", {\n \"Bucket\": \"mybucket\"\n})\n\napi_call.provider.add_to_role_policy({\n \"Effect\": \"Allow\",\n \"Action\": [\"s3:GetObject\", \"s3:ListBucket\"],\n \"Resource\": [\"*\"]\n})\n```\n\nWhen executing `waitForAssertion()`, it is necessary to add an IAM policy using `waiterProvider.addToRolePolicy()`.\nBecause `IApiCall` does not have a `waiterProvider` property, you need to cast it to `AwsApiCall`.\n\n```python\n# integ: IntegTest\n\n\napi_call = integ.assertions.aws_api_call(\"S3\", \"listObjectsV2\", {\n \"Bucket\": \"mybucket\"\n}).wait_for_assertions()\n\napi_call.waiter_provider.add_to_role_policy({\n \"Effect\": \"Allow\",\n \"Action\": [\"s3:GetObject\", \"s3:ListBucket\"],\n \"Resource\": [\"*\"]\n})\n```\n\nNote that addToRolePolicy() uses direct IAM JSON policy blobs, not a iam.PolicyStatement\nobject like you will see in the rest of the CDK.\n\n### EqualsAssertion\n\nThis library currently provides the ability to assert that two values are equal\nto one another by utilizing the `EqualsAssertion` class. This utilizes a Lambda\nbacked `CustomResource` which in tern uses the [Match](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.assertions.Match.html) utility from the\n[@aws-cdk/assertions](https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.assertions-readme.html) library.\n\n```python\n# app: App\n# stack: Stack\n# queue: sqs.Queue\n# fn: lambda.IFunction\n\n\ninteg = IntegTest(app, \"Integ\",\n test_cases=[stack]\n)\n\ninteg.assertions.invoke_function(\n function_name=fn.function_name,\n invocation_type=InvocationType.EVENT,\n payload=JSON.stringify({\"status\": \"OK\"})\n)\n\nmessage = integ.assertions.aws_api_call(\"SQS\", \"receiveMessage\", {\n \"QueueUrl\": queue.queue_url,\n \"WaitTimeSeconds\": 20\n})\n\nmessage.assert_at_path(\"Messages.0.Body\", ExpectedResult.object_like({\n \"request_context\": {\n \"condition\": \"Success\"\n },\n \"request_payload\": {\n \"status\": \"OK\"\n },\n \"response_context\": {\n \"status_code\": 200\n },\n \"response_payload\": \"success\"\n}))\n```\n\n#### Match\n\n`integ-tests` also provides a `Match` utility similar to the `@aws-cdk/assertions` module. `Match`\ncan be used to construct the `ExpectedResult`. While the utility is similar, only a subset of methods are currently available on the `Match` utility of this module: `arrayWith`, `objectLike`, `stringLikeRegexp` and `serializedJson`.\n\n```python\n# message: AwsApiCall\n\n\nmessage.expect(ExpectedResult.object_like({\n \"Messages\": Match.array_with([{\n \"Payload\": Match.serialized_json({\"key\": \"value\"})\n }, {\n \"Body\": {\n \"Values\": Match.array_with([{\"Asdf\": 3}]),\n \"Message\": Match.string_like_regexp(\"message\")\n }\n }\n ])\n}))\n```\n\n### Examples\n\n#### Invoke a Lambda Function\n\nIn this example there is a Lambda Function that is invoked and\nwe assert that the payload that is returned is equal to '200'.\n\n```python\n# lambda_function: lambda.IFunction\n# app: App\n\n\nstack = Stack(app, \"cdk-integ-lambda-bundling\")\n\ninteg = IntegTest(app, \"IntegTest\",\n test_cases=[stack]\n)\n\ninvoke = integ.assertions.invoke_function(\n function_name=lambda_function.function_name\n)\ninvoke.expect(ExpectedResult.object_like({\n \"Payload\": \"200\"\n}))\n```\n\nThe above example will by default create a CloudWatch log group that's never\nexpired. If you want to configure it with custom log retention days, you need\nto specify the `logRetention` property.\n\n```python\nimport aws_cdk.aws_logs as logs\n\n# lambda_function: lambda.IFunction\n# app: App\n\n\nstack = Stack(app, \"cdk-integ-lambda-bundling\")\n\ninteg = IntegTest(app, \"IntegTest\",\n test_cases=[stack]\n)\n\ninvoke = integ.assertions.invoke_function(\n function_name=lambda_function.function_name,\n log_retention=logs.RetentionDays.ONE_WEEK\n)\n```\n\n#### Make an AWS API Call\n\nIn this example there is a StepFunctions state machine that is executed\nand then we assert that the result of the execution is successful.\n\n```python\n# app: App\n# stack: Stack\n# sm: IStateMachine\n\n\ntest_case = IntegTest(app, \"IntegTest\",\n test_cases=[stack]\n)\n\n# Start an execution\nstart = test_case.assertions.aws_api_call(\"StepFunctions\", \"startExecution\", {\n \"state_machine_arn\": sm.state_machine_arn\n})\n\n# describe the results of the execution\ndescribe = test_case.assertions.aws_api_call(\"StepFunctions\", \"describeExecution\", {\n \"execution_arn\": start.get_att_string(\"executionArn\")\n})\n\n# assert the results\ndescribe.expect(ExpectedResult.object_like({\n \"status\": \"SUCCEEDED\"\n}))\n```\n\n#### Chain ApiCalls\n\nSometimes it may be necessary to chain API Calls. Since each API call is its own resource, all you\nneed to do is add a dependency between the calls. There is an helper method `next` that can be used.\n\n```python\n# integ: IntegTest\n\n\ninteg.assertions.aws_api_call(\"S3\", \"putObject\", {\n \"Bucket\": \"amzn-s3-demo-bucket\",\n \"Key\": \"my-key\",\n \"Body\": \"helloWorld\"\n}).next(integ.assertions.aws_api_call(\"S3\", \"getObject\", {\n \"Bucket\": \"amzn-s3-demo-bucket\",\n \"Key\": \"my-key\"\n}))\n```\n\n#### Wait for results\n\nA common use case when performing assertions is to wait for a condition to pass. Sometimes the thing\nthat you are asserting against is not done provisioning by the time the assertion runs. In these\ncases it is possible to run the assertion asynchronously by calling the `waitForAssertions()` method.\n\nTaking the example above of executing a StepFunctions state machine, depending on the complexity of\nthe state machine, it might take a while for it to complete.\n\n```python\n# app: App\n# stack: Stack\n# sm: IStateMachine\n\n\ntest_case = IntegTest(app, \"IntegTest\",\n test_cases=[stack]\n)\n\n# Start an execution\nstart = test_case.assertions.aws_api_call(\"StepFunctions\", \"startExecution\", {\n \"state_machine_arn\": sm.state_machine_arn\n})\n\n# describe the results of the execution\ndescribe = test_case.assertions.aws_api_call(\"StepFunctions\", \"describeExecution\", {\n \"execution_arn\": start.get_att_string(\"executionArn\")\n}).expect(ExpectedResult.object_like({\n \"status\": \"SUCCEEDED\"\n})).wait_for_assertions()\n```\n\nWhen you call `waitForAssertions()` the assertion provider will continuously make the `awsApiCall` until the\n`ExpectedResult` is met. You can also control the parameters for waiting, for example:\n\n```python\n# test_case: IntegTest\n# start: IApiCall\n\n\ndescribe = test_case.assertions.aws_api_call(\"StepFunctions\", \"describeExecution\", {\n \"execution_arn\": start.get_att_string(\"executionArn\")\n}).expect(ExpectedResult.object_like({\n \"status\": \"SUCCEEDED\"\n})).wait_for_assertions(\n total_timeout=Duration.minutes(5),\n interval=Duration.seconds(15),\n backoff_rate=3\n)\n```\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "CDK Integration Testing Constructs",
"version": "2.170.0a0",
"project_urls": {
"Homepage": "https://github.com/aws/aws-cdk",
"Source": "https://github.com/aws/aws-cdk.git"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "37987b6dcfc91c759a7fc2596a56f6ebc006cfc6f3a0c157cbc329c0347855e0",
"md5": "a2ea974d4694594579b37a7c48b5a38f",
"sha256": "b74c93da35599558c76cd0150f10864859612b36d49ffe9480654fe26db1bdc8"
},
"downloads": -1,
"filename": "aws_cdk.integ_tests_alpha-2.170.0a0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a2ea974d4694594579b37a7c48b5a38f",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "~=3.8",
"size": 720504,
"upload_time": "2024-11-22T04:42:06",
"upload_time_iso_8601": "2024-11-22T04:42:06.913297Z",
"url": "https://files.pythonhosted.org/packages/37/98/7b6dcfc91c759a7fc2596a56f6ebc006cfc6f3a0c157cbc329c0347855e0/aws_cdk.integ_tests_alpha-2.170.0a0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "50a572e70db5ad4e3a64769480bd1c7020b68c86104a1f1470b40e02f22983ba",
"md5": "4482d9074e16f4ece82fe792777afe23",
"sha256": "7247beb321ea1939d02ddfe97939edadd670d7155817b4ee62c9e98a67a429d9"
},
"downloads": -1,
"filename": "aws_cdk_integ_tests_alpha-2.170.0a0.tar.gz",
"has_sig": false,
"md5_digest": "4482d9074e16f4ece82fe792777afe23",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "~=3.8",
"size": 722451,
"upload_time": "2024-11-22T04:42:49",
"upload_time_iso_8601": "2024-11-22T04:42:49.630866Z",
"url": "https://files.pythonhosted.org/packages/50/a5/72e70db5ad4e3a64769480bd1c7020b68c86104a1f1470b40e02f22983ba/aws_cdk_integ_tests_alpha-2.170.0a0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-22 04:42:49",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "aws",
"github_project": "aws-cdk",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "aws-cdk.integ-tests-alpha"
}
JSON
