text: #!/usr/bin/env python2.7
  
  # Amazon FPGA Hardware Development Kit
  #
  # Copyright 2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
  #
  # Licensed under the Amazon Software License (the "License"). You may not use
  # this file except in compliance with the License. A copy of the License is
  # located at
  #
  #    http://aws.amazon.com/asl/
  #
  # or in the "license" file accompanying this file. This file is distributed on
  # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, express or
  # implied. See the License for the specific language governing permissions and
  # limitations under the License.
  
  '''
  Base class for pytest modules
  
  See TESTING.md for details.
  '''
  
  from __future__ import print_function
  import boto3
  import os
  from os.path import basename, dirname, realpath, stat
  import glob
  import pytest
  import re
  import subprocess
  import sys
  import traceback
  import json
  try:
      import aws_fpga_test_utils
      from aws_fpga_test_utils import get_git_repo_root
      import aws_fpga_utils
  except ImportError as e:
      traceback.print_tb(sys.exc_info()[2])
      print("error: {}\nMake sure to source hdk_setup.sh".format(sys.exc_info()[1]))
      sys.exit(1)
  
  logger = aws_fpga_utils.get_logger(__name__)
  
  class AwsFpgaTestBase(object):
      '''
      Pytest test class.
  
      NOTE: Cannot have an __init__ method.
  
      Load AFI created by test_create_afi.py
      '''
  
      DCP_BUILD_STRATEGIES = [
          'BASIC',
          'DEFAULT',
          'EXPLORE',
          'TIMING',
          'CONGESTION']
  
      DCP_CLOCK_RECIPES = aws_fpga_test_utils.read_clock_recipes()
  
      DCP_URAM_OPTIONS = ['2', '3', '4']
  
      git_repo_dir = get_git_repo_root(dirname(__file__))
      WORKSPACE = git_repo_dir
  
      ADD_BATCH = False
      ADD_SIMULATOR = False
      ADD_EXAMPLEPATH = False
      ADD_RTENAME = False
      ADD_XILINX_VERSION = False
  
      msix_agfi = 'agfi-09c2a21805a8b9257'
  
      # sdk request timeout in seconds
      DEFAULT_REQUEST_TIMEOUT = 6000
  
      @classmethod
      def setup_class(cls, derived_cls, filename_of_test_class):
          AwsFpgaTestBase.s3_bucket = 'aws-fpga-jenkins-testing'
          AwsFpgaTestBase.__ec2_client = None
          AwsFpgaTestBase.__s3_client = None
          AwsFpgaTestBase.test_dir = dirname(realpath(filename_of_test_class))
  
          # SDAccel locations
          # Need to move to either a config file somewhere or a subclass
          AwsFpgaTestBase.xilinx_sdaccel_examples_prefix_path = "SDAccel/examples/xilinx"
          AwsFpgaTestBase.xilinx_sdaccel_examples_dir = AwsFpgaTestBase.git_repo_dir + "/" + AwsFpgaTestBase.xilinx_sdaccel_examples_prefix_path
          AwsFpgaTestBase.xilinx_sdaccel_examples_list_file = AwsFpgaTestBase.WORKSPACE + "/sdaccel_examples_list.json"
  
          # Vitis locations
          # Need to move to either a config file somewhere or a subclass
          AwsFpgaTestBase.xilinx_vitis_examples_prefix_path = "Vitis/examples/xilinx"
          AwsFpgaTestBase.xilinx_vitis_examples_dir = AwsFpgaTestBase.git_repo_dir + "/" + AwsFpgaTestBase.xilinx_vitis_examples_prefix_path
          AwsFpgaTestBase.xilinx_vitis_examples_list_file = AwsFpgaTestBase.WORKSPACE + "/vitis_examples_list.json"
  
          if 'WORKSPACE' in os.environ:
              assert os.environ['WORKSPACE'] == AwsFpgaTestBase.git_repo_dir, "WORKSPACE incorrect"
          else:
              os.environ['WORKSPACE'] = AwsFpgaTestBase.WORKSPACE
          if 'BUILD_TAG' not in os.environ:
              os.environ['BUILD_TAG'] = 'test'
              logger.info('Set BUILD_TAG to {}'.format(os.environ['BUILD_TAG']))
          AwsFpgaTestBase.instance_type = aws_fpga_test_utils.get_instance_type()
          AwsFpgaTestBase.num_slots = aws_fpga_test_utils.get_num_fpga_slots(AwsFpgaTestBase.instance_type)
          return
  
      @staticmethod
      def ec2_client():
          if not AwsFpgaTestBase.__ec2_client:
              AwsFpgaTestBase.__ec2_client = boto3.client('ec2')
          return AwsFpgaTestBase.__ec2_client
  
      @staticmethod
      def s3_client():
          if not AwsFpgaTestBase.__s3_client:
              AwsFpgaTestBase.__s3_client = boto3.client('s3')
          return AwsFpgaTestBase.__s3_client
  
      @staticmethod
      def assert_hdk_setup():
          assert 'AWS_FPGA_REPO_DIR' in os.environ, "AWS_FPGA_REPO_DIR not set. source {}/hdk_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['AWS_FPGA_REPO_DIR'] == AwsFpgaTestBase.git_repo_dir, "AWS_FPGA_REPO_DIR not set to the repo dir. source {}/hdk_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert 'HDK_DIR' in os.environ, "HDK_DIR not set. source {}/hdk_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['HDK_DIR'] == os.path.join(AwsFpgaTestBase.git_repo_dir, 'hdk'), "HDK_DIR incorrect. source {}/hdk_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
  
      @staticmethod
      def assert_sdk_setup():
          assert 'SDK_DIR' in os.environ, "SDK_DIR not set. source {}/sdk_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['SDK_DIR'] == os.path.join(AwsFpgaTestBase.git_repo_dir, 'sdk'), "SDK_DIR incorrect. source {}/sdk_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
  
      @staticmethod
      def assert_sdaccel_setup():
          assert 'AWS_FPGA_REPO_DIR' in os.environ, "AWS_FPGA_REPO_DIR not set. source {}/sdaccel_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['AWS_FPGA_REPO_DIR'] == AwsFpgaTestBase.git_repo_dir, "AWS_FPGA_REPO_DIR not set to the repo dir. source {}/sdaccel_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert 'SDK_DIR' in os.environ, "SDK_DIR not set. source {}/sdaccel_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['SDK_DIR'] == os.path.join(AwsFpgaTestBase.git_repo_dir, 'sdk'), "SDK_DIR incorrect. source {}/sdaccel_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert 'SDACCEL_DIR' in os.environ, "SDACCEL_DIR not set. source {}/sdaccel_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['SDACCEL_DIR'] == os.path.join(AwsFpgaTestBase.git_repo_dir, 'SDAccel'), "SDACCEL_DIR incorrect. source {}/sdaccel_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ.get('AWS_PLATFORM') != 'None', "Environment Var AWS_PLATFORM not set. source {}/sdaccel_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ.get('XILINX_SDX') != 'None', "Environment Var XILINX_SDX not set. Please check the AMI."
  
      @staticmethod
      def assert_vitis_setup():
          assert 'AWS_FPGA_REPO_DIR' in os.environ, "AWS_FPGA_REPO_DIR not set. source {}/vitis_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['AWS_FPGA_REPO_DIR'] == AwsFpgaTestBase.git_repo_dir, "AWS_FPGA_REPO_DIR not set to the repo dir. source {}/vitis_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert 'SDK_DIR' in os.environ, "SDK_DIR not set. source {}/vitis_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['SDK_DIR'] == os.path.join(AwsFpgaTestBase.git_repo_dir, 'sdk'), "SDK_DIR incorrect. source {}/vitis_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert 'VITIS_DIR' in os.environ, "VITIS_DIR not set. source {}/vitis_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ['VITIS_DIR'] == os.path.join(AwsFpgaTestBase.git_repo_dir, 'Vitis'), "VITIS_DIR incorrect. source {}/vitis_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ.get('AWS_PLATFORM') != 'None', "Environment Var AWS_PLATFORM not set. source {}/vitis_setup.sh".format(AwsFpgaTestBase.git_repo_dir)
          assert os.environ.get('XILINX_VITIS') != 'None', "Environment Var XILINX_VITIS not set. Please check the AMI."
  
      @staticmethod
      def running_on_f1_instance():
          '''
          Check to see if running on an F1 instance
          '''
          instance_type = aws_fpga_test_utils.get_instance_type()
          return re.match(r'f1\.', instance_type)
  
  
      @staticmethod
      def load_msix_workaround(slot=0):
  
          AwsFpgaTestBase.fpga_clear_local_image(slot)
  
          logger.info("Loading MSI-X workaround into slot {}".format(slot))
          AwsFpgaTestBase.fpga_load_local_image(AwsFpgaTestBase.msix_agfi, slot)
  
          logger.info("Checking slot {} AFI Load status".format(slot))
          assert AwsFpgaTestBase.check_fpga_afi_loaded(AwsFpgaTestBase.msix_agfi, slot), "{} not loaded in slot {}".format(AwsFpgaTestBase.msix_agfi, slot)
  
          AwsFpgaTestBase.fpga_clear_local_image(slot)
  
      @staticmethod
      def run_cmd(cmd, echo=False, check=True):
          if echo:
              logger.info("Running: {}".format(cmd))
          p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
          (stdout_data, stderr_data) = p.communicate()
          stdout_lines = stdout_data.split('\n')
          stderr_lines = stderr_data.split('\n')
          if check and p.returncode:
              logger.error("Cmd failed with rc={}\ncmd: {}\nstdout:\n{}\nstderr:\n{}".format(
                  p.returncode, cmd, stdout_data, stderr_data))
          elif echo:
              logger.info("rc={}\nstdout:\n{}\nstderr:\n{}\n".format(p.returncode, stdout_data, stderr_data))
          return (p.returncode, stdout_lines, stderr_lines)
  
      @staticmethod
      def run_hdk_cmd(cmd, echo=False, check=True):
          source_hdk_cmd = "source {}/hdk_setup.sh &> /dev/null".format(AwsFpgaTestBase.git_repo_dir)
          cmd = source_hdk_cmd + " && " + cmd
          return AwsFpgaTestBase.run_cmd(cmd, echo, check)
  
      @staticmethod
      def run_sdk_cmd(cmd, echo=False, check=True):
          source_sdk_cmd = "source {}/sdk_setup.sh &> /dev/null".format(AwsFpgaTestBase.git_repo_dir)
          cmd = source_sdk_cmd + " && " + cmd
          return AwsFpgaTestBase.run_cmd(cmd, echo, check)
  
      @staticmethod
      def run_sdaccel_cmd(cmd, echo=False, check=True):
          source_sdaccel_cmd = "source {}/sdaccel_setup.sh &> /dev/null".format(AwsFpgaTestBase.git_repo_dir)
          cmd = source_sdaccel_cmd + " && " + cmd
          return AwsFpgaTestBase.run_cmd(cmd, echo, check)
  
      @staticmethod
      def run_vitis_cmd(cmd, echo=False, check=True):
          source_vitis_cmd = "source {}/vitis_setup.sh &> /dev/null".format(AwsFpgaTestBase.git_repo_dir)
          cmd = source_vitis_cmd + " && " + cmd
          return AwsFpgaTestBase.run_cmd(cmd, echo, check)
  
      @staticmethod
      def get_shell_version():
          shell_link = os.path.join(AwsFpgaTestBase.WORKSPACE, 'hdk/common/shell_stable')
          link = basename(os.readlink(shell_link))
          shell_version = re.sub('^shell_v', '0x', link)
          return shell_version
  
      @staticmethod
      def get_cl_dir(cl):
          return "{}/hdk/cl/examples/{}".format(AwsFpgaTestBase.WORKSPACE, cl)
  
      @staticmethod
      def get_cl_to_aws_dir(cl):
          return os.path.join(AwsFpgaTestBase.get_cl_dir(cl), 'build/checkpoints/to_aws')
  
      @staticmethod
      def get_cl_afi_id_filename(cl):
          return os.path.join(AwsFpgaTestBase.get_cl_dir(cl), 'build/create-afi/afi_ids.txt')
  
      @staticmethod
      def get_cl_scripts_dir(cl):
          return os.path.join(AwsFpgaTestBase.get_cl_dir(cl), 'build/scripts')
  
      @staticmethod
      def get_cl_s3_dcp_tag(cl, option_tag, xilinxVersion):
          '''
          @param option_tag: A tag that is unique for each build.
              Required because a CL can be built with different options such as clock recipes.
          '''
          assert option_tag != ''
          assert xilinxVersion != ''
          return "jenkins/{}/cl/{}/{}/{}/dcp".format(os.environ['BUILD_TAG'], xilinxVersion, cl, option_tag)
  
      @staticmethod
      def get_cl_s3_afi_tag(cl, option_tag, xilinxVersion):
          '''
          @param option_tag: A tag that is unique for each build.
              Required because a CL can be built with different options such as clock recipes.
          '''
          assert option_tag != ''
          assert xilinxVersion != ''
          return "jenkins/{}/cl/{}/{}/{}/create-afi/afi_ids.txt".format(os.environ['BUILD_TAG'], xilinxVersion, cl, option_tag)
  
      @staticmethod
      def get_sdaccel_xclbin_dir(examplePath):
          return os.path.join(AwsFpgaTestBase.get_sdaccel_example_fullpath(examplePath=examplePath), 'xclbin')
  
      @staticmethod
      def get_vitis_xclbin_dir(examplePath, target='hw'):
          return os.path.join(AwsFpgaTestBase.get_sdaccel_example_fullpath(examplePath=examplePath), "build_dir.{}.xilinx_aws-vu9p-f1_shell-v04261818_201920_1".format(target))
  
      @staticmethod
      def get_sdaccel_example_s3_root_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx SDAccel example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          @param xilinxVersion: The Xilinx tool version
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          example_relative_path = os.path.relpath(examplePath, AwsFpgaTestBase.xilinx_sdaccel_examples_prefix_path)
          return "jenkins/{}/SDAccel/{}/{}/{}/{}".format(os.environ['BUILD_TAG'], xilinxVersion, rteName, example_relative_path, target)
  
      @staticmethod
      def get_vitis_example_s3_root_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx Vitis example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          @param xilinxVersion: The Xilinx tool version
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          example_relative_path = os.path.relpath(examplePath, AwsFpgaTestBase.xilinx_vitis_examples_prefix_path)
          return "jenkins/{}/Vitis/{}/{}/{}/{}".format(os.environ['BUILD_TAG'], xilinxVersion, rteName, example_relative_path, target)
  
      @staticmethod
      def get_sdaccel_example_s3_xclbin_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx SDAccel example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          @param xilinxVersion: The Xilinx tool version
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          root_tag = AwsFpgaTestBase.get_sdaccel_example_s3_root_tag(examplePath, target, rteName, xilinxVersion)
  
          return "{}/xclbin".format(root_tag)
  
      @staticmethod
      def get_vitis_example_s3_xclbin_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx Vitis example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          @param xilinxVersion: The Xilinx tool version
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          root_tag = AwsFpgaTestBase.get_vitis_example_s3_root_tag(examplePath, target, rteName, xilinxVersion)
  
          return "{}/xclbin".format(root_tag)
  
      @staticmethod
      def get_sdaccel_example_s3_dcp_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx SDAccel example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          @param xilinxVersion: The Xilinx tool version
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          root_tag = AwsFpgaTestBase.get_sdaccel_example_s3_root_tag(examplePath, target, rteName, xilinxVersion)
  
          return "{}/dcp".format(root_tag)
  
      @staticmethod
      def get_vitis_example_s3_dcp_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx Vitis example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          @param xilinxVersion: The Xilinx tool version
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          root_tag = AwsFpgaTestBase.get_vitis_example_s3_root_tag(examplePath, target, rteName, xilinxVersion)
  
          return "{}/dcp".format(root_tag)
  
      @staticmethod
      def get_sdaccel_example_s3_afi_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx SDAccel example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          root_tag = AwsFpgaTestBase.get_sdaccel_example_s3_root_tag(examplePath, target, rteName, xilinxVersion)
  
          return "{}/create-afi/afi-ids.txt".format(root_tag)
  
      @staticmethod
      def get_vitis_example_s3_afi_tag(examplePath, target, rteName, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx Vitis example
          @param target: The target to build. For eg: hw, hw_emu, sw_emu
          @param rteName: The runtime environment
          '''
          assert target != ''
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          root_tag = AwsFpgaTestBase.get_vitis_example_s3_root_tag(examplePath, target, rteName, xilinxVersion)
  
          return "{}/create-afi/afi-ids.txt".format(root_tag)
  
      @staticmethod
      def get_sdaccel_example_run_cmd(examplePath, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx SDAccel example
          @param xilinxVersion: The Xilinx tool version
          '''
          description = AwsFpgaTestBase.get_sdaccel_example_description(examplePath)
          if description.get("em_cmd", None):
              run_cmd = description.get("em_cmd", None)
          else:
              if description.get("host_exe", None):
                  run_cmd = "./{}".format(description.get("host_exe", None))
                  if description.get("cmd_args", None):
                     if "PROJECT" not in description.get("cmd_args", None) and "BUILD" not in description.get("cmd_args", None):
                         if "2019.1" not in xilinxVersion:
                             run_cmd += " {}".format(description.get("cmd_args", None))
                         else:
                             run_cmd += " {}".format(description.get("cmd_args", None).replace(".xclbin",".hw.xilinx_aws-vu9p-f1-04261818_dynamic_5_0.xclbin"))
                     else:
                         if "2019.1" not in xilinxVersion:
                             run_cmd += " {}".format((description.get("cmd_args", None).replace("PROJECT",".")).replace("BUILD","./xclbin"))
                         else:
                             run_cmd += " {}".format(((description.get("cmd_args", None).replace(".xclbin",".hw.xilinx_aws-vu9p-f1-04261818_dynamic_5_0.awsxclbin")).replace("PROJECT",".")).replace("BUILD","./xclbin"))
  
          assert run_cmd is not None, "Could not find run_cmd(em_cmd) or (host_exe) in the example description here {}".format(examplePath)
          
          return run_cmd
  
      @staticmethod
      def get_vitis_example_run_cmd(examplePath, xilinxVersion):
          '''
          @param examplePath: Path of the Xilinx Vitis example
          @param xilinxVersion: The Xilinx tool version
          '''
          description = AwsFpgaTestBase.get_vitis_example_description(examplePath)
  
          host_description = description.get("host", None)
          assert host_description is not None, "Could not find host key in the description.json here {}".format(
              examplePath)
  
          launch_description = description.get("launch", None)
          assert launch_description is not None, "Could not find launch/cmd_args key in the description.json here {}".format(
              examplePath)
  
          if host_description.get("host_exe", None):
              run_cmd = "./{}".format(host_description.get("host_exe", None))
  
          if launch_description[0].get("cmd_args", None):
              run_cmd += " {}".format(((launch_description[0].get("cmd_args", None).replace(".xclbin",
                                                                                          ".awsxclbin")).replace(
                          "PROJECT", ".")).replace("BUILD", "./build_dir.hw.xilinx_aws-vu9p-f1_shell-v04261818_201920_1"))
  
          assert run_cmd is not None, "Could not find run_cmd(em_cmd) or (host_exe) in the example description here {}".format(
              examplePath)
  
          return run_cmd
  
      @staticmethod
      def get_sdaccel_example_description(examplePath):
          '''
          @param examplePath: Path of the Xilinx SDAccel example
          '''
  
          example_description = AwsFpgaTestBase.assert_non_zero_file(os.path.join(AwsFpgaTestBase.get_sdaccel_example_fullpath(examplePath), "description.json"))
  
          with open(example_description) as json_data:
              description = json.load(json_data)
              return description
  
      @staticmethod
      def get_vitis_example_description(examplePath):
          '''
          @param examplePath: Path of the Xilinx Vitis example
          '''
  
          example_description = AwsFpgaTestBase.assert_non_zero_file(os.path.join(AwsFpgaTestBase.get_vitis_example_fullpath(examplePath), "description.json"))
  
          with open(example_description) as json_data:
              description = json.load(json_data)
              return description
  
      @staticmethod
      def get_sdaccel_example_fullpath(examplePath):
          return "{}/{}/".format(AwsFpgaTestBase.WORKSPACE, examplePath)
  
      @staticmethod
      def get_vitis_example_fullpath(examplePath):
          return "{}/{}/".format(AwsFpgaTestBase.WORKSPACE, examplePath)
  
      @staticmethod
      def fetch_sdaccel_xclbin_folder_from_s3(examplePath, rteName, xilinxVersion):
          cwd = os.getcwd()
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
  
          os.chdir(AwsFpgaTestBase.get_sdaccel_example_fullpath(examplePath))
          rc = os.system("aws s3 cp s3://{}/{} {} --recursive".format(AwsFpgaTestBase.s3_bucket, AwsFpgaTestBase.get_sdaccel_example_s3_xclbin_tag(examplePath=examplePath, target="hw", rteName=rteName, xilinxVersion=xilinxVersion), AwsFpgaTestBase.get_sdaccel_xclbin_dir(examplePath=examplePath)))
          assert rc == 0, "Error while copying from s3://{}/{} to {}".format(AwsFpgaTestBase.s3_bucket, AwsFpgaTestBase.get_sdaccel_example_s3_xclbin_tag(examplePath=examplePath, target="hw", rteName=rteName, xilinxVersion=xilinxVersion), AwsFpgaTestBase.get_sdaccel_xclbin_dir(examplePath=examplePath))
          xclbin_path = AwsFpgaTestBase.get_sdaccel_xclbin_dir(examplePath=examplePath)
  
          logger.debug(xclbin_path)
          assert os.path.exists(xclbin_path), "SDAccel Example xclbin path={} does not exist".format(xclbin_path)
  
          os.chdir(cwd)
          return xclbin_path
  
      @staticmethod
      def fetch_vitis_xclbin_folder_from_s3(examplePath, rteName, xilinxVersion):
          cwd = os.getcwd()
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
  
          os.chdir(AwsFpgaTestBase.get_vitis_example_fullpath(examplePath))
          rc = os.system("aws s3 cp s3://{}/{} {} --recursive".format(AwsFpgaTestBase.s3_bucket, AwsFpgaTestBase.get_vitis_example_s3_xclbin_tag(examplePath=examplePath, target="hw", rteName=rteName, xilinxVersion=xilinxVersion), AwsFpgaTestBase.get_vitis_xclbin_dir(examplePath=examplePath)))
          assert rc == 0, "Error while copying from s3://{}/{} to {}".format(AwsFpgaTestBase.s3_bucket, AwsFpgaTestBase.get_vitis_example_s3_xclbin_tag(examplePath=examplePath, target="hw", rteName=rteName, xilinxVersion=xilinxVersion), AwsFpgaTestBase.get_vitis_xclbin_dir(examplePath=examplePath))
          xclbin_path = AwsFpgaTestBase.get_vitis_xclbin_dir(examplePath=examplePath)
  
          logger.debug(xclbin_path)
          assert os.path.exists(xclbin_path), "Vitis Example xclbin path={} does not exist".format(xclbin_path)
  
          os.chdir(cwd)
          return xclbin_path
  
      @staticmethod
      def get_sdaccel_xclbin_file(examplePath, rteName, xilinxVersion):
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          xclbin_path = AwsFpgaTestBase.fetch_sdaccel_xclbin_folder_from_s3(examplePath, rteName, xilinxVersion)
          logger.info("Checking that a non zero size xclbin file exists in {}".format(xclbin_path))
  
          xclbin = AwsFpgaTestBase.assert_non_zero_file(os.path.join(xclbin_path, "*.{}.*.xclbin".format("hw")))
          return xclbin
  
      @staticmethod
      def get_vitis_xclbin_file(examplePath, rteName, xilinxVersion):
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
          xclbin_path = AwsFpgaTestBase.fetch_vitis_xclbin_folder_from_s3(examplePath, rteName, xilinxVersion)
          logger.info("Checking that a non zero size xclbin file exists in {}".format(xclbin_path))
  
          xclbin = AwsFpgaTestBase.assert_non_zero_file(os.path.join(xclbin_path, "*.xclbin"))
          return xclbin
  
      @staticmethod
      def get_sdaccel_aws_xclbin_file(examplePath, rteName, xilinxVersion):
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
  
          xclbin_path = AwsFpgaTestBase.fetch_sdaccel_xclbin_folder_from_s3(examplePath, rteName, xilinxVersion)
          logger.info("Checking that a non zero size awsxclbin file exists in {}".format(xclbin_path))
          aws_xclbin = AwsFpgaTestBase.assert_non_zero_file(os.path.join(xclbin_path, "*.{}.*.awsxclbin".format("hw")))
          return aws_xclbin
  
      @staticmethod
      def get_vitis_aws_xclbin_file(examplePath, rteName, xilinxVersion):
          assert examplePath != ''
          assert rteName != ''
          assert xilinxVersion != ''
  
          xclbin_path = AwsFpgaTestBase.fetch_vitis_xclbin_folder_from_s3(examplePath, rteName, xilinxVersion)
          logger.info("Checking that a non zero size awsxclbin file exists in {}".format(xclbin_path))
          aws_xclbin = AwsFpgaTestBase.assert_non_zero_file(os.path.join(xclbin_path, "*.awsxclbin"))
          return aws_xclbin
  
      @staticmethod
      def assert_afi_available(afi):
          # Check the status of the afi
          logger.info("Checking the status of {}".format(afi))
          afi_state = AwsFpgaTestBase.ec2_client().describe_fpga_images(FpgaImageIds=[afi])['FpgaImages'][0]['State']['Code']
          logger.info("{} state={}".format(afi, afi_state))
          assert afi_state == 'available'
  
      @staticmethod
      def assert_afi_public(afi):
          # Check the status of the afi
          logger.info("Checking that {} is public".format(afi))
          loadPermissions = AwsFpgaTestBase.ec2_client().describe_fpga_image_attribute(FpgaImageId=afi, Attribute='loadPermission')['FpgaImageAttribute']['LoadPermissions']
          logger.info("{} loadPermissions:".format(afi))
          for loadPermission in loadPermissions:
              if 'UserId' in loadPermission:
                  logger.info("  UserId={}".format(loadPermission['UserId']))
              else:
                  logger.info("  Group={}".format(loadPermission['Group']))
          is_public = AwsFpgaTestBase.ec2_client().describe_fpga_images(FpgaImageIds=[afi])['FpgaImages'][0]['Public']
          logger.info("  Public={}".format(is_public))
          assert is_public, "{} is not public. To make public:\n{}".format(afi,
              "aws ec2 modify-fpga-image-attribute --fpga-image-id {} --load-permission \'Add=[{{Group=all}}]\'".format(afi))
  
      @staticmethod
      def get_agfi_from_readme(cl):
          cl_dir = "{}/hdk/cl/examples/{}".format(AwsFpgaTestBase.WORKSPACE, cl)
          assert os.path.exists(cl_dir)
          agfi = subprocess.check_output("cat {}/README.md | grep \'Pre-generated AGFI ID\' | cut -d \"|\" -f 3".format(cl_dir), shell=True).lstrip().rstrip()
          afi = subprocess.check_output("cat {}/README.md | grep \'Pre-generated AFI ID\'  | cut -d \"|\" -f 3".format(cl_dir), shell=True).lstrip().rstrip()
          logger.info("AGFI from README: {}".format(agfi))
          logger.info("AFI  from README: {}".format(afi))
          return (agfi, afi)
  
      @staticmethod
      def exec_as_user(as_root, command):
          if as_root:
              return "sudo {}".format(command)
          else:
              return command
  
      @staticmethod
      def fpga_clear_local_image(slot, request_timeout=6000, sync_timeout=180,
              as_root=True):
          logger.info("Clearing FPGA slot {}".format(slot))
          cmd = "{} -S {} --request-timeout {} --sync-timeout {}".format(
                  AwsFpgaTestBase.exec_as_user(as_root, "fpga-clear-local-image"),  slot,
                  request_timeout, sync_timeout)
          (rc, stdout_lines, stderr_lines) = AwsFpgaTestBase.run_cmd(cmd)
          assert rc == 0, "Clearing FPGA slot {} failed.".format(slot)
  
      @staticmethod
      def fpga_load_local_image(agfi, slot, request_timeout=6000,
              sync_timeout=180, as_root=True):
          logger.info("Loading {} into slot {}".format(agfi, slot))
          cmd = "{} -S {} -I {} --request-timeout {} --sync-timeout {}".format(
                  AwsFpgaTestBase.exec_as_user(as_root, "fpga-load-local-image"), slot, agfi,
                  request_timeout, sync_timeout)
          (rc, stdout_lines, stderr_lines) = AwsFpgaTestBase.run_cmd(cmd)
          assert rc == 0, "Failed to load {} in slot {}.".format(agfi, slot)
  
      @staticmethod
      def check_fpga_afi_loaded(agfi, slot):
          fpgaLocalImage = aws_fpga_test_utils.fpga_describe_local_image(slot)
          assert fpgaLocalImage.statusName == 'loaded', "{} FPGA StatusName != loaded: {}".format(agfi, fpgaLocalImage.statusName)
          assert fpgaLocalImage.statusCode == '0', "{} status code != 0: {}".format(agfi, fpgaLocalImage.statusCode)
          assert fpgaLocalImage.errorName == 'ok', "{} FPGA ErrorName != ok: {}".format(agfi, fpgaLocalImage.errorName)
          assert fpgaLocalImage.errorCode == '0', "{} ErrorCode != 0: {}".format(agfi, fpgaLocalImage.errorCode)
          assert fpgaLocalImage.agfi == agfi, "Expected {}, actual {}".format(agfi, fpgaLocalImage.agfi)
          return fpgaLocalImage
  
      @staticmethod
      def fpga_get_virtual_led(slot, remove_dashes=False, as_root=True):
          cmd = "{} -S {}".format(AwsFpgaTestBase.exec_as_user(as_root, "fpga-get-virtual-led"),
              slot)
          (rc, stdout_lines, stderr_lines) = AwsFpgaTestBase.run_cmd(cmd)
          assert rc == 0, "Failed to get virtual LEDs from slot {}.".format(slot)
          value = stdout_lines[1]
          if remove_dashes:
              value = re.sub('-', '', value)
          return value
  
      @staticmethod
      def fpga_get_virtual_dip_switch(slot, remove_dashes=False, as_root=True):
          cmd = "{} -S {}".format(AwsFpgaTestBase.exec_as_user(as_root, "fpga-get-virtual-dip-switch"), slot)
          (rc, stdout_lines, stderr_lines) = AwsFpgaTestBase.run_cmd(cmd)
          assert rc == 0, "Failed to get virtual DIP switches from slot {}.".format(slot)
          value = stdout_lines[1]
          if remove_dashes:
              value = re.sub('-', '', value)
          return value
  
      @staticmethod
      def fpga_set_virtual_dip_switch(value, slot, as_root=True):
          value = re.sub('-', '', value)
          cmd = "{} -S {} -D {}".format(AwsFpgaTestBase.exec_as_user(as_root, "fpga-set-virtual-dip-switch"),
              slot, value)
          (rc, stdout_lines, stderr_lines) = AwsFpgaTestBase.run_cmd(cmd)
          assert rc == 0, "Failed to set virtual DIP switches in slot {} to {}.".format(slot, value)
  
      @staticmethod
      def assert_non_zero_file(filter):
  
          filenames = glob.glob(filter)
  
          assert len(filenames) > 0, "No {} file found in {}".format(filter, os.getcwd())
          assert len(filenames) == 1, "More than 1 {} file found: {}\n{}".format(filter, len(filenames), filenames)
  
          filename = filenames[0]
          assert os.stat(filename).st_size != 0, "{} is 0 size".format(filename)
          return filename
  
      @staticmethod
      def get_fio_tool_root():
          return os.path.join(AwsFpgaTestBase.WORKSPACE, 'sdk/tests/fio_dma_tools')
  
      @staticmethod
      def get_fio_tool_install_path():
          return os.path.join(AwsFpgaTestBase.get_fio_tool_root(), 'scripts/fio_github_repo')
  
      @staticmethod
      def get_fio_tool_install_script():
          return os.path.join(AwsFpgaTestBase.get_fio_tool_root(), 'scripts/fio_install.py')
  
      @staticmethod
      def get_fio_tool_run_script():
          return os.path.join(AwsFpgaTestBase.get_fio_tool_root(), 'scripts/fio')
  
      @staticmethod
      def get_fio_verify_script(driver='xdma'):
          return os.path.join(AwsFpgaTestBase.get_fio_tool_root(), "scripts/{}_4-ch_4-1M_verify.fio".format(driver))
  
      @staticmethod
      def get_fio_read_benchmark_script(driver='xdma'):
          return os.path.join(AwsFpgaTestBase.get_fio_tool_root(), "scripts/{}_4-ch_4-1M_read.fio".format(driver))
  
      @staticmethod
      def get_fio_write_benchmark_script(driver='xdma'):
          return os.path.join(AwsFpgaTestBase.get_fio_tool_root(), "scripts/{}_4-ch_4-1M_write.fio".format(driver))
  
      @staticmethod
      def setup_fio_tools():
          '''Install and setup fio tools'''
          # If downloaded repo already, exists, delete it so we can fetch again
          if os.path.exists(AwsFpgaTestBase.get_fio_tool_install_path()):
              AwsFpgaTestBase.run_cmd("rm -rf {}".format(AwsFpgaTestBase.get_fio_tool_install_path()), echo=True)
  
          logger.info("Installing fio_dma_tools")
  
          (rc, stdout_lines, stderr_lines) = AwsFpgaTestBase.run_cmd("python {} {}".format(AwsFpgaTestBase.get_fio_tool_install_script(), AwsFpgaTestBase.get_fio_tool_install_path()), echo=True)
          assert rc == 0
          assert os.path.exists("{}".format(AwsFpgaTestBase.get_fio_tool_run_script()))
  
          (rc, stdout_lines, stderr_lines) = AwsFpgaTestBase.run_cmd("chmod +x {}".format(AwsFpgaTestBase.get_fio_tool_run_script()))
          assert rc == 0
  