Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use set_exit_code method in Slash

Best Python code snippet using slash

check-oceanstor.py

Source:check-oceanstor.py

...10    failed_health_status = ["Offline", "Pre-fail", "Fault", "No Input", "--"]11    failed_running_status = ["Offline", "Reconstruction", "Balancing", "--"]12    def check_empty_respone():13        pass14    def set_exit_code(code):15        # If code is more critical than actual level set it16        global exit_code17        if code > exit_code:18            exit_code = code19    def lslun():20        ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('show lun general')21        ssh_lines = ssh_stdout.readlines()[4:]22        output_info = ""23        24        # return if there are no entries on storage system25        if len(ssh_lines) == 0:26            return "OK: There are no LUNs defined\n"27            28        # Check if there are any critical LUNs29        if not any( line.split()[4] in failed_health_status for line in ssh_lines ):30            output_info += "OK: All LUNs Online \n"31        else:32            output_info += "CRITICAL: check your LUN status below \n"33            set_exit_code(2)34        for line in ssh_lines:35            # Assign values36            name, status = line.split()[1], line.split()[4]37            # Check for errors38            if status == "Normal":39                output_info += "OK: LUN {} status: {}\n".format(name, status)40            else:41                output_info += "CRITICAL: LUN {} status: {}\n".format(name, status)42                43        return output_info44    def lsdisk():45        ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('show disk general')46        ssh_lines = ssh_stdout.readlines()[4:]47        output_info = ""48        49        # return if there are no entries on storage system50        if len(ssh_lines) == 0:51            return "OK: There are no DISKs defined\n"52        53        # Check if there are any critical DISKs54        if not any( line.split()[1] in failed_health_status for line in ssh_lines ):55            output_info += "OK: All DISKs Online and Healthy \n"56        else:57            output_info += "CRITICAL: check your DISK status below \n"58            set_exit_code(2)59        for line in ssh_lines:60            # Assign values61            slot, status, disk_type, capacity, role = line.split()[0], line.split()[1], line.split()[3], line.split()[4], line.split()[5]62            # Check for errors63            if status == "Normal":64                output_info += "OK: DISK {} status: {}\n".format(slot, status)65            else:66                output_info += "CRITICAL: DISK {} status: {}\n role: {}\n type: {}\n capacity: {}\n".format(slot, status, role, disk_type, capacity)67                68        return output_info69    def lsdiskdomain():70        ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('show disk_domain general')71        ssh_lines = ssh_stdout.readlines()[4:]72        output_info = ""73        74        # return if there are no entries on storage system75        if len(ssh_lines) == 0:76            return "OK: There are no DISK DOMAINs defined\n"77        78        # Check if there are any critical DISK DOMAINs by Health status79        if not any( line.split()[2] in failed_health_status for line in ssh_lines ):80            output_info += "OK: All DISK DOMAINs Online \n"81        else:82            output_info += "CRITICAL: check your DISK DOMAIN status \n"83            set_exit_code(2)84        # Check if there are any critical DISK DOMAINs by Running status85        if any( line.split()[3] in failed_running_status for line in ssh_lines ):86            87            # Clear OK/Critical message set by Health Status, because Running is Critical88            output_info = ""89            output_info += "CRITICAL: Check your DISK DOMAIN status \n"90            set_exit_code(2)91        for line in ssh_lines:92            # Assign values93            name, health_status, running_status = line.split()[1], line.split()[2], line.split()[3]94            # Check for errors in health status95            if health_status == "Normal":96                # Check for errors in running status97                if running_status in failed_running_status:98                    output_info += "CRITICAL: DISK DOMAIN {} health status: {} running status: {}\n".format(name, health_status, running_status)99                else:100                    output_info += "OK: DISK DOMAIN {} health status: {} running status: {}\n".format(name, health_status, running_status)101            else:102                output_info += "CRITICAL: DISK DOMAIN {} health status: {} running status: {}\n".format(name, health_status, running_status)103                104        return output_info105    def lsexpansionmodule():106        ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('show expansion_module')107        ssh_lines = ssh_stdout.readlines()[4:]108        output_info = ""109        110        # return if there are no entries on storage system111        if len(ssh_lines) == 0:112            return "OK: There are no EXPANSION MODULEs defined\n"113        114        # Check if there are any critical EXPANSION MODULEs by Health status115        if not any( line.split()[1] in failed_health_status for line in ssh_lines ):116            output_info += "OK: All EXPANSION MODULEs Online \n"117        else:118            output_info += "CRITICAL: check your EXPANSION MODULEs status \n"119            set_exit_code(2)120        # Check if there are any critical EXPANSION MODULEs by Running status121        if any( line.split()[2] in failed_running_status for line in ssh_lines ):122            # Clear OK/Critical message set by Health Status, because Running is Critical123            output_info = ""124            output_info += "CRITICAL: Check your EXPANSION MODULEs status \n"125            set_exit_code(2)126        for line in ssh_lines:127            # Assign values128            expansion_id, health_status, running_status = line.split()[0], line.split()[1], line.split()[2]129            # Check for errors in health status130            if health_status == "Normal":131                # Check for errors in running status132                if running_status in failed_running_status:133                    output_info += "CRITICAL: EXPANSION MODULE {} health status: {} running status: {}\n".format(expansion_id, health_status, running_status)134                else:135                    output_info += "OK: EXPANSION MODULE {} health status: {} running status: {}\n".format(expansion_id, health_status, running_status)136            else:137                output_info += "CRITICAL: EXPANSION MODULE {} health status: {} running status: {}\n".format(expansion_id, health_status, running_status)138                139        return output_info140    def lsinitiator():141        ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('show initiator')142        ssh_lines = ssh_stdout.readlines()[4:]143        output_info = ""144        145        # return if there are no entries on storage system146        if len(ssh_lines) == 0:147            return "OK: There are no INITIATORs defined\n"148        149        # Check if there are any critical INITIATORs150        if not any( line.split()[1] == "Offline" for line in ssh_lines ):151            output_info += "OK: All INITIATORs Online \n"152        else:153            output_info += "WARNING: INITIATOR OFFLINE \n"154            set_exit_code(1)155        for line in ssh_lines:156            # Assign values157            name, status = line.split()[0], line.split()[1]158            # Check for errors159            if status == "Online":160                output_info += "OK: INITIATOR {} status: {}\n".format(name, status)161            else:162                output_info += "WARNING: INITIATOR {} status: {}\n".format(name, status)163                164        return output_info165    def lsstoragepool():166        ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('show storage_pool general')167        ssh_lines = ssh_stdout.readlines()[4:]168        output_info = ""169        170        # return if there are no entries on storage system171        if len(ssh_lines) == 0:172            return "OK: There are no STORAGE POOLs defined\n"173        174        # Check if there are any critical STORAGE POOLs by Health status175        if not any( line.split()[3] in failed_health_status for line in ssh_lines ):176            output_info += "OK: All STORAGE POOLs Online \n"177        else:178            output_info += "CRITICAL: Check your STORAGE POOL status \n"179            set_exit_code(2)180        # Check if there are any critical STORAGE POOLs by Running status181        if any( line.split()[4] in failed_running_status for line in ssh_lines ):182            183            # Clear OK/Critical message set by Health Status, because Running is Critical184            output_info = ""185            output_info += "CRITICAL: Check your STORAGE POOL status \n"186            set_exit_code(2)187        for line in ssh_lines:188            # Assign values189            name, health_status, running_status = line.split()[1], line.split()[3], line.split()[4]190            # Check for errors191            if running_status == "Online":192                output_info += "OK: STORAGE POOL {} health status: {} running status: {}\n".format(name, health_status, running_status)193            else:194                output_info += "CRITICAL: STORAGE POOL {} health status: {} running status: {}\n".format(name, health_status, running_status)195                196        return output_info197    def lspsu():198        ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('show power_supply')199        ssh_lines = ssh_stdout.readlines()[4:]200        output_info = ""201        # return if there are no entries on storage system202        if len(ssh_lines) == 0:203            output_info += "CRITICAL: No PSUs were found \n"204            set_exit_code(2)205        # Check if there are any critical STORAGE POOLs206        if not any( [x for x in re.split("\s{2,}",line) if x][2] in failed_running_status for line in ssh_lines ):207            output_info += "OK: All PSU Online \n"208        else:209            output_info += "CRITICAL: Check your PSU status \n"210            set_exit_code(2)211        for line in ssh_lines:212            # split string per double spaces because of status "No Input" at 2nd column213            split_line = [x for x in re.split("\s{2,}",line) if x]214            # Assign values215            name, health_status, running_status = split_line[0], split_line[1], split_line[2]216            # Check for errors217            if running_status == "Online":218                output_info += "OK: PSU {} health status: {} running status: {}\n".format(name, health_status, running_status)219            else:220                output_info += "CRITICAL: PSU {} health status: {} running status: {}\n".format(name, health_status, running_status)221        return output_info222    def lsallstatuses():223        global output_info224        output_info += lslun() + "\n"...

check_sesam_migration.py

Source:check_sesam_migration.py

...7WARNING = ['CANCELLED', 'PARTIALLY_DELETED', 'DELETED']8CRITICAL = ['ERROR', 'SUPPRESSED']9UNKOWN = ['NONE']10exit_code = 011def set_exit_code(code):12    global exit_code13    if code > exit_code:14        exit_code = code15def get_status():16    if exit_code == 0:17        return 'OK'18    elif exit_code == 1:19        return 'WARNING'20    elif exit_code == 2:21        return 'CRITICAL'22    else:23        return 'UNKNOWN'24def set_status(status):25    if status in OK:26        set_exit_code(0)27    elif status in CRITICAL:28        set_exit_code(2)29    elif status in WARNING:30        set_exit_code(1)31    else:32        set_exit_code(3)33def main():34    parser = argparse.ArgumentParser(description='Monitoring check for SEP Sesam migrationq states')35    parser.add_argument('--user',36                        '-u',37                        required=True,38                        type=str,39                        help='Username for the sesam REST API',40                        dest='user')41    parser.add_argument('--password',42                        '-p',43                        required=True,44                        type=str,45                        help='Password to the given username',46                        dest='password')...

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.