Goblet GCP Client is a util library with support for creating GCP resource clients, GCP integration tests, and other related utils.
To install run:
pip install goblet-gcp-client
from goblet_gcp_client import Client
cloudfunction_client = Client(
"cloudfunctions",
"v1",
calls="projects.locations.functions",
parent_schema="projects/{project_id}/locations/{location_id}",
)
scheduler_client = Client(
"cloudscheduler",
"v1",
calls="projects.locations.jobs",
parent_schema="projects/{project_id}/locations/{location_id}",
)
# override default parent_key with "name"
cloudfunction_client.execute(
"get", parent_key="name", parent_schema="projects/{project_id}/locations/{location_id}/functions/{name}"
)
# pass in body params
sample_body = {}
resp = cloudfunction_client.execute(
"patch",
parent_key="name",
parent_schema="projects/{project_id}/locations/{location_id}/functions/{name}",
params={"body": sample_body},
)
# Wait for previous operation to complete
cloudfunction_client.wait_for_operation(resp["name"], calls="operations")
Write your test and set env variable G_HTTP_TEST
to RECORD
By default responses will be written to the /tests/data
folder. You can customize this by setting G_TEST_DATA_DIR
env variable.
Setting the G_MOCK_CREDENTIALS
environment variable will use AnonymousCredentials.
def TestDeploy(self):
monkeypatch.setenv("G_HTTP_TEST", "RECORD")
monkeypatch.setenv("G_TEST_NAME", "TEST_NAME")
cloudfunction_client = Client(
"cloudfunctions",
"v1",
calls="projects.locations.functions",
parent_schema="projects/{project_id}/locations/{location_id}",
)
cloudfunction_client.execute(
"get", parent_key="name", parent_schema="projects/{project_id}/locations/{location_id}/functions/{name}"
)
Running your test will record all responses that your Client makes
Now you can run your tests with G_HTTP_TEST
with REPLAY
. You can access the responses with get_responses
or get_response
from goblet_gcp_client import get_responses, get_response, get_replay_count, reset_replay_count
def TestDeploy(self):
monkeypatch.setenv("G_HTTP_TEST", "REPLAY")
monkeypatch.setenv("G_TEST_NAME", "TEST_NAME")
reset_replay_count()
cloudfunction_client = Client(
"cloudfunctions",
"v1",
calls="projects.locations.functions",
parent_schema="projects/{project_id}/locations/{location_id}",
)
cloudfunction_client.execute(
"get", parent_key="name", parent_schema="projects/{project_id}/locations/{location_id}/functions/{name}"
)
responses = get_responses("TEST_NAME")
assert len(responses) == 2
assert "test_value" in responses[0]["body"]
assert get_replay_count() == 1
You can assert that your tests are making all required api called by using get_replay_count
. If you are running multiple tests you can run reset_replay_count
before each test to ensure the counts are accurate.
You can point to an emulator by setting the G_EMULATOR_HOST
environment variable, or specifying in the Client constructor.
spanner_instances_client = Client(
resource="spanner",
version="v1",
calls="projects.instances"
emulator_host="localhost:9020",
)
- GCP resource clients
- GCP HTTP Test Recording and Replaying
Please file any issues, bugs or feature requests as an issue on our GitHub page.
If you would like to contribute to the library (e.g. by improving the documentation, solving a bug or adding a cool new feature) submit a pull request.