Python
Latest SDK Version
Version 0.3+ of this SDK uses V2 of the Signadot API, in which some type and method names have changed. If you're upgrading from an older version, you'll need to update your code to match the new API.
Installation
The SDK can be added as a dependency to a Python project as follows:
python3 -m pip install signadot-sdk
Usage
Create Sandbox
Assume a hypothetical Route
service in the below example. This Route
service is changed and an image corresponding to the new version has been built and pushed to the registry. You wish to run integration tests against that new untested version.
# Initialize client
configuration = Configuration()
configuration.api_key['signadot-api-key'] = SIGNADOT_API_KEY
api_client = ApiClient(configuration)
route_fork = SandboxFork(
# A particular service that we want to fork, starting from the specified
# Kubernetes workload of kind "Deployment" with name "route" and
# namespace "hotrod". To fork an Argo Rollout, use kind "Rollout".
fork_of=SandboxForkOf(
kind="Deployment",
name="route",
namespace="hotrod"
),
# customizations to be applied on the fork
customizations=SandboxCustomizations(
images=[
# docker image that the fork should run
SandboxImage(image="signadot/hotrod-route:540fadfd2fe619e20b794d56ce404761ce2b45a3")
],
env=[
# any env vars to add / modify
SandboxEnvVar(name="abc", value="xyz")
]
)
)
# Define the default routegroup, along with the endpoints that we want to expose in order to run
# tests. Here, `http://route.hotrod.deploy:8083` defines it as an endpoint to HTTP port 8083 of
# the forked deployment (and hence `.deploy`) of the `route` service in the `hotrod` namespace.
default_route_group=SandboxDefaultRouteGroup(
endpoints=[
RouteGroupSpecEndpoint(name="hotrod-route", target="http://route.hotrod.deploy:8083")
]
)
# assemble the request to create a sandbox
request = Sandbox(
# pass sandbox spec
spec=SandboxSpec(
cluster="your-cluster",
description="route service test sandbox",
forks=[route_fork],
default_route_group=default_route_group
)
)
# submit the request to create sandbox via the API
sandboxes_api = SandboxesApi(api_client)
response = sandboxes_api.apply_sandbox("org-name", "sandbox-name", request)
You specify one or more forks
which are new versions of a particular Microservice. Each fork
starts with a particular baseline Deployment
running in the Kubernetes cluster and then can have one or more customizations
applied to it.
The customizations that are currently supported are as follows:
- docker images
- environment variables
Finally, create a request object by setting the sandbox spec comprising cluster name, description, forks and the default routegroup with the endpoint definitions.
Retrieve Endpoint
The final call above to apply_sandbox()
will create the sandbox within the cluster. You can use the response and retrieve the specified endpoint URLs to run tests.
sandbox_id = api_response.sandbox_id
# Filter the endpoint of interest from the list of endpoints in the response
endpoint = list(filter(lambda ep: ep.name == "hotrod-route", api_response.endpoints))[0]
endpoint_url = endpoint.url
Wait for Readiness
Before running tests, you need to ensure that the service has come up within the cluster and is ready to take requests.
while not response.status.ready:
time.sleep(5)
response = sandboxes_api.get_sandbox("org-name", "sandbox-name")
Run tests
Once the sandbox is ready, you can run tests on the forked service.
def test_valid_result(self):
pickup = "123"
dropoff = "456"
service_url = "{}/route?pickup={}&dropoff={}".format(self.endpoint_url, pickup, dropoff)
response = requests.get(service_url, headers=self.headers_dict)
response.raise_for_status()
data = response.json()
self.assertEqual(data["Pickup"], pickup, "Pickup must equal {}".format(pickup))
self.assertEqual(data["Dropoff"], dropoff, "Dropoff must equal {}".format(dropoff))
self.assertGreater(data["ETA"], -1, "ETA must be non-negative")
Delete Sandbox
Once all the tests are executed and have passed, you can delete the sandbox. This will tear down any resources that were set up such a pods, deployments, services etc from the Kubernetes cluster.
sandboxes_api.delete_sandbox("org-name", "sandbox-name")