From 4127bd4d5181892ec52c3a3dae57df4c5bbddde0 Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Tue, 5 Nov 2019 17:53:21 -0500 Subject: [PATCH 1/5] tests(dlp): normalize VPCSC configuration in systests Toward #9580 --- dlp/noxfile.py | 16 +- .../v2/test_system_dlp_service_v2_vpcsc.py | 534 ++++++------------ 2 files changed, 179 insertions(+), 371 deletions(-) diff --git a/dlp/noxfile.py b/dlp/noxfile.py index 45ecc8fff050..5d56790447f4 100644 --- a/dlp/noxfile.py +++ b/dlp/noxfile.py @@ -118,24 +118,12 @@ def system(session): session.install("-e", "../test_utils/") session.install("-e", ".") - env = {} - # Additional setup for VPCSC system tests - if os.environ.get("GOOGLE_CLOUD_TESTS_IN_VPCSC", "false").lower() != "true": - # Unset PROJECT_ID, since VPCSC system tests expect this to be a project - # within the VPCSC perimeter. - env = { - "PROJECT_ID": "", - "GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT": os.environ.get( - "PROJECT_ID" - ), - } - # Run py.test against the system tests. if system_test_exists: - session.run("py.test", "--quiet", system_test_path, env=env, *session.posargs) + session.run("py.test", "--quiet", system_test_path, *session.posargs) if system_test_folder_exists: session.run( - "py.test", "--quiet", system_test_folder_path, env=env, *session.posargs + "py.test", "--quiet", system_test_folder_path, *session.posargs ) diff --git a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py index f075044d00e1..01fd0b1f9dc0 100644 --- a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py +++ b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py @@ -22,13 +22,7 @@ from google.cloud.dlp_v2.proto import dlp_pb2 from google.api_core import exceptions -PROJECT_INSIDE = os.environ.get("PROJECT_ID", None) -PROJECT_OUTSIDE = os.environ.get( - "GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", None -) -IS_INSIDE_VPCSC = ( - os.environ.get("GOOGLE_CLOUD_TESTS_IN_VPCSC", "false").lower() == "true" -) +from test_utils.vpcsc_config import vpcsc_config class TestSystemDlpService(object): @@ -42,480 +36,306 @@ def _is_rejected(call): pass return False - @staticmethod - def _do_test(delayed_inside, delayed_outside): - if IS_INSIDE_VPCSC: - assert TestSystemDlpService._is_rejected(delayed_outside) - assert not (TestSystemDlpService._is_rejected(delayed_inside)) + def _do_test(self, delayed_inside, delayed_outside): + if vpcsc_config.inside_vpcsc: + assert self._is_rejected(delayed_outside) + assert not (self._is_rejected(delayed_inside)) else: - assert not (TestSystemDlpService._is_rejected(delayed_outside)) - assert TestSystemDlpService._is_rejected(delayed_inside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + assert not (self._is_rejected(delayed_outside)) + assert self._is_rejected(delayed_inside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_inspect_content(self): + import pdb; pdb.set_trace() client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.inspect_content(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.inspect_content(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_redact_image(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.redact_image(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.redact_image(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_deidentify_content(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.deidentify_content(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.deidentify_content(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_reidentify_content(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.reidentify_content(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.reidentify_content(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_create_inspect_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.create_inspect_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.create_inspect_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_update_inspect_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.update_inspect_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.update_inspect_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_get_inspect_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.get_inspect_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.get_inspect_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) + self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_list_inspect_templates(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.list_inspect_templates(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.list_inspect_templates(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_delete_inspect_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.delete_inspect_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.delete_inspect_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_create_deidentify_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.create_deidentify_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.create_deidentify_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_update_deidentify_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.update_deidentify_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.update_deidentify_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_get_deidentify_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.get_deidentify_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.get_deidentify_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) + self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_list_deidentify_templates(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.list_deidentify_templates(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.list_deidentify_templates(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_delete_deidentify_template(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.delete_deidentify_template(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.delete_deidentify_template(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_create_dlp_job(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.create_dlp_job(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.create_dlp_job(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) + self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_list_dlp_jobs(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.list_dlp_jobs(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.list_dlp_jobs(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_get_dlp_job(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.get_dlp_job(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.get_dlp_job(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_delete_dlp_job(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.delete_dlp_job(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.delete_dlp_job(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_cancel_dlp_job(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.cancel_dlp_job(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.cancel_dlp_job(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) + self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_list_job_triggers(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.list_job_triggers(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.list_job_triggers(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_get_job_trigger(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.get_job_trigger(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.get_job_trigger(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_delete_job_trigger(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.delete_job_trigger(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.delete_job_trigger(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_update_job_trigger(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.update_job_trigger(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.update_job_trigger(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_create_job_trigger(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.create_job_trigger(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.create_job_trigger(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_create_stored_info_type(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.create_stored_info_type(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.create_stored_info_type(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_update_stored_info_type(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.update_stored_info_type(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.update_stored_info_type(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_get_stored_info_type(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.get_stored_info_type(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.get_stored_info_type(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) + self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_list_stored_info_types(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.list_stored_info_types(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.list_stored_info_types(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skipif( - not IS_INSIDE_VPCSC, - reason="This test requires a VPCSC and setting GOOGLE_CLOUD_TESTS_IN_VPCSC", - ) - @pytest.mark.skipif( - PROJECT_OUTSIDE is None, - reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT", - ) + self._do_test(delayed_inside, delayed_outside) + + @vpcsc_config.skip_unless_inside_vpcsc + @vpcsc_config.skip_unless_outside_project def test_delete_stored_info_type(self): client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(PROJECT_INSIDE) + name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.delete_stored_info_type(name_inside) - name_outside = client.project_path(PROJECT_OUTSIDE) + name_outside = client.project_path(vpcsc_config.project_outside) delayed_outside = lambda: client.delete_stored_info_type(name_outside) - TestSystemDlpService._do_test(delayed_inside, delayed_outside) + self._do_test(delayed_inside, delayed_outside) From 0584705997839fe19a1a7408fa581ff9d567ccb8 Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Wed, 6 Nov 2019 11:03:02 -0500 Subject: [PATCH 2/5] Remove debugging pdb. --- dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py | 1 - 1 file changed, 1 deletion(-) diff --git a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py index 01fd0b1f9dc0..731e274787ef 100644 --- a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py +++ b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py @@ -47,7 +47,6 @@ def _do_test(self, delayed_inside, delayed_outside): @vpcsc_config.skip_unless_inside_vpcsc @vpcsc_config.skip_unless_outside_project def test_inspect_content(self): - import pdb; pdb.set_trace() client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) delayed_inside = lambda: client.inspect_content(name_inside) From ca16b96a39dd1dc62cd9247d98e209131796db8c Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Wed, 6 Nov 2019 11:42:44 -0500 Subject: [PATCH 3/5] test: all these tests must run inside VPCSC - Move 'skip_unless_inside_vpcsc' to class scope. - Drop 'skip_unless_outside_project': it is guaranteed to be set when running tests inside VPCSC. - Drop never-reached branch for outside in '_do_test'. - Don't swallow unexpected exceptions inside '_is_rejected. --- .../v2/test_system_dlp_service_v2_vpcsc.py | 71 ++----------------- 1 file changed, 4 insertions(+), 67 deletions(-) diff --git a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py index 731e274787ef..c3e57b9d70b5 100644 --- a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py +++ b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py @@ -25,6 +25,7 @@ from test_utils.vpcsc_config import vpcsc_config +@vpcsc_config.skip_unless_inside_vpcsc class TestSystemDlpService(object): @staticmethod def _is_rejected(call): @@ -32,20 +33,12 @@ def _is_rejected(call): responses = call() except exceptions.PermissionDenied as e: return e.message == "Request is prohibited by organization's policy" - except: - pass return False def _do_test(self, delayed_inside, delayed_outside): - if vpcsc_config.inside_vpcsc: - assert self._is_rejected(delayed_outside) - assert not (self._is_rejected(delayed_inside)) - else: - assert not (self._is_rejected(delayed_outside)) - assert self._is_rejected(delayed_inside) - - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project + assert self._is_rejected(delayed_outside) + assert not (self._is_rejected(delayed_inside)) + def test_inspect_content(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -54,8 +47,6 @@ def test_inspect_content(self): delayed_outside = lambda: client.inspect_content(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_redact_image(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -64,8 +55,6 @@ def test_redact_image(self): delayed_outside = lambda: client.redact_image(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_deidentify_content(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -74,8 +63,6 @@ def test_deidentify_content(self): delayed_outside = lambda: client.deidentify_content(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_reidentify_content(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -84,8 +71,6 @@ def test_reidentify_content(self): delayed_outside = lambda: client.reidentify_content(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_create_inspect_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -94,8 +79,6 @@ def test_create_inspect_template(self): delayed_outside = lambda: client.create_inspect_template(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_update_inspect_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -104,8 +87,6 @@ def test_update_inspect_template(self): delayed_outside = lambda: client.update_inspect_template(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_get_inspect_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -115,8 +96,6 @@ def test_get_inspect_template(self): self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_list_inspect_templates(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -125,8 +104,6 @@ def test_list_inspect_templates(self): delayed_outside = lambda: client.list_inspect_templates(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_delete_inspect_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -135,8 +112,6 @@ def test_delete_inspect_template(self): delayed_outside = lambda: client.delete_inspect_template(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_create_deidentify_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -145,8 +120,6 @@ def test_create_deidentify_template(self): delayed_outside = lambda: client.create_deidentify_template(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_update_deidentify_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -155,8 +128,6 @@ def test_update_deidentify_template(self): delayed_outside = lambda: client.update_deidentify_template(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_get_deidentify_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -166,8 +137,6 @@ def test_get_deidentify_template(self): self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_list_deidentify_templates(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -176,8 +145,6 @@ def test_list_deidentify_templates(self): delayed_outside = lambda: client.list_deidentify_templates(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_delete_deidentify_template(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -186,8 +153,6 @@ def test_delete_deidentify_template(self): delayed_outside = lambda: client.delete_deidentify_template(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_create_dlp_job(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -197,8 +162,6 @@ def test_create_dlp_job(self): self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_list_dlp_jobs(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -207,8 +170,6 @@ def test_list_dlp_jobs(self): delayed_outside = lambda: client.list_dlp_jobs(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_get_dlp_job(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -217,8 +178,6 @@ def test_get_dlp_job(self): delayed_outside = lambda: client.get_dlp_job(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_delete_dlp_job(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -227,8 +186,6 @@ def test_delete_dlp_job(self): delayed_outside = lambda: client.delete_dlp_job(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_cancel_dlp_job(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -238,8 +195,6 @@ def test_cancel_dlp_job(self): self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_list_job_triggers(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -248,8 +203,6 @@ def test_list_job_triggers(self): delayed_outside = lambda: client.list_job_triggers(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_get_job_trigger(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -258,8 +211,6 @@ def test_get_job_trigger(self): delayed_outside = lambda: client.get_job_trigger(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_delete_job_trigger(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -268,8 +219,6 @@ def test_delete_job_trigger(self): delayed_outside = lambda: client.delete_job_trigger(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_update_job_trigger(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -278,8 +227,6 @@ def test_update_job_trigger(self): delayed_outside = lambda: client.update_job_trigger(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_create_job_trigger(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -288,8 +235,6 @@ def test_create_job_trigger(self): delayed_outside = lambda: client.create_job_trigger(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_create_stored_info_type(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -298,8 +243,6 @@ def test_create_stored_info_type(self): delayed_outside = lambda: client.create_stored_info_type(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_update_stored_info_type(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -308,8 +251,6 @@ def test_update_stored_info_type(self): delayed_outside = lambda: client.update_stored_info_type(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_get_stored_info_type(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -319,8 +260,6 @@ def test_get_stored_info_type(self): self._do_test(delayed_inside, delayed_outside) @pytest.mark.skip(reason="List tests are currently not supported") - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_list_stored_info_types(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) @@ -329,8 +268,6 @@ def test_list_stored_info_types(self): delayed_outside = lambda: client.list_stored_info_types(name_outside) self._do_test(delayed_inside, delayed_outside) - @vpcsc_config.skip_unless_inside_vpcsc - @vpcsc_config.skip_unless_outside_project def test_delete_stored_info_type(self): client = dlp_v2.DlpServiceClient() name_inside = client.project_path(vpcsc_config.project_inside) From 56b8b42d426382a5a0b261fa8bf905975f60cdf0 Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Wed, 6 Nov 2019 12:15:18 -0500 Subject: [PATCH 4/5] fix: blacken noxfile.py --- dlp/noxfile.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/dlp/noxfile.py b/dlp/noxfile.py index 5d56790447f4..a2eefbb6765f 100644 --- a/dlp/noxfile.py +++ b/dlp/noxfile.py @@ -122,9 +122,7 @@ def system(session): if system_test_exists: session.run("py.test", "--quiet", system_test_path, *session.posargs) if system_test_folder_exists: - session.run( - "py.test", "--quiet", system_test_folder_path, *session.posargs - ) + session.run("py.test", "--quiet", system_test_folder_path, *session.posargs) @nox.session(python="3.7") From 02ccdc72c4a6599e879ab04e15806865933dbe88 Mon Sep 17 00:00:00 2001 From: Tres Seaver Date: Mon, 11 Nov 2019 17:08:47 -0500 Subject: [PATCH 5/5] refactor: use one func per testcase --- .../v2/test_system_dlp_service_v2_vpcsc.py | 720 ++++++++++++------ 1 file changed, 471 insertions(+), 249 deletions(-) diff --git a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py index c3e57b9d70b5..289e927dd505 100644 --- a/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py +++ b/dlp/tests/system/gapic/v2/test_system_dlp_service_v2_vpcsc.py @@ -25,253 +25,475 @@ from test_utils.vpcsc_config import vpcsc_config +_VPCSC_PROHIBITED_MESSAGE = "Request is prohibited by organization's policy" + + +@pytest.fixture(scope="module") +def client(): + return dlp_v2.DlpServiceClient() + + +@pytest.fixture(scope="module") +def name_inside(client): + return client.project_path(vpcsc_config.project_inside) + + +@pytest.fixture(scope="module") +def name_outside(client): + return client.project_path(vpcsc_config.project_outside) + + +@pytest.fixture(scope="module") +def content_item(): + return dlp_pb2.ContentItem(value="testing") + + +@pytest.fixture(scope="module") +def bytes_content_item(): + return dlp_pb2.ByteContentItem(data=b"DEADBEEF") + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_inspect_content_inside(client, name_inside, content_item): + client.inspect_content(name_inside, item=content_item) # no perms issue + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_inspect_content_outside(client, name_outside, content_item): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.inspect_content(name_outside, item=content_item) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_redact_image_inside(client, name_inside, bytes_content_item): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.redact_image(name_inside, byte_item=bytes_content_item) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_redact_image_outside(client, name_outside, bytes_content_item): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.redact_image(name_outside, byte_item=bytes_content_item) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_deidentify_content_inside(client, name_inside, content_item): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.deidentify_content(name_inside, item=content_item) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_deidentify_content_outside(client, name_outside, content_item): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.deidentify_content(name_outside, item=content_item) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_reidentify_content_inside(client, name_inside, content_item): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.reidentify_content(name_inside, item=content_item) + + +@vpcsc_config.skip_unless_inside_vpcsc +def test_reidentify_content_outside(client, name_outside, content_item): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.reidentify_content(name_outside, item=content_item) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@pytest.fixture(scope="module") +def inspect_template_path_inside(client): + inspect_template_id = 1234567 + return client.project_inspect_template_path( + vpcsc_config.project_inside, inspect_template_id + ) + + +@pytest.fixture(scope="module") +def inspect_template_path_outside(client): + inspect_template_id = 1234567 + return client.project_inspect_template_path( + vpcsc_config.project_outside, inspect_template_id + ) + + +@vpcsc_config.skip_unless_inside_vpcsc +class TestCRUDInspectTemplate(object): + @staticmethod + def test_create_inspect_template_inside(client, name_inside): + client.create_inspect_template(name_inside) # no perms issue + + @staticmethod + def test_create_inspect_template_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.create_inspect_template(name_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_list_inspect_templates_inside(client, name_inside): + list(client.list_inspect_templates(name_inside)) + + @staticmethod + def test_list_inspect_templates_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + list(client.list_inspect_templates(name_outside)) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_update_inspect_template_inside(client, inspect_template_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.update_inspect_template(inspect_template_path_inside) + + @staticmethod + def test_update_inspect_template_outside(client, inspect_template_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.update_inspect_template(inspect_template_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_get_inspect_template_inside(client, inspect_template_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.get_inspect_template(inspect_template_path_inside) + + @staticmethod + def test_get_inspect_template_outside(client, inspect_template_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.get_inspect_template(inspect_template_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_delete_inspect_template_inside(client, inspect_template_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.delete_inspect_template(inspect_template_path_inside) + + @staticmethod + def test_delete_inspect_template_outside(client, inspect_template_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.delete_inspect_template(inspect_template_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@pytest.fixture(scope="module") +def deidentify_template_path_inside(client): + deidentify_template_id = 1234567 + return client.project_deidentify_template_path( + vpcsc_config.project_inside, deidentify_template_id + ) + + +@pytest.fixture(scope="module") +def deidentify_template_path_outside(client): + deidentify_template_id = 1234567 + return client.project_deidentify_template_path( + vpcsc_config.project_outside, deidentify_template_id + ) + + @vpcsc_config.skip_unless_inside_vpcsc -class TestSystemDlpService(object): - @staticmethod - def _is_rejected(call): - try: - responses = call() - except exceptions.PermissionDenied as e: - return e.message == "Request is prohibited by organization's policy" - return False - - def _do_test(self, delayed_inside, delayed_outside): - assert self._is_rejected(delayed_outside) - assert not (self._is_rejected(delayed_inside)) - - def test_inspect_content(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.inspect_content(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.inspect_content(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_redact_image(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.redact_image(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.redact_image(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_deidentify_content(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.deidentify_content(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.deidentify_content(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_reidentify_content(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.reidentify_content(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.reidentify_content(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_create_inspect_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.create_inspect_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.create_inspect_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_update_inspect_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.update_inspect_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.update_inspect_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_get_inspect_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.get_inspect_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.get_inspect_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skip(reason="List tests are currently not supported") - def test_list_inspect_templates(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.list_inspect_templates(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.list_inspect_templates(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_delete_inspect_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.delete_inspect_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.delete_inspect_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_create_deidentify_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.create_deidentify_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.create_deidentify_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_update_deidentify_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.update_deidentify_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.update_deidentify_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_get_deidentify_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.get_deidentify_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.get_deidentify_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skip(reason="List tests are currently not supported") - def test_list_deidentify_templates(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.list_deidentify_templates(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.list_deidentify_templates(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_delete_deidentify_template(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.delete_deidentify_template(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.delete_deidentify_template(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_create_dlp_job(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.create_dlp_job(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.create_dlp_job(name_outside) - self._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skip(reason="List tests are currently not supported") - def test_list_dlp_jobs(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.list_dlp_jobs(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.list_dlp_jobs(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_get_dlp_job(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.get_dlp_job(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.get_dlp_job(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_delete_dlp_job(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.delete_dlp_job(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.delete_dlp_job(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_cancel_dlp_job(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.cancel_dlp_job(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.cancel_dlp_job(name_outside) - self._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skip(reason="List tests are currently not supported") - def test_list_job_triggers(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.list_job_triggers(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.list_job_triggers(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_get_job_trigger(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.get_job_trigger(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.get_job_trigger(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_delete_job_trigger(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.delete_job_trigger(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.delete_job_trigger(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_update_job_trigger(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.update_job_trigger(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.update_job_trigger(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_create_job_trigger(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.create_job_trigger(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.create_job_trigger(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_create_stored_info_type(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.create_stored_info_type(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.create_stored_info_type(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_update_stored_info_type(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.update_stored_info_type(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.update_stored_info_type(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_get_stored_info_type(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.get_stored_info_type(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.get_stored_info_type(name_outside) - self._do_test(delayed_inside, delayed_outside) - - @pytest.mark.skip(reason="List tests are currently not supported") - def test_list_stored_info_types(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.list_stored_info_types(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.list_stored_info_types(name_outside) - self._do_test(delayed_inside, delayed_outside) - - def test_delete_stored_info_type(self): - client = dlp_v2.DlpServiceClient() - name_inside = client.project_path(vpcsc_config.project_inside) - delayed_inside = lambda: client.delete_stored_info_type(name_inside) - name_outside = client.project_path(vpcsc_config.project_outside) - delayed_outside = lambda: client.delete_stored_info_type(name_outside) - self._do_test(delayed_inside, delayed_outside) +class TestCRUDDeidentifyTemplate(object): + @staticmethod + def test_create_deidentify_template_inside(client, name_inside): + client.create_deidentify_template(name_inside) + + @staticmethod + def test_create_deidentify_template_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.create_deidentify_template(name_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_list_deidentify_templates_inside(client, name_inside): + list(client.list_deidentify_templates(name_inside)) + + @staticmethod + def test_list_deidentify_templates_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + list(client.list_deidentify_templates(name_outside)) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_update_deidentify_template_inside(client, deidentify_template_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.update_deidentify_template(deidentify_template_path_inside) + + @staticmethod + def test_update_deidentify_template_outside( + client, deidentify_template_path_outside + ): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.update_deidentify_template(deidentify_template_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_get_deidentify_template_inside(client, deidentify_template_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.get_deidentify_template(deidentify_template_path_inside) + + @staticmethod + def test_get_deidentify_template_outside(client, deidentify_template_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.get_deidentify_template(deidentify_template_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_delete_deidentify_template_inside(client, deidentify_template_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.delete_deidentify_template(deidentify_template_path_inside) + + @staticmethod + def test_delete_deidentify_template_outside( + client, deidentify_template_path_outside + ): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.delete_deidentify_template(deidentify_template_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@pytest.fixture(scope="module") +def job_path_inside(name_inside): + job_id = 1234567 + return "{}/jobs/{}".format(name_inside, job_id) + + +@pytest.fixture(scope="module") +def job_path_outside(name_outside): + job_id = 1234567 + return "{}/jobs/{}".format(name_outside, job_id) + + +@pytest.fixture(scope="module") +def inspect_job(): + from google.cloud.dlp_v2.proto.dlp_pb2 import InspectJobConfig + + return InspectJobConfig() + + +@vpcsc_config.skip_unless_inside_vpcsc +class TestCRUDDlpJob(object): + @staticmethod + def test_create_dlp_job_inside(client, name_inside, inspect_job): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.create_dlp_job(name_inside, inspect_job=inspect_job) + + @staticmethod + def test_create_dlp_job_outside(client, name_outside, inspect_job): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.create_dlp_job(name_outside, inspect_job=inspect_job) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_list_dlp_jobs_inside(client, name_inside): + list(client.list_dlp_jobs(name_inside)) + + @staticmethod + def test_list_dlp_jobs_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + list(client.list_dlp_jobs(name_outside)) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_get_dlp_job_inside(client, job_path_inside): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.get_dlp_job(job_path_inside) + + @staticmethod + def test_get_dlp_job_outside(client, job_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.get_dlp_job(job_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_delete_dlp_job_inside(client, job_path_inside): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.delete_dlp_job(job_path_inside) + + @staticmethod + def test_delete_dlp_job_outside(client, job_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.delete_dlp_job(job_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_cancel_dlp_job_inside(client, job_path_inside): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.cancel_dlp_job(job_path_inside) + + @staticmethod + def test_cancel_dlp_job_outside(client, job_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.cancel_dlp_job(job_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@pytest.fixture(scope="module") +def job_trigger_path_inside(client): + job_trigger_id = 1234567 + return client.project_job_trigger_path(vpcsc_config.project_inside, job_trigger_id) + + +@pytest.fixture(scope="module") +def job_trigger_path_outside(client): + job_trigger_id = 1234567 + return client.project_job_trigger_path(vpcsc_config.project_outside, job_trigger_id) + + +@vpcsc_config.skip_unless_inside_vpcsc +class TestCRUDJobTrigger(object): + @staticmethod + def test_create_job_trigger_inside(client, name_inside): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.create_job_trigger(name_inside) + + @staticmethod + def test_create_job_trigger_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.create_job_trigger(name_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_list_job_triggers_inside(client, name_inside): + list(client.list_job_triggers(name_inside)) + + @staticmethod + def test_list_job_triggers_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + list(client.list_job_triggers(name_outside)) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_get_job_trigger_inside(client, job_trigger_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.get_job_trigger(job_trigger_path_inside) + + @staticmethod + def test_get_job_trigger_outside(client, job_trigger_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.get_job_trigger(job_trigger_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_update_job_trigger_inside(client, job_trigger_path_inside): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.update_job_trigger(job_trigger_path_inside) + + @staticmethod + def test_update_job_trigger_outside(client, job_trigger_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.update_job_trigger(job_trigger_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_delete_job_trigger_inside(client, job_trigger_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.delete_job_trigger(job_trigger_path_inside) + + @staticmethod + def test_delete_job_trigger_outside(client, job_trigger_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.delete_job_trigger(job_trigger_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + +@pytest.fixture(scope="module") +def stored_info_type_path_inside(client): + stored_info_type_id = 1234567 + return client.project_stored_info_type_path( + vpcsc_config.project_inside, stored_info_type_id + ) + + +@pytest.fixture(scope="module") +def stored_info_type_path_outside(client): + stored_info_type_id = 1234567 + return client.project_stored_info_type_path( + vpcsc_config.project_outside, stored_info_type_id + ) + + +@vpcsc_config.skip_unless_inside_vpcsc +class TestCRUDStoredInfoType(object): + @staticmethod + def test_create_stored_info_type_inside(client, name_inside): + with pytest.raises(exceptions.InvalidArgument): # no perms issue + client.create_stored_info_type(name_inside) + + @staticmethod + def test_create_stored_info_type_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.create_stored_info_type(name_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_list_stored_info_types_inside(client, name_inside): + list(client.list_stored_info_types(name_inside)) + + @staticmethod + def test_list_stored_info_types_outside(client, name_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + list(client.list_stored_info_types(name_outside)) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_get_stored_info_type_inside(client, stored_info_type_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.get_stored_info_type(stored_info_type_path_inside) + + @staticmethod + def test_get_stored_info_type_outside(client, stored_info_type_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.get_stored_info_type(stored_info_type_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_update_stored_info_type_inside(client, stored_info_type_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.update_stored_info_type(stored_info_type_path_inside) + + @staticmethod + def test_update_stored_info_type_outside(client, stored_info_type_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.update_stored_info_type(stored_info_type_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message + + @staticmethod + def test_delete_stored_info_type_inside(client, stored_info_type_path_inside): + with pytest.raises(exceptions.NotFound): # no perms issue + client.delete_stored_info_type(stored_info_type_path_inside) + + @staticmethod + def test_delete_stored_info_type_outside(client, stored_info_type_path_outside): + with pytest.raises(exceptions.PermissionDenied) as exc: + client.delete_stored_info_type(stored_info_type_path_outside) + + assert _VPCSC_PROHIBITED_MESSAGE in exc.value.message