Goblet GCP Client

Goblet GCP Client is a util library with support for creating GCP resource clients, GCP integration tests, and other related utils.

Getting started

To install run:

pip install goblet-gcp-client

Using a 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")

Writing Integration Tests:

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.

Point to Emulator

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",
)

Features

GCP resource clients
GCP HTTP Test Recording and Replaying

Examples

Client and Testing Examples

Issues

Please file any issues, bugs or feature requests as an issue on our GitHub page.

Want to Contribute

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.

nthings / goblet_gcp_client Goto Github PK