KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use test_ignore_status method in autotest

Best Python code snippet using autotest_python
simpleRender.py
Source:simpleRender.py
1import argparse2import os3import re4import json5import sys6import platform7import psutil8import subprocess9from datetime import datetime10from shutil import copyfile, SameFileError11from pathlib import Path12# Configure script context and importing jobs_launcher and logger to it (DO NOT REPLACE THIS CODE)13ROOT_DIR = os.path.abspath(14    os.path.join(os.path.dirname(__file__), os.path.pardir, os.path.pardir)15)16sys.path.append(ROOT_DIR)17import jobs_launcher.core.config as core_config18import jobs_launcher.core.system_info as system_info19from jobs_launcher.core.kill_process import kill_process20from usdParser import UsdParser21LOG = core_config.main_logger22HUSK_NAME = ['husk', 'husk.exe']23# Makes report and execute case24class Renderer:25    PLATFORM = None26    LOG = None27    TOOL = None28    ASSETS_PATH = None29    BASELINE_PATH = None30    PACKAGE = None31    COMMON_REPORT_PATH = None32    # case - render scenario; output_dir - output directory for report and images33    def __init__(self, case, output_dir, update_refs, engine, res_x, res_y, min_samples, max_samples, 34            threshold, retries, stucking_time, rpr_traces_path, rif_traces_path):35        self.case = case36        self.output = output_dir37        self.update_refs = update_refs38        self.retries = retries39        self.scene_path = os.path.join(Renderer.ASSETS_PATH, Renderer.PACKAGE, case['scene'])40        self.case_report_path = os.path.join(self.output, case['case'] + core_config.CASE_REPORT_SUFFIX)41        for d in ['Color', 'render_tool_logs']:42            Path(os.path.join(output_dir, d)).mkdir(parents=True, exist_ok=True)43        Renderer.COMMON_REPORT_PATH = os.path.join(output_dir, 'renderTool.log')44        self.engine = engine45        self.width = int(res_x)46        self.height = int(res_y)47        self.min_samples = int(min_samples)48        self.max_samples = int(max_samples)49        self.threshold = float(threshold)50        self.stucking_time = int(stucking_time)51        self.rpr_traces_path = rpr_traces_path52        self.rif_traces_path = rif_traces_path53        if Renderer.TOOL is None or Renderer.ASSETS_PATH is None:54            raise Exception("Path to tool executable didn't set")55        else:56            self.__prepare_report()57    # Copy baselines images to work dirs58    def __copy_baseline(self):59        # Get original baseline json report from assets folder60        orig_baselines_dir = os.path.join(Renderer.BASELINE_PATH, self.PACKAGE)61        orig_baseline_path = os.path.join(orig_baselines_dir, self.case['case'] + core_config.CASE_REPORT_SUFFIX)62        # Create dir for baselines json for current case group in Work/Baseline/group_name63        copied_baselines_dir = os.path.join(self.output, os.pardir, os.pardir, os.pardir, 'Baseline', self.PACKAGE)64        if not os.path.exists(copied_baselines_dir):65            os.makedirs(copied_baselines_dir)66            # Create dir for baselines images for current case group in Work/Baseline/group_name/Color67            os.makedirs(os.path.join(copied_baselines_dir, 'Color'))68        copied_baseline_path = os.path.join(copied_baselines_dir, self.case['case'] + core_config.CASE_REPORT_SUFFIX)69        try:70            copyfile(orig_baseline_path, copied_baseline_path)71            with open(os.path.join(copied_baseline_path)) as f:72                baseline_json = json.load(f)73            for thumb in [''] + core_config.THUMBNAIL_PREFIXES:74                orig_thumbnail = os.path.join(orig_baselines_dir, baseline_json[thumb + 'render_color_path'])75                copied_thumbnail = os.path.join(copied_baselines_dir, baseline_json[thumb + 'render_color_path'])76                if thumb + 'render_color_path' and os.path.exists(orig_thumbnail):77                    copyfile(orig_thumbnail, copied_thumbnail)78        except Exception as e:79            LOG.error(f'Failed to copy baseline {repr(e)} from: {orig_baseline_path} to: {copied_baseline_path}')80    # Creates stub image which will be replaced on success render81    def __copy_stub_image(self, status):82        try:83            root_dir_path = os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir, os.path.pardir))84            orig_stub_path = os.path.join(root_dir_path, 'jobs_launcher', 'common', 'img', status + '.jpg')85            copied_stub_path = os.path.join(self.output, 'Color', self.case['case'] + '.jpg')86            copyfile(orig_stub_path, copied_stub_path)87        except OSError or FileNotFoundError as e:88            LOG.error(f"Can't create img stub: {str(e)}")89    def __is_case_skipped(self):90        render_platform = set({Renderer.PLATFORM["GPU"], Renderer.PLATFORM["OS"]})91        if sum([render_platform & set(skip_conf) == set(skip_conf) for skip_conf in self.case.get('skip_config', '')]):92            for i in self.case['skip_config']:93                skip_config = set(i)94                if render_platform.intersection(skip_config) == skip_config:95                    return True96        for eng in self.case.get('skip_engine', []):97            eng = set(eng.split(" & "))98            if render_platform.intersection(eng):99                return True100        if sum([1 for eng in self.case.get('skip_engine', []) if eng == self.engine]):101            return True102        return False103    def __get_tool_version(self):104        if Renderer.is_windows():105            return str(Renderer.TOOL).split("\\")[-3]106        elif Renderer.is_macos():107            return str(Renderer.TOOL).split("/")[3]108        else:109            return str(Renderer.TOOL).split("/")[-3]110    def __prepare_report(self):111        if self.__is_case_skipped(): self.case['status'] = core_config.TEST_IGNORE_STATUS112        report = core_config.RENDER_REPORT_BASE.copy()113        plugin_info = Renderer.PLATFORM['PLUGIN']114        report.update({115            'test_case': self.case['case'],116            'test_group': Renderer.PACKAGE,117            'script_info': self.case['script_info'] if 'script_info' in self.case else [],118            'render_device': Renderer.PLATFORM.get('GPU', 'Unknown'),119            'scene_name': self.case['scene'],120            'tool': self.__get_tool_version(),121            'engine': self.engine,122            'min_samples': self.min_samples,123            'max_samples': self.max_samples,124            'threshold': self.threshold,125            'stucking_time': self.stucking_time,126            'date_time': datetime.now().strftime('%m/%d/%Y %H:%M:%S'),127            'file_name': self.case['case'] + self.case.get('extension', '.jpg'),128            'render_color_path': os.path.join('Color', self.case['case'] + self.case.get('extension', '.jpg')),129            'render_version': plugin_info['plugin_version'],130            'core_version': plugin_info['core_version'],131        })132        if 'jira_issue' in self.case:133            report['jira_issue'] = self.case['jira_issue']134        if 'frame' in self.case:135            report['frame'] = self.case['frame']136        if self.width > 0 and self.height > 0:137            report['width'] = self.width138            report['height'] = self.height139        if self.case['status'] == core_config.TEST_IGNORE_STATUS:140            report['test_status'] = core_config.TEST_IGNORE_STATUS141            report['group_timeout_exceeded'] = False142            self.__copy_stub_image(core_config.TEST_IGNORE_STATUS)143        else:144            report['test_status'] = core_config.TEST_CRASH_STATUS145            self.__copy_stub_image('error')146        with open(self.case_report_path, 'w') as f:147            json.dump([report], f, indent=4)148        if 'Update' not in self.update_refs:149            self.__copy_baseline()150    def __complete_report(self, try_number):151        case_log_path = os.path.join('render_tool_logs', self.case['case'] + '.log')152        with open(Renderer.COMMON_REPORT_PATH, "a") as common_log:153            with open(case_log_path, 'r') as case_log:154                common_log.write(case_log.read())155        with open(self.case_report_path, 'r') as f:156            report = json.load(f)[0]157        if self.case['status'] == "done_observed":158            self.case['status'] = core_config.TEST_OBSERVED_STATUS159        elif self.case['status'] == "done":160            self.case['status'] = core_config.TEST_SUCCESS_STATUS161        else:162            report['message'] = ["Test case wasn't executed successfully. Number of tries: " + str(try_number)]163        if os.path.isfile(case_log_path):164            with open(case_log_path, 'r') as f:165                tool_log = [line.strip() for line in f]166            render_time_start = False167            for i in range(0, len(tool_log)):168                if "Lap=" in tool_log[i] and not render_time_start:169                    render_time_start = True170                if not "Lap=" in tool_log[i] and render_time_start:171                    render_time_start = False172                    try:173                        time = datetime.strptime(tool_log[i-1].split()[2].replace('Lap=', ''), '%H:%M:%S.%f')174                        total_seconds = float(time.second + time.minute * 60 + time.hour * 3600) + (time.microsecond / 100000)175                        report['render_time'] = total_seconds176                    except:177                        LOG.error(f"Failed to parse render time from {tool_log[i-1]}")178                        report['render_time'] = 0179                if "Peak Memory Usage" in tool_log[i]: report["gpu_memory_max"] = ' '.join(tool_log[i].split()[-2:])180                if "Current Memory Usage" in tool_log[i]: report["gpu_memory_usage"] = ' '.join(tool_log[i].split()[-2:])181                if f"Run time {self.stucking_time} exceeded" in tool_log[i]:182                    report["testcase_timeout_exceeded"] = True183                    self.case['status'] = core_config.TEST_CRASH_STATUS184        report['render_log'] = case_log_path185        report['test_status'] = self.case['status']186        report['group_timeout_exceeded'] = self.case['group_timeout_exceeded']187        report['number_of_tries'] = try_number188        report['render_mode'] = 'GPU'189        with open(self.case_report_path, 'w') as f:190            json.dump([report], f, indent=4)191    def __update_scene(self):192        193        usd_scene = UsdParser(194            assets_path=Renderer.ASSETS_PATH,195            source_file=self.case['scene'],196            test_group=Renderer.PACKAGE,197            test_case=self.case['case']198        )199        200        usd_scene.update_or_create_render_setting(f"token rpr:core:renderQuality = \"{self.engine}\"")201        usd_scene.update_or_create_render_setting(f"int rpr:adaptiveSampling:minSamples = {self.min_samples}")202        usd_scene.update_or_create_render_setting(f"int rpr:maxSamples = {self.max_samples}")203        usd_scene.update_or_create_render_setting(f"int rpr:adaptiveSampling:noiseTreshold = {self.threshold}")204        if 'render_settings' in self.case.keys():205            for value in self.case['render_settings']:206                usd_scene.update_or_create_render_setting(value)207        self.scene_path = usd_scene.save()208    def __remove_scene(self):209        scene_name = os.path.basename(self.scene_path)210        if scene_name.startswith(self.case['case']):211            try:212                LOG.info(f"Scene {self.scene_path} removed.")213                os.remove(self.scene_path)214            except Exception as e:215                LOG.error(f"Failed to remove {self.scene_path}. Exception: {e}")216    def render(self):217        if self.case['status'] != core_config.TEST_IGNORE_STATUS:218            if self.rpr_traces_path and self.rpr_traces_path != "No":219                rpr_case_traces_path = os.path.join(self.rpr_traces_path, Renderer.PACKAGE, self.case['case'])220                LOG.info(f"RPR Traces path: {rpr_case_traces_path}")221                if not os.path.exists(rpr_case_traces_path): os.makedirs(rpr_case_traces_path)222                os.environ['RPRTRACEPATH'] = os.path.abspath(rpr_case_traces_path)223            if self.rif_traces_path and self.rif_traces_path != "No":224                rif_case_traces_path = os.path.join(self.rif_traces_path, Renderer.PACKAGE, self.case['case'])225                LOG.info(f"RIF Traces path: {rpr_case_traces_path}")226                if not os.path.exists(rif_case_traces_path): os.makedirs(rif_case_traces_path)227                os.environ['RIF_TRACING_ENABLED'] = "1"228                os.environ['RIF_TRACING_PATH'] = os.path.abspath(rif_case_traces_path)229            # create new scene with needed render settings230            self.scene_path = os.path.abspath(self.scene_path)231            self.__update_scene()232            233            if self.case['status'] == core_config.TEST_OBSERVED_STATUS:234                self.case['status'] = 'inprogress_observed'235            else:236                self.case['status'] = 'inprogress'237            hip_log = os.path.join('render_tool_logs', self.case['case'] + '.log')238            hip_command = f"""239                "{Renderer.TOOL}" "{self.scene_path}" --renderer RPR --verbose 9 \240                --output "{os.path.join('Color', self.case['case'] + '.jpg')}" \241                --append-stderr "{hip_log}" --append-stdout {hip_log} \242                --timelimit {self.stucking_time} --timelimit-nosave-partial \243            """244            if self.width > 0 and self.height > 0:245                hip_command += f" --res {self.width} {self.height}"246            else:247                hip_command += f" --res 1920 1080"248            if 'frame' in self.case:249                hip_command += f" --frame {self.case['frame']}"250            if 'camera' in self.case:251                hip_command += f" --camera {self.case['camera']}"252            hip_command = re.sub(r"\s+", " ", hip_command)253            LOG.info(f"CMD {hip_command}")254            # saving render command to script for debugging purpose255            shell_script_path = os.path.join(self.output, (self.case['case'] + '_render') + '.bat' if Renderer.is_windows() else '.sh')256            with open(shell_script_path, 'w') as f:257                f.write(hip_command)258            if not Renderer.is_windows():259                try:260                    os.system('chmod +x ' + shell_script_path)261                except OSError as e:262                    LOG.error(f'Error while setting right for script execution {str(e)}')263            os.chdir(self.output)264            def execute_task(timeout=self.stucking_time + 30):265                p = psutil.Popen(shell_script_path, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)266                try:267                    p.wait(timeout=timeout)268                    return p.returncode269                except (psutil.TimeoutExpired, subprocess.TimeoutExpired) as e:270                    LOG.error(f'Render has been aborted by timeout.{str(e)}')271                    try:272                        for child in reversed(p.children(recursive=True)):273                            child.terminate()274                        p.terminate()275                        kill_process(HUSK_NAME)276                    except Exception as e1:277                        LOG.error("Failed to terminate running process: {}".format(e1))278                    return 1279            success_flag = False280            try_number = 0281            while try_number < self.retries:282                try_number += 1283                rc = execute_task()284                LOG.info(f'Husk return code {rc}')285                if rc == 0:286                    success_flag = True287                    break288            if success_flag:289                if self.case['status'] == 'inprogress_observed':290                    self.case['status'] = 'done_observed'291                else:292                    self.case['status'] = 'done'293            else:294                self.case['status'] = core_config.TEST_CRASH_STATUS295            self.case['group_timeout_exceeded'] = False296            test_cases_path = os.path.join(self.output, 'test_cases.json')297            with open(test_cases_path, 'r') as f:298                test_cases = json.load(f)299            for case in test_cases:300                if case['case'] == self.case['case']:301                    case['status'] = self.case['status']302            with open(test_cases_path, 'w') as f:303                json.dump(test_cases, f, indent=4)304            self.__complete_report(try_number)305            # it will be cleanup in pipelines, used for local work306            # self.__remove_scene()307    @staticmethod308    def is_windows():309        return platform.system() == "Windows"310    @staticmethod311    def is_macos():312        return platform.system() == "Darwin"313# Sets up the script parser314def create_parser():315    args = argparse.ArgumentParser()316    args.add_argument('--resolution_x', required=True, help='Width of image')317    args.add_argument('--resolution_y', required=True, help='Height of image')318    args.add_argument('--engine', required=True, default='Northstar', help='Render engine')319    args.add_argument('--min_samples', required=True, default=30, help='Min samples count')320    args.add_argument('--max_samples', required=True, default=50, help='Max samples count')321    args.add_argument('--threshold', required=True, default=0.05, help='Render threshold')322    args.add_argument('--stucking_time', required=False, default=600, help='Timeout for render')323    args.add_argument('--update_refs', required=True, help='Update or not references')324    args.add_argument('--tool', required=True, metavar='<path>', help='Path to render executable file')325    args.add_argument('--res_path', required=True, help='Path to folder with scenes')326    args.add_argument('--output', required=True, metavar='<path>', help='Path to folder where will be stored images and logs')327    args.add_argument('--test_cases', required=True, help='Path to json-file with test cases')328    args.add_argument('--package_name', required=True, help='Name of group of test cases')329    args.add_argument('--retries', required=False, default=2, type=int, help='The number of attempts to launch the case.')330    args.add_argument('--rpr_traces_path', required=False, default="No", help='Root path of RPR traces. RPR traces aren\'t collected if this param is No')331    args.add_argument('--rif_traces_path', required=False, default="No", help='Root path of RIF traces. RIF traces aren\'t collected if this param is No')332    return args333# Configure output_dir334def configure_output_dir(output, tests):335    try:336        os.makedirs(output)337        test_cases_path = os.path.realpath(os.path.join(os.path.abspath(output), 'test_cases.json'))338        copyfile(tests, test_cases_path)339        with open(test_cases_path, 'r') as f:340            test_cases = json.load(f)341            for case in test_cases:342                if 'status' not in case:343                    case['status'] = 'active'344            with open(test_cases_path, 'w') as copied_file:345                json.dump(test_cases, copied_file, indent=4)346        LOG.info(f"Scenes to render: {[name['scene'] for name in test_cases]}")347        return test_cases348    except OSError as e:349        LOG.error("Failed to read test_cases.json")350        raise e351    except (SameFileError, IOError) as e:352        LOG.error("Can't copy test_cases.json")353        raise e354def extract_plugin_versions():355    v = {'core_version': '0', 'plugin_version': '0'}356    for dir in os.listdir(ROOT_DIR):357        if 'hdRpr' in dir and not "tar.gz" in dir:358            try:359                with open(os.path.join(ROOT_DIR, dir, 'version'), 'r') as f:360                    raw = [line.strip().split(':') for line in f.readlines()]361                    for r in raw:362                        r[0] += '_version'363                        r[1] = str(r[1])364                    v = dict(raw)365            except FileNotFoundError as e:366                LOG.error(f"Can't find file with info about versions {repr(e)}")367    return v368def main():369    args = create_parser().parse_args()370    test_cases = []371    try:372        test_cases = configure_output_dir(args.output, args.test_cases)373    except Exception as e:374        LOG.error(repr(e))375        return 1376    # Defines the characteristics of machines which used to execute this script377    try:378        gpu = system_info.get_gpu()379    except:380        LOG.error("Can't get gpu name")381        gpu = 'Unknown'382    Renderer.PLATFORM = {383        'GPU': gpu,384        'OS': platform.system(),385        'PLUGIN': extract_plugin_versions()386    }387    Renderer.TOOL = args.tool388    Renderer.LOG = LOG389    Renderer.ASSETS_PATH = args.res_path390    Renderer.BASELINE_PATH = os.path.join(args.res_path, "..", "usd_houdini_autotests_baselines")391    Renderer.PACKAGE = args.package_name392    [case.render() for case in393        [Renderer(case, args.output, args.update_refs, args.engine, args.resolution_x, args.resolution_y, 394            args.min_samples, args.max_samples, args.threshold, args.retries, args.stucking_time, args.rpr_traces_path, args.rif_traces_path) 395        for case in test_cases]396     ]397    return 0398if __name__ == '__main__':...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.