[Release_sdk_status] Fix release_sdk_status for run test break (Azure#27906)

* fix release_sdk_status

* Update release_sdk_status.yml

* Delete mgmt_settings_real_.py

* Update main.py

* Update main.py

* Update release_sdk_status.yml

* Update release_sdk_status.yml

Co-authored-by: Zhenbiao Wei (WICRESOFT NORTH AMERICA LTD) <v-zhenbwei@microsoft.com>

Author: Wzb123456789

scripts/release_sdk_status/main.py

@@ -7,6 +7,7 @@
 import subprocess as sp
 from typing import List, Dict
 from github import Github
+from functools import wraps
 from github.Repository import Repository
 import time
 from packaging.version import parse
@@ -19,10 +20,27 @@
 FAILED_RESULT = []
 SKIP_TEXT = '-, -, -, -\n'
 
+
+def return_origin_path(func):
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        current_path = os.getcwd()
+        result = func(*args, **kwargs)
+        os.chdir(current_path)
+        return result
+
+    return wrapper
+
+
 def my_print(cmd):
     print('== ' + cmd + ' ==\n')
 
 
+def print_exec(cmd: str):
+    my_print(cmd)
+    sp.call(cmd, shell=True)
+
+
 def print_check(cmd, path=''):
     my_print(cmd)
     if path:
@@ -36,6 +54,10 @@ def print_call(cmd):
     sp.call(cmd, shell=True)
 
 
+def start_test_proxy():
+    print_check('pwsh {}/eng/common/testproxy/docker-start-proxy.ps1 \"start\"'.format(os.getenv('SDK_REPO')))
+
+
 def version_sort(versions: List[str]) -> List[str]:
     versions_package = [parse(version) for version in versions]
     versions_package.sort()
@@ -126,21 +148,21 @@ def write_to_list(self, sdk_folder: str) -> str:
             return '{sdk_folder}/{package_name},{pypi_link},{track1_latest_version},{track1_latest_release_date},' \
                    '{track1_ga_version},{track2_latest_version},{track2_latest_release_date},{track2_ga_version},' \
                    '{cli_version},{track_config},{bot},{readme_link},{multiapi},{whl_size},'.format(
-                        sdk_folder=sdk_folder.split('/')[0],
-                        package_name=self._package_name,
-                        pypi_link=self.pypi_link,
-                        track1_latest_version=self.track1_latest_version,
-                        track1_latest_release_date=self.version_date_dict[self.track1_latest_version],
-                        track1_ga_version=self.track1_ga_version,
-                        track2_latest_version=self.track2_latest_version,
-                        track2_latest_release_date=self.version_date_dict[self.track2_latest_version],
-                        track2_ga_version=self.track2_ga_version,
-                        cli_version=self.cli_version,
-                        track_config=self.track_config,
-                        bot=self.bot_warning,
-                        readme_link=self.rm_link,
-                        multiapi=self.multi_api,
-                        whl_size=self.output_package_size())
+                sdk_folder=sdk_folder.split('/')[0],
+                package_name=self._package_name,
+                pypi_link=self.pypi_link,
+                track1_latest_version=self.track1_latest_version,
+                track1_latest_release_date=self.version_date_dict[self.track1_latest_version],
+                track1_ga_version=self.track1_ga_version,
+                track2_latest_version=self.track2_latest_version,
+                track2_latest_release_date=self.version_date_dict[self.track2_latest_version],
+                track2_ga_version=self.track2_ga_version,
+                cli_version=self.cli_version,
+                track_config=self.track_config,
+                bot=self.bot_warning,
+                readme_link=self.rm_link,
+                multiapi=self.multi_api,
+                whl_size=self.output_package_size())
         else:
             self.pypi_link = 'NA'
         return ''
@@ -155,7 +177,7 @@ def find_track2_ga_version(self, versions: List[str]) -> None:
                 self.track2_ga_version = version
                 break
 
-    def handle_cli_version(self, track1_versions: List[str], track2_versions: List[str]) ->None:
+    def handle_cli_version(self, track1_versions: List[str], track2_versions: List[str]) -> None:
         if self.cli_version == 'NA':
             return
         if self.cli_version in track1_versions:
@@ -209,9 +231,61 @@ def bot_analysis(self):
             self.bot_warning += 'Need to add track2 config.'
 
 
+def sdk_code_path(service_name, sdk_name) -> str:
+    return str(Path(os.getenv('SDK_REPO') + f'/sdk/{service_name}/{sdk_name}'))
+
+
+@return_origin_path
+def install_package_locally(service_name, sdk_name):
+    os.chdir(sdk_code_path(service_name, sdk_name))
+    print_check('pip install -e .')
+    print_exec('pip install -r dev_requirements.txt')
+
+
+@return_origin_path
+def run_test_proc(sdk_name, service_name, sdk_folder):
+    # run test
+    if os.getenv('SKIP_COVERAGE') in ('true', 'yes'):
+        return SKIP_TEXT
+
+    coverage_path = ''.join([os.getenv('SDK_REPO'), '/sdk/', sdk_folder])
+    service_path = coverage_path.split('/azure/mgmt')[0]
+    os.chdir(sdk_code_path(service_name, sdk_name))
+    try:
+        print_check(f'pytest  --collect-only')
+    except:
+        print('live test run done, do not find any test !!!')
+        return '-, -, -, -\n'
+    if os.path.exists(coverage_path + '/operations') and os.path.exists(coverage_path + '/models'):
+        operations_path = coverage_path + '/operations'
+        models_path = coverage_path + '/models'
+        try:
+            start_time = int(time.time())
+            print_check(f'pytest tests -s --cov={operations_path} --cov={models_path} >result.txt', path=service_path)
+            cost_time = int(time.time()) - start_time
+            my_print(f'{service_name} play_back cost {cost_time} seconds({cost_time // 60} minutes)')
+        except Exception as e:
+            print(f'{service_name} test ERROR')
+            return '-, 0, 0, 0\n'
+    else:
+        try:
+            print_check(f'pytest tests -s >result.txt', path=service_path)
+        except Exception as e:
+            return '-, 0, 0, 0\n'
+    if os.path.exists(service_path + '/result.txt'):
+        return get_test_result(service_path + '/result.txt')
+
+
+def run_test(sdk_name, service_name, sdk_folder):
+    install_package_locally(service_name, sdk_name)
+    test_result = run_test_proc(sdk_name, service_name, sdk_folder)
+    return test_result
+
+
 def sdk_info_from_pypi(sdk_info: List[Dict[str, str]], cli_dependency):
     all_sdk_status = []
-    add_certificate(str(Path('../venv-sdk/lib/python3.8/site-packages/certifi/cacert.pem')))
+    add_certificate()
+    start_test_proxy()
     for package in sdk_info:
         sdk_name = package['package_name']
         if sdk_name in cli_dependency.keys():
@@ -230,7 +304,7 @@ def sdk_info_from_pypi(sdk_info: List[Dict[str, str]], cli_dependency):
         if pypi_ins.pypi_link != 'NA':
             test_result = SKIP_TEXT
             try:
-                test_result = run_playback_test(service_name, sdk_folder)
+                test_result = run_test(sdk_name, service_name, sdk_folder)
             except:
                 print(f'[Error] fail to play back test recordings: {sdk_name}')
             text_to_write += test_result
@@ -260,40 +334,6 @@ def get_test_result(txt_path):
     return f'{coverage}, {passed}, {failed}, {skipped}\n'
 
 
-def run_playback_test(service_name: str, sdk_folder: str):
-    if os.getenv('SKIP_COVERAGE') in ('true', 'yes'):
-        return SKIP_TEXT
-
-    # eg: coverage_path='$(pwd)/sdk-repo/sdk/containerregistry/azure-mgmt-containerregistry/azure/mgmt/containerregistry/'
-    coverage_path = ''.join([os.getenv('SDK_REPO'), '/sdk/', sdk_folder])
-    service_path = coverage_path.split('/azure/mgmt')[0]
-    test_path = service_path + '/tests'
-    if os.path.exists(test_path):
-        print_check('pip install -r dev_requirements.txt', path=service_path)
-        print_check('pip install -e .', path=service_path)
-        # print_check('python setup.py install', path=service_path)
-        if os.path.exists(coverage_path+'/operations') and os.path.exists(coverage_path+'/models'):
-            operations_path = coverage_path+'/operations'
-            models_path = coverage_path+'/models'
-            try:
-                start_time = int(time.time())
-                print_check(f'pytest -s tests --cov={operations_path} --cov={models_path} >result.txt', path=service_path)
-                cost_time = int(time.time()) - start_time
-                my_print(f'{service_name} play_back cost {cost_time} seconds({cost_time // 60} minutes)')
-            except Exception as e:
-                print(f'{service_name} test ERROR')
-                return '-, 0, 0, 0\n'
-        else:
-            try:
-                print_check(f'pytest -s tests >result.txt', path=service_path)
-            except Exception as e:
-                return '-, 0, 0, 0\n'
-        if os.path.exists(service_path+'/result.txt'):
-            return get_test_result(service_path+'/result.txt')
-
-    return SKIP_TEXT
-
-
 def write_to_csv(sdk_status_list, csv_name):
     with open(csv_name, 'w') as file_out:
         file_out.write('foler/package name,'

scripts/release_sdk_status/release_sdk_status.yml

@@ -60,6 +60,7 @@ jobs:
         export SWAGGER_REPO=$(pwd)/azure-rest-api-specs
         export SDK_REPO=$(pwd)/sdk-repo
         export SKIP_COVERAGE=$(SKIP_COVERAGE)
+        export AZURE_TEST_RUN_LIVE=false
 
         # create virtual env
         python -m venv venv-sdk
@@ -71,4 +72,4 @@ jobs:
         git checkout release-sdk-status
 
         # run
-        python $script_path/main.py
+        python $script_path/main.py

scripts/release_sdk_status/util.py

@@ -8,8 +8,9 @@ def _find_certificate():
         return fr.read()
 
 
-def add_certificate(file: str):
+def add_certificate():
     certification = _find_certificate()
-    with open(file, 'a+') as f:
+    cacert_path = Path('../venv-sdk/lib/python3.8/site-packages/certifi/cacert.pem')
+    with open(cacert_path, 'a+') as f:
         f.seek(0, 0)
         f.write(certification)