Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_output_stdout(self):
output, _, _, _, _ = self.generate_tables_and_check_produced_files(
[
result_file("test.2015-03-03_1613.results.predicateAnalysis.xml"),
"-f",
"csv",
"-q",
],
table_prefix="test.2015-03-03_1613.results.predicateAnalysis",
formats=[],
output_path="-",
)
expected = benchexec.util.read_file(
here, "expected", "test.2015-03-03_1613.results.predicateAnalysis" + ".csv"
)
self.assertMultiLineEqual(output.strip(), expected)
here,
"expected",
(result_diff_prefix or diff_prefix) + "." + ending,
]
(
output,
html_file,
html_diff_file,
csv_file,
csv_diff_file,
) = self.generate_tables_and_check_produced_files(
args, table_prefix, diff_prefix
)
generated_csv = benchexec.util.read_file(csv_file)
self.assert_file_content_equals(generated_csv, expected_file_name("csv"))
generated_html = self.read_table_from_html(html_file)
self.assert_file_content_equals(generated_html, expected_file_name("html"))
if diff_prefix:
generated_csv_diff = benchexec.util.read_file(csv_diff_file)
self.assert_file_content_equals(
generated_csv_diff, expected_diff_file_name("csv")
)
generated_html_diff = self.read_table_from_html(html_diff_file)
self.assert_file_content_equals(
generated_html_diff, expected_diff_file_name("html")
)
def read_table_from_html(self, file):
content = benchexec.util.read_file(file)
# only keep table
content = content[
content.index("const data = {") + 13 : content.index("\n};") + 2
]
# Pretty-print JSON for better diffs
content = json.dumps(json.loads(content), indent=" ", sort_keys=True)
content = content.replace(
'\n "version": "{}"\n'.format(benchexec.__version__),
'\n "version": "(test)"\n',
)
return content
def is_turbo_boost_enabled():
"""
Check whether Turbo Boost (scaling CPU frequency beyond nominal frequency)
is active on this system.
@return: A bool, or None if Turbo Boost is not supported.
"""
try:
if os.path.exists(_TURBO_BOOST_FILE):
boost_enabled = int(util.read_file(_TURBO_BOOST_FILE))
if not (0 <= boost_enabled <= 1):
raise ValueError(
"Invalid value {} for turbo boost activation".format(boost_enabled)
)
return boost_enabled != 0
if os.path.exists(_TURBO_BOOST_FILE_PSTATE):
boost_disabled = int(util.read_file(_TURBO_BOOST_FILE_PSTATE))
if not (0 <= boost_disabled <= 1):
raise ValueError(
"Invalid value {} for turbo boost activation".format(boost_enabled)
)
return boost_disabled != 1
except ValueError as e:
sys.exit("Could not read turbo-boost information from kernel: {0}".format(e))
def is_turbo_boost_enabled():
"""
Check whether Turbo Boost (scaling CPU frequency beyond nominal frequency)
is active on this system.
@return: A bool, or None if Turbo Boost is not supported.
"""
try:
if os.path.exists(_TURBO_BOOST_FILE):
boost_enabled = int(util.read_file(_TURBO_BOOST_FILE))
if not (0 <= boost_enabled <= 1):
raise ValueError(
"Invalid value {} for turbo boost activation".format(boost_enabled)
)
return boost_enabled != 0
if os.path.exists(_TURBO_BOOST_FILE_PSTATE):
boost_disabled = int(util.read_file(_TURBO_BOOST_FILE_PSTATE))
if not (0 <= boost_disabled <= 1):
raise ValueError(
"Invalid value {} for turbo boost activation".format(boost_enabled)
)
return boost_disabled != 1
except ValueError as e:
sys.exit("Could not read turbo-boost information from kernel: {0}".format(e))
def get_value(self, subsystem, option):
"""
Read the given value from the given subsystem.
Do not include the subsystem name in the option name.
Only call this method if the given subsystem is available.
"""
assert subsystem in self, "Subsystem {} is missing".format(subsystem)
return util.read_file(self.per_subsystem[subsystem], subsystem + "." + option)
def get_cpu_package_for_core(core):
"""Get the number of the physical package (socket) a core belongs to."""
return int(
util.read_file(
"/sys/devices/system/cpu/cpu{0}/topology/physical_package_id".format(core)
)
"""
logging.debug("I'm loading the benchmark %s.", benchmark_file)
self.config = config
self.benchmark_file = benchmark_file
self.base_dir = os.path.dirname(self.benchmark_file)
# get benchmark-name
self.name = os.path.basename(benchmark_file)[:-4] # remove ending ".xml"
if config.name:
self.name += "." + config.name
self.description = None
if config.description_file is not None:
try:
self.description = util.read_file(config.description_file)
except (OSError, UnicodeDecodeError) as e:
raise BenchExecException(
"File '{}' given for description could not be read: {}".format(
config.description_file, e
)
)
self.start_time = start_time
self.instance = start_time.strftime(util.TIMESTAMP_FILENAME_FORMAT)
self.output_base_name = config.output_path + self.name + "." + self.instance
self.log_folder = self.output_base_name + ".logfiles" + os.path.sep
self.log_zip = self.output_base_name + ".logfiles.zip"
self.result_files_folder = self.output_base_name + ".files"
# parse XML
def __init__(self, cores=None):
"""
Create an instance that monitors the given list of cores (or all CPUs).
"""
self.cpu_throttle_count = {}
cpu_pattern = "[{0}]".format(",".join(map(str, cores))) if cores else "*"
for file in glob.glob(
"/sys/devices/system/cpu/cpu{}/thermal_throttle/*_throttle_count".format(
cpu_pattern
)
):
try:
self.cpu_throttle_count[file] = int(util.read_file(file))
except Exception as e:
logging.warning(
"Cannot read throttling count of CPU from kernel: %s", e
)
def has_throttled(self):
"""
Check whether any of the CPU cores monitored by this instance has
throttled since this instance was created.
@return a boolean value
"""
for file, value in self.cpu_throttle_count.items():
try:
new_value = int(util.read_file(file))
if new_value > value:
return True
except Exception as e:
logging.warning(
"Cannot read throttling count of CPU from kernel: %s", e
)
return False