add updated logic to resolve assets.json and pass along to record|playback/start

Author: scbedd

.gitignore

@@ -34,6 +34,7 @@ build/
 # Test results
 TestResults/
 ENV_DIR/
+.assets/
 
 # Perf test profiling
 cProfile-*.pstats

tools/azure-sdk-tools/devtools_testutils/proxy_startup.py

@@ -31,6 +31,7 @@
 PROXY_CHECK_URL = PROXY_URL.rstrip("/") + "/Info/Available"
 TOOL_ENV_VAR = "PROXY_PID"
 
+discovered_roots = []
 
 def get_image_tag(repo_root: str) -> str:
     """Gets the test proxy Docker image tag from the target_version.txt file in /eng/common/testproxy"""
@@ -59,6 +60,8 @@ def ascend_to_root(start_dir_or_file: str) -> str:
 
         # we need the git check to prevent ascending out of the repo
         if os.path.exists(possible_root):
+            if current_dir not in discovered_roots:
+                discovered_roots.append(current_dir)
             return current_dir
         else:
             current_dir = os.path.dirname(current_dir)
@@ -137,13 +140,18 @@ def start_test_proxy(request) -> None:
             else:
                 envname = os.getenv("TOX_ENV_NAME", "default")
                 root = os.getenv("BUILD_SOURCESDIRECTORY", repo_root)
+
+
+
                 log = open(os.path.join(root, "_proxy_log_{}.log".format(envname)), "a")
+                env = {}
 
                 _LOGGER.info("{} is calculated repo root".format(root))
                 proc = subprocess.Popen(
                     shlex.split('test-proxy start --storage-location="{}" -- --urls "{}"'.format(root, PROXY_URL)),
                     stdout=log,
                     stderr=log,
+                    env = {}
                 )
                 os.environ[TOOL_ENV_VAR] = str(proc.pid)
         else:

tools/azure-sdk-tools/devtools_testutils/proxy_testcase.py

@@ -6,6 +6,7 @@
 import logging
 import requests
 import six
+import os
 from typing import TYPE_CHECKING
 import urllib.parse as url_parse
 
@@ -19,6 +20,7 @@
 from azure_devtools.scenario_tests.utilities import trim_kwargs_from_test_function
 from .config import PROXY_URL
 from .helpers import get_test_id, is_live, is_live_and_not_recording, set_recording_id
+from .proxy_startup import discovered_roots
 
 if TYPE_CHECKING:
     from typing import Callable, Dict, Tuple
@@ -34,6 +36,26 @@
 PLAYBACK_START_URL = "{}/playback/start".format(PROXY_URL)
 PLAYBACK_STOP_URL = "{}/playback/stop".format(PROXY_URL)
 
+def get_recording_assets(test_id: str) -> str:
+    """
+    Used to retrieve the assets.json given a PYTEST_CURRENT_TEST test id.
+    """
+    for root in discovered_roots:
+        current_dir = os.path.dirname(test_id)
+        while current_dir is not None and not (os.path.dirname(current_dir) == current_dir):
+            possible_assets = os.path.join(current_dir, "assets.json")
+            possible_root = os.path.join(current_dir, ".git")
+
+            # we need to check for assets.json first!
+            if os.path.exists(os.path.join(root, possible_assets)):
+                return os.path.abspath(os.path.join(root, possible_assets))
+            # we need the git check to prevent ascending out of the repo
+            elif os.path.exists(os.path.join(root, possible_root)):
+                return None
+            else:
+                current_dir = os.path.dirname(current_dir)
+
+    return None
 
 def start_record_or_playback(test_id: str) -> "Tuple[str, Dict[str, str]]":
     """Sends a request to begin recording or playing back the provided test.
@@ -42,11 +64,16 @@ def start_record_or_playback(test_id: str) -> "Tuple[str, Dict[str, str]]":
     test variables to values. If no variable dictionary was stored when the test was recorded, b is an empty dictionary.
     """
     variables = {}  # this stores a dictionary of test variable values that could have been stored with a recording
+    
+    json_payload = {"x-recording-file": test_id}
+    assets_json = get_recording_assets(test_id)
+    if assets_json:
+        json_payload["x-recording-assets-file"] = assets_json    
 
     if is_live():
         result = requests.post(
             RECORDING_START_URL,
-            json={"x-recording-file": test_id},
+            json=json_payload,
         )
         if result.status_code != 200:
             message = six.ensure_str(result._content)
@@ -56,7 +83,7 @@ def start_record_or_playback(test_id: str) -> "Tuple[str, Dict[str, str]]":
     else:
         result = requests.post(
             PLAYBACK_START_URL,
-            json={"x-recording-file": test_id},
+            json=json_payload,
         )
         if result.status_code != 200:
             message = six.ensure_str(result._content)