Kaiyang expansion project 2022

appengine/standard_python3/bundled-services/blobstore/django/main_test.py

@@ -17,38 +17,55 @@
 import subprocess
 import uuid
 
+import backoff
 import pytest
 import requests
 
 
-@pytest.fixture
-def version():
-    """Launch a new version of the app for testing, and yield the
-    project and version number so tests can invoke it, then delete it.
+@backoff.on_exception(backoff.expo, Exception, max_tries=3)
+def gcloud_cli(command):
     """
+    Runs the gcloud CLI with given options, parses the json formatted output
+    and returns the resulting Python object.
+
+    Usage: gcloud_cli(options)
+        options: command line options
+
+    Example:
+        result = gcloud_cli("app deploy --no-promote")
+        print(f"Deployed version {result['versions'][0]['id']}")
 
+    Raises Exception with the stderr output of the last attempt on failure.
+    """
     output = subprocess.run(
-        f"gcloud app deploy --no-promote --quiet --format=json --version={uuid.uuid4().hex}",
+        f"gcloud {command} --quiet --format=json",
         capture_output=True,
         shell=True,
+        check=True,
     )
-
     try:
-        result = json.loads(output.stdout)
-        version_id = result["versions"][0]["id"]
-        project_id = result["versions"][0]["project"]
-    except Exception as e:
-        print(f"New version deployment output not usable: {e}")
-        print(f"Command stderr is '{output.stderr}'")
-        raise ValueError
+        entries = json.loads(output.stdout)
+        return entries
+    except Exception:
+        print("Failed to read log")
+        print(f"gcloud stderr was {output.stderr}")
+
+    raise Exception(output.stderr)
+
+
+@pytest.fixture
+def version():
+    """Launch a new version of the app for testing, and yield the
+    project and version number so tests can invoke it, then delete it.
+    """
+
+    result = gcloud_cli(f"app deploy --no-promote --version={uuid.uuid4().hex}")
+    version_id = result["versions"][0]["id"]
+    project_id = result["versions"][0]["project"]
 
     yield project_id, version_id
 
-    output = subprocess.run(
-        f"gcloud app versions delete {version_id}",
-        capture_output=True,
-        shell=True,
-    )
+    gcloud_cli(f"app versions delete {version_id}")
 
 
 def test_upload_and_view(version):

appengine/standard_python3/bundled-services/blobstore/django/requirements-test.txt

@@ -1,2 +1,3 @@
+backoff==2.1.2
 pytest==7.1.2
 requests==2.28.1

appengine/standard_python3/bundled-services/blobstore/flask/main_test.py

@@ -17,42 +17,55 @@
 import subprocess
 import uuid
 
+import backoff
 import pytest
 import requests
 
 
-@pytest.fixture
-def version():
-    """Launch a new version of the app for testing, and yield the
-    project and version number so tests can invoke it, then delete it.
+@backoff.on_exception(backoff.expo, Exception, max_tries=3)
+def gcloud_cli(command):
     """
+    Runs the gcloud CLI with given options, parses the json formatted output
+    and returns the resulting Python object.
+
+    Usage: gcloud_cli(options)
+        options: command line options
 
-    debug_output = subprocess.run("gcloud config list", capture_output=True, shell=True)
-    print(f"Debug stdout of list is {debug_output.stdout}")
-    print(f"Debug stderr of list is {debug_output.stderr}")
+    Example:
+        result = gcloud_cli("app deploy --no-promote")
+        print(f"Deployed version {result['versions'][0]['id']}")
 
+    Raises Exception with the stderr output of the last attempt on failure.
+    """
     output = subprocess.run(
-        f"gcloud app deploy --no-promote --quiet --format=json --version={uuid.uuid4().hex}",
+        f"gcloud {command} --quiet --format=json",
         capture_output=True,
         shell=True,
+        check=True,
     )
-
     try:
-        result = json.loads(output.stdout)
-        version_id = result["versions"][0]["id"]
-        project_id = result["versions"][0]["project"]
-    except Exception as e:
-        print(f"New version deployment output not usable: {e}")
-        print(f"Command stderr is '{output.stderr}'")
-        raise ValueError
+        entries = json.loads(output.stdout)
+        return entries
+    except Exception:
+        print("Failed to read log")
+        print(f"gcloud stderr was {output.stderr}")
+
+    raise Exception(output.stderr)
+
+
+@pytest.fixture
+def version():
+    """Launch a new version of the app for testing, and yield the
+    project and version number so tests can invoke it, then delete it.
+    """
+
+    result = gcloud_cli(f"app deploy --no-promote --version={uuid.uuid4().hex}")
+    version_id = result["versions"][0]["id"]
+    project_id = result["versions"][0]["project"]
 
     yield project_id, version_id
 
-    output = subprocess.run(
-        f"gcloud app versions delete {version_id}",
-        capture_output=True,
-        shell=True,
-    )
+    gcloud_cli(f"app versions delete {version_id}")
 
 
 def test_upload_and_view(version):

appengine/standard_python3/bundled-services/blobstore/flask/requirements-test.txt

@@ -1,2 +1,3 @@
+backoff==2.1.2
 pytest==7.1.2
 requests==2.28.1

appengine/standard_python3/bundled-services/blobstore/wsgi/main_test.py

@@ -17,38 +17,55 @@
 import subprocess
 import uuid
 
+import backoff
 import pytest
 import requests
 
 
-@pytest.fixture
-def version():
-    """Launch a new version of the app for testing, and yield the
-    project and version number so tests can invoke it, then delete it.
+@backoff.on_exception(backoff.expo, Exception, max_tries=3)
+def gcloud_cli(command):
     """
+    Runs the gcloud CLI with given options, parses the json formatted output
+    and returns the resulting Python object.
+
+    Usage: gcloud_cli(options)
+        options: command line options
+
+    Example:
+        result = gcloud_cli("app deploy --no-promote")
+        print(f"Deployed version {result['versions'][0]['id']}")
 
+    Raises Exception with the stderr output of the last attempt on failure.
+    """
     output = subprocess.run(
-        f"gcloud app deploy --no-promote --quiet --format=json --version={uuid.uuid4().hex}",
+        f"gcloud {command} --quiet --format=json",
         capture_output=True,
         shell=True,
+        check=True,
     )
-
     try:
-        result = json.loads(output.stdout)
-        version_id = result["versions"][0]["id"]
-        project_id = result["versions"][0]["project"]
-    except Exception as e:
-        print(f"New version deployment output not usable: {e}")
-        print(f"Command stderr is '{output.stderr}'")
-        raise ValueError
+        entries = json.loads(output.stdout)
+        return entries
+    except Exception:
+        print("Failed to read log")
+        print(f"gcloud stderr was {output.stderr}")
+
+    raise Exception(output.stderr)
+
+
+@pytest.fixture
+def version():
+    """Launch a new version of the app for testing, and yield the
+    project and version number so tests can invoke it, then delete it.
+    """
+
+    result = gcloud_cli(f"app deploy --no-promote --version={uuid.uuid4().hex}")
+    version_id = result["versions"][0]["id"]
+    project_id = result["versions"][0]["project"]
 
     yield project_id, version_id
 
-    output = subprocess.run(
-        f"gcloud app versions delete {version_id}",
-        capture_output=True,
-        shell=True,
-    )
+    gcloud_cli(f"app versions delete {version_id}")
 
 
 def test_upload_and_view(version):

appengine/standard_python3/bundled-services/blobstore/wsgi/requirements-test.txt

@@ -1,2 +1,3 @@
+backoff==2.1.2
 pytest==7.1.2
 requests==2.28.1

appengine/standard_python3/bundled-services/deferred/django/main_test.py

@@ -17,38 +17,55 @@
 import time
 import uuid
 
+import backoff
 import pytest
 import requests
 
 
-@pytest.fixture
-def version():
-    """Launch a new version of the app for testing, and yield the
-    project and version number so tests can invoke it, then delete it.
+@backoff.on_exception(backoff.expo, Exception, max_tries=3)
+def gcloud_cli(command):
     """
+    Runs the gcloud CLI with given options, parses the json formatted output
+    and returns the resulting Python object.
+
+    Usage: gcloud_cli(options)
+        options: command line options
+
+    Example:
+        result = gcloud_cli("app deploy --no-promote")
+        print(f"Deployed version {result['versions'][0]['id']}")
 
+    Raises Exception with the stderr output of the last attempt on failure.
+    """
     output = subprocess.run(
-        f"gcloud app deploy --no-promote --quiet --format=json --version={uuid.uuid4().hex}",
+        f"gcloud {command} --quiet --format=json",
         capture_output=True,
         shell=True,
+        check=True,
     )
-
     try:
-        result = json.loads(output.stdout)
-        version_id = result["versions"][0]["id"]
-        project_id = result["versions"][0]["project"]
-    except Exception as e:
-        print(f"New version deployment output not usable: {e}")
-        print(f"Command stderr is '{output.stderr}'")
-        raise ValueError
+        entries = json.loads(output.stdout)
+        return entries
+    except Exception:
+        print("Failed to read log")
+        print(f"gcloud stderr was {output.stderr}")
+
+    raise Exception(output.stderr)
+
+
+@pytest.fixture
+def version():
+    """Launch a new version of the app for testing, and yield the
+    project and version number so tests can invoke it, then delete it.
+    """
+
+    result = gcloud_cli(f"app deploy --no-promote --version={uuid.uuid4().hex}")
+    version_id = result["versions"][0]["id"]
+    project_id = result["versions"][0]["project"]
 
     yield project_id, version_id
 
-    output = subprocess.run(
-        f"gcloud app versions delete {version_id}",
-        capture_output=True,
-        shell=True,
-    )
+    gcloud_cli(f"app versions delete {version_id}")
 
 
 def test_upload_and_view(version):

appengine/standard_python3/bundled-services/deferred/django/requirements-test.txt

@@ -1,2 +1,3 @@
+backoff==2.1.2
 pytest==7.1.2
 requests==2.28.1

appengine/standard_python3/bundled-services/deferred/flask/main_test.py

@@ -17,38 +17,55 @@
 import time
 import uuid
 
+import backoff
 import pytest
 import requests
 
 
-@pytest.fixture
-def version():
-    """Launch a new version of the app for testing, and yield the
-    project and version number so tests can invoke it, then delete it.
+@backoff.on_exception(backoff.expo, Exception, max_tries=3)
+def gcloud_cli(command):
     """
+    Runs the gcloud CLI with given options, parses the json formatted output
+    and returns the resulting Python object.
+
+    Usage: gcloud_cli(options)
+        options: command line options
+
+    Example:
+        result = gcloud_cli("app deploy --no-promote")
+        print(f"Deployed version {result['versions'][0]['id']}")
 
+    Raises Exception with the stderr output of the last attempt on failure.
+    """
     output = subprocess.run(
-        f"gcloud app deploy --no-promote --quiet --format=json --version={uuid.uuid4().hex}",
+        f"gcloud {command} --quiet --format=json",
         capture_output=True,
         shell=True,
+        check=True,
     )
-
     try:
-        result = json.loads(output.stdout)
-        version_id = result["versions"][0]["id"]
-        project_id = result["versions"][0]["project"]
-    except Exception as e:
-        print(f"New version deployment output not usable: {e}")
-        print(f"Command stderr is '{output.stderr}'")
-        raise ValueError
+        entries = json.loads(output.stdout)
+        return entries
+    except Exception:
+        print("Failed to read log")
+        print(f"gcloud stderr was {output.stderr}")
+
+    raise Exception(output.stderr)
+
+
+@pytest.fixture
+def version():
+    """Launch a new version of the app for testing, and yield the
+    project and version number so tests can invoke it, then delete it.
+    """
+
+    result = gcloud_cli(f"app deploy --no-promote --version={uuid.uuid4().hex}")
+    version_id = result["versions"][0]["id"]
+    project_id = result["versions"][0]["project"]
 
     yield project_id, version_id
 
-    output = subprocess.run(
-        f"gcloud app versions delete {version_id}",
-        capture_output=True,
-        shell=True,
-    )
+    gcloud_cli(f"app versions delete {version_id}")
 
 
 def test_upload_and_view(version):

appengine/standard_python3/bundled-services/deferred/flask/requirements-test.txt

@@ -1,2 +1,3 @@
+backoff==2.1.2
 pytest==7.1.2
 requests==2.28.1