From 87ba8852e2835e44aaf63183b7dad287b0b9a9af Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Wed, 6 Nov 2019 14:43:42 -0500 Subject: [PATCH 1/2] tests(translate): normalize VPCSC configuration in systests - Skip based on 'vpcsc_config.skip_unless_inside_vpcsc'. - Refactor VPCSC tests into one function per case, w/ expected exceptions. - Invoke client methods directly using 'vpcsc_config' to decode env vars. --- translate/tests/system/test_vpcsc.py | 245 ++++++++++++++------------- 1 file changed, 123 insertions(+), 122 deletions(-) diff --git a/translate/tests/system/test_vpcsc.py b/translate/tests/system/test_vpcsc.py index 427d0be6757d..63157be37b37 100644 --- a/translate/tests/system/test_vpcsc.py +++ b/translate/tests/system/test_vpcsc.py @@ -20,147 +20,148 @@ from google.api_core import exceptions from google.cloud import translate_v3beta1 +from test_utils.vpcsc_config import vpcsc_config +_VPCSC_PROHIBITED_MESSAGE = "Request is prohibited by organization's policy" -IS_INSIDE_VPCSC = "GOOGLE_CLOUD_TESTS_IN_VPCSC" in os.environ -# If IS_INSIDE_VPCSC is set, these environment variables should also be set -if IS_INSIDE_VPCSC: - PROJECT_INSIDE = os.environ["PROJECT_ID"] - PROJECT_OUTSIDE = os.environ["GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT"] +@pytest.fixture(scope="module") +def client(): + return translate_v3beta1.TranslationServiceClient() -class TestVPCServiceControl(object): - @classmethod - def setup(self): - self._client = translate_v3beta1.TranslationServiceClient() - self._parent_inside = self._client.location_path(PROJECT_INSIDE, "us-central1") - self._parent_outside = self._client.location_path( - PROJECT_OUTSIDE, "us-central1" - ) - def make_glossary_name(project_id): - return "projects/{0}/locations/us-central1/glossaries/fake_glossary".format( - project_id - ) - - self._glossary_name_inside = make_glossary_name(PROJECT_INSIDE) - self._glossary_name_outside = make_glossary_name(PROJECT_OUTSIDE) - - @staticmethod - def _is_rejected(call): - try: - responses = call() - print("responses: ", responses) - except exceptions.PermissionDenied as e: - print("PermissionDenied Exception: ", e) - return e.message == "Request is prohibited by organization's policy" - except Exception as e: - print("Other Exception: ", e) - pass - return False - - @staticmethod - def _do_test(delayed_inside, delayed_outside): - assert TestVPCServiceControl._is_rejected(delayed_outside) - assert not (TestVPCServiceControl._is_rejected(delayed_inside)) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test must be run in VPCSC. To enable this test, set the environment variable GOOGLE_CLOUD_TESTS_IN_VPCSC to True", - ) - def test_create_glossary(self): - def make_glossary(project_id): - return { - "name": "projects/{0}/locations/us-central1/glossaries/fake_glossary".format( - project_id - ), - "language_codes_set": {"language_codes": ["en", "ja"]}, - "input_config": { - "gcs_source": {"input_uri": "gs://fake-bucket/fake_glossary.csv"} - }, - } +@pytest.fixture(scope="module") +def parent_inside(client): + return client.location_path(vpcsc_config.project_inside, "us-central1") - glossary_inside = make_glossary(PROJECT_INSIDE) - def delayed_inside(): - return self._client.create_glossary(self._parent_inside, glossary_inside) +@pytest.fixture(scope="module") +def parent_outside(client): + return client.location_path(vpcsc_config.project_outside, "us-central1") - glossary_outside = make_glossary(PROJECT_OUTSIDE) - def delayed_outside(): - return self._client.create_glossary(self._parent_outside, glossary_outside) +@pytest.fixture(scope="module") +def glossary_name_inside(client): + return client.glossary_path( + vpcsc_config.project_inside, "us-central1", "fake_glossary" + ) - TestVPCServiceControl._do_test(delayed_inside, delayed_outside) - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test must be run in VPCSC. To enable this test, set the environment variable GOOGLE_CLOUD_TESTS_IN_VPCSC to True", +@pytest.fixture(scope="module") +def glossary_name_outside(client): + return client.glossary_path( + vpcsc_config.project_outside, "us-central1", "fake_glossary" ) - def test_list_glossaries(self): - # list_glossaries() returns an GRPCIterator instance, and we need to actually iterate through it - # by calling _next_page() to get real response. - def delayed_inside(): - return self._client.list_glossaries(self._parent_inside)._next_page() - def delayed_outside(): - return self._client.list_glossaries(self._parent_outside)._next_page() - TestVPCServiceControl._do_test(delayed_inside, delayed_outside) +def _make_glossary(name): + return { + "name": name, + "language_codes_set": {"language_codes": ["en", "ja"]}, + "input_config": { + "gcs_source": {"input_uri": "gs://fake-bucket/fake_glossary.csv"} + }, + } - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test must be run in VPCSC. To enable this test, set the environment variable GOOGLE_CLOUD_TESTS_IN_VPCSC to True", - ) - def test_get_glossary(self): - def delayed_inside(): - return self._client.get_glossary(self._glossary_name_inside) - def delayed_outside(): - return self._client.get_glossary(self._glossary_name_outside) +@pytest.fixture(scope="module") +def glossary_inside(glossary_name_inside): + return _make_glossary(glossary_name_inside) - TestVPCServiceControl._do_test(delayed_inside, delayed_outside) - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test must be run in VPCSC. To enable this test, set the environment variable GOOGLE_CLOUD_TESTS_IN_VPCSC to True", - ) - def test_delete_glossary(self): - def delayed_inside(): - return self._client.delete_glossary(self._glossary_name_inside) +@pytest.fixture(scope="module") +def glossary_outside(glossary_name_outside): + return _make_glossary(glossary_name_outside) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_create_glossary_w_inside(client, parent_inside, glossary_inside): + client.create_glossary(parent_inside, glossary_inside) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_create_glossary_w_outside(client, parent_outside, glossary_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.create_glossary(parent_outside, glossary_outside) + + assert exc.value.message.startswith(_VPCSC_PROHIBITED_MESSAGE) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_list_glossaries_w_inside(client, parent_inside): + list(client.list_glossaries(parent_inside)) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_list_glossaries_w_outside(client, parent_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + list(client.list_glossaries(parent_outside)) - def delayed_outside(): - return self._client.delete_glossary(self._glossary_name_outside) + assert exc.value.message.startswith(_VPCSC_PROHIBITED_MESSAGE) - TestVPCServiceControl._do_test(delayed_inside, delayed_outside) - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test must be run in VPCSC. To enable this test, set the environment variable GOOGLE_CLOUD_TESTS_IN_VPCSC to True", +@vpcsc_config.skip_unless_inside_vpcsc +def test_get_glossary_w_inside(client, glossary_name_inside): + try: + client.get_glossary(glossary_name_inside) + except exceptions.NotFound: # no perms issue + pass + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_get_glossary_w_outside(client, glossary_name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.get_glossary(glossary_name_outside) + + assert exc.value.message.startswith(_VPCSC_PROHIBITED_MESSAGE) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_delete_glossary_w_inside(client, glossary_name_inside): + try: + client.delete_glossary(glossary_name_inside) + except exceptions.NotFound: # no perms issue + pass + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_delete_glossary_w_outside(client, glossary_name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.delete_glossary(glossary_name_outside) + + assert exc.value.message.startswith(_VPCSC_PROHIBITED_MESSAGE) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_batch_translate_text_w_inside(client, parent_inside): + source_language_code = "en" + target_language_codes = ["es"] + input_configs = [{"gcs_source": {"input_uri": "gs://fake-bucket/*"}}] + output_config = { + "gcs_destination": {"output_uri_prefix": "gs://fake-bucket/output/"} + } + client.batch_translate_text( # no perms issue + parent_inside, + source_language_code, + target_language_codes, + input_configs, + output_config, ) - def test_batch_translate_text(self): - source_language_code = "en" - target_language_codes = ["es"] - input_configs = [{"gcs_source": {"input_uri": "gs://fake-bucket/*"}}] - output_config = { - "gcs_destination": {"output_uri_prefix": "gs://fake-bucket/output/"} - } - - def delayed_inside(): - return self._client.batch_translate_text( - self._parent_inside, - source_language_code, - target_language_codes, - input_configs, - output_config, - ) - - def delayed_outside(): - return self._client.batch_translate_text( - self._parent_outside, - source_language_code, - target_language_codes, - input_configs, - output_config, - ) - - TestVPCServiceControl._do_test(delayed_inside, delayed_outside) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_batch_translate_text_w_outside(client, parent_outside): + source_language_code = "en" + target_language_codes = ["es"] + input_configs = [{"gcs_source": {"input_uri": "gs://fake-bucket/*"}}] + output_config = { + "gcs_destination": {"output_uri_prefix": "gs://fake-bucket/output/"} + } + with pytest.raises(exceptions.PermissionDenied) as exc: + client.batch_translate_text( + parent_outside, + source_language_code, + target_language_codes, + input_configs, + output_config, + ) From 61c0c012ef05931daf006ac36953d05e2c93a414 Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Thu, 7 Nov 2019 16:03:35 -0500 Subject: [PATCH 2/2] fix: lint --- translate/tests/system/test_vpcsc.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/translate/tests/system/test_vpcsc.py b/translate/tests/system/test_vpcsc.py index 63157be37b37..814ab8759c13 100644 --- a/translate/tests/system/test_vpcsc.py +++ b/translate/tests/system/test_vpcsc.py @@ -15,7 +15,6 @@ # limitations under the License. """Unit tests for VPC-SC.""" -import os import pytest from google.api_core import exceptions @@ -165,3 +164,5 @@ def test_batch_translate_text_w_outside(client, parent_outside): input_configs, output_config, ) + + assert exc.value.message.startswith(_VPCSC_PROHIBITED_MESSAGE)