Asset: Add VPCSC tests.

asset/noxfile.py

@@ -118,11 +118,21 @@ def system(session):
     session.install("-e", "../test_utils/")
     session.install("-e", ".")
 
+    # Additional setup for VPCSC system tests
+    env = {
+        "PROJECT_ID": "secure-gcp-test-project-4",
+        "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, *session.posargs)
+        session.run("py.test", "--quiet", system_test_path, env=env, *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, env=env, *session.posargs
+        )
 
 
 @nox.session(python="3.7")

asset/synth.py

@@ -84,6 +84,6 @@
 # Add templated files
 # ----------------------------------------------------------------------------
 templated_files = gcp.CommonTemplates().py_library(unit_cov_level=79, cov_level=80)
-s.move(templated_files)
+s.move(templated_files, excludes=["noxfile.py"])
 
 s.shell.run(["nox", "-s", "blacken"], hide_output=False)

asset/tests/system/test_vpcsc.py

@@ -0,0 +1,88 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright 2019 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# flake8: noqa
+
+import os
+import pytest
+
+from google.api_core import exceptions
+from google.cloud import asset_v1
+from google.cloud.asset_v1 import enums
+
+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")
+
+
+class TestVPCServiceControl(object):
+    @staticmethod
+    def _is_rejected(call):
+        try:
+            responses = call()
+        except exceptions.PermissionDenied as e:
+            return e.message == "Request is prohibited by organization's policy"
+        except:
+            pass
+        return False
+
+    @staticmethod
+    def _do_test(delayed_inside, delayed_outside):
+        if IS_INSIDE_VPCSC.lower() == "true":
+            assert TestVPCServiceControl._is_rejected(delayed_outside)
+            assert not (TestVPCServiceControl._is_rejected(delayed_inside))
+        else:
+            assert not (TestVPCServiceControl._is_rejected(delayed_outside))
+            assert TestVPCServiceControl._is_rejected(delayed_inside)
+
+    @pytest.mark.skipif(
+        PROJECT_INSIDE is None, reason="Missing environment variable: PROJECT_ID"
+    )
+    @pytest.mark.skipif(
+        PROJECT_OUTSIDE is None,
+        reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT",
+    )
+    def test_export_assets(self):
+        client = asset_v1.AssetServiceClient()
+        output_config = {}
+        parent_inside = "projects/" + PROJECT_INSIDE
+        delayed_inside = lambda: client.export_assets(parent_inside, output_config)
+        parent_outside = "projects/" + PROJECT_OUTSIDE
+        delayed_outside = lambda: client.export_assets(parent_outside, output_config)
+        TestVPCServiceControl._do_test(delayed_inside, delayed_outside)
+
+    @pytest.mark.skipif(
+        PROJECT_INSIDE is None, reason="Missing environment variable: PROJECT_ID"
+    )
+    @pytest.mark.skipif(
+        PROJECT_OUTSIDE is None,
+        reason="Missing environment variable: GOOGLE_CLOUD_TESTS_VPCSC_OUTSIDE_PERIMETER_PROJECT",
+    )
+    def test_batch_get_assets_history(self):
+        client = asset_v1.AssetServiceClient()
+        content_type = enums.ContentType.CONTENT_TYPE_UNSPECIFIED
+        read_time_window = {}
+        parent_inside = "projects/" + PROJECT_INSIDE
+        delayed_inside = lambda: client.batch_get_assets_history(
+            parent_inside, content_type, read_time_window
+        )
+        parent_outside = "projects/" + PROJECT_OUTSIDE
+        delayed_outside = lambda: client.batch_get_assets_history(
+            parent_outside, content_type, read_time_window
+        )
+        TestVPCServiceControl._do_test(delayed_inside, delayed_outside)