tests/functional: Convert the acpi-bits test into a standalone test

Mostly a straight-forward conversion. Looks like we can simply drop
the avocado datadrainer stuff when not using the avocado framework
anymore.

Reviewed-by: Philippe Mathieu-Daudé <philmd@linaro.org>
Tested-by: Philippe Mathieu-Daudé <philmd@linaro.org>
Message-ID: <20240830133841.142644-30-thuth@redhat.com>
Signed-off-by: Thomas Huth <thuth@redhat.com>

tests/functional/acpi-bits/bits-config/bits-cfg.txt

@@ -0,0 +1,18 @@
+# BITS configuration file
+[bits]
+
+# To run BITS in batch mode, set batch to a list of one or more of the
+# following keywords; BITS will then run all of the requested operations, then
+# save the log file to disk.
+#
+# test: Run the full BITS testsuite.
+# acpi: Dump all ACPI structures.
+# smbios: Dump all SMBIOS structures.
+#
+# Leave batch set to an empty string to disable batch mode.
+# batch =
+
+# Uncomment the following to run all available batch operations
+# please take a look at boot/python/init.py in bits zip file
+# to see how these options are parsed and used.
+batch = test acpi smbios

tests/functional/acpi-bits/bits-tests/smilatency.py2

@@ -0,0 +1,107 @@
+# Copyright (c) 2015, Intel Corporation
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+#     * Redistributions of source code must retain the above copyright notice,
+#       this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above copyright notice,
+#       this list of conditions and the following disclaimer in the documentation
+#       and/or other materials provided with the distribution.
+#     * Neither the name of Intel Corporation nor the names of its contributors
+#       may be used to endorse or promote products derived from this software
+#       without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+# This script runs only from the biosbits VM.
+
+"""SMI latency test."""
+
+import bits
+from collections import namedtuple
+import testsuite
+import time
+import usb
+
+def register_tests():
+     pass
+#    testsuite.add_test("SMI latency test", smi_latency);
+#    testsuite.add_test("SMI latency test with USB disabled via BIOS handoff", test_with_usb_disabled, runall=False);
+
+def smi_latency():
+    MSR_SMI_COUNT = 0x34
+
+    print "Warning: touching the keyboard can affect the results of this test."
+
+    tsc_per_sec = bits.tsc_per_sec()
+    tsc_per_usec = tsc_per_sec / (1000 * 1000)
+    bins = [long(tsc_per_usec * 10**i) for i in range(9)]
+    bin_descs = [
+        "0     < t <=   1us",
+        "1us   < t <=  10us",
+        "10us  < t <= 100us",
+        "100us < t <=   1ms",
+        "1ms   < t <=  10ms",
+        "10ms  < t <= 100ms",
+        "100ms < t <=   1s ",
+        "1s    < t <=  10s ",
+        "10s   < t <= 100s ",
+        "100s  < t         ",
+    ]
+
+    print "Starting test. Wait here, I will be back in 15 seconds."
+    (max_latency, smi_count_delta, bins) = bits.smi_latency(long(15 * tsc_per_sec), bins)
+    BinType = namedtuple('BinType', ("max", "total", "count", "times"))
+    bins = [BinType(*b) for b in bins]
+
+    testsuite.test("SMI latency < 150us to minimize risk of OS timeouts", max_latency / tsc_per_usec <= 150)
+    if not testsuite.show_detail():
+        return
+
+    for bin, desc in zip(bins, bin_descs):
+        if bin.count == 0:
+            continue
+        testsuite.print_detail("{}; average = {}; count = {}".format(desc, bits.format_tsc(bin.total/bin.count), bin.count))
+        deltas = (bits.format_tsc(t2 - t1) for t1,t2 in zip(bin.times, bin.times[1:]))
+        testsuite.print_detail(" Times between first few observations: {}".format(" ".join("{:>6}".format(delta) for delta in deltas)))
+
+    if smi_count_delta is not None:
+        testsuite.print_detail("{} SMI detected using MSR_SMI_COUNT (MSR {:#x})".format(smi_count_delta, MSR_SMI_COUNT))
+
+    testsuite.print_detail("Summary of impact: observed maximum latency = {}".format(bits.format_tsc(max_latency)))
+
+def test_with_usb_disabled():
+    if usb.handoff_to_os():
+        smi_latency()
+
+def average_io_smi(port, value, count):
+    def f():
+        tsc_start = bits.rdtsc()
+        bits.outb(port, value)
+        return bits.rdtsc() - tsc_start
+    counts = [f() for i in range(count)]
+    return sum(counts)/len(counts)
+
+def time_io_smi(port=0xb2, value=0, count=1000):
+    count_for_estimate = 10
+    start = time.time()
+    average_io_smi(port, value, count_for_estimate)
+    avg10 = time.time() - start
+    estimate = avg10 * count/count_for_estimate
+    if estimate > 1:
+        print "Running test, estimated time: {}s".format(int(estimate))
+    average = average_io_smi(port, value, count)
+    print "Average of {} SMIs (via outb, port={:#x}, value={:#x}): {}".format(count, port, value, bits.format_tsc(average))

tests/functional/acpi-bits/bits-tests/testacpi.py2

@@ -0,0 +1,287 @@
+# Copyright (c) 2015, Intel Corporation
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+#     * Redistributions of source code must retain the above copyright notice,
+#       this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above copyright notice,
+#       this list of conditions and the following disclaimer in the documentation
+#       and/or other materials provided with the distribution.
+#     * Neither the name of Intel Corporation nor the names of its contributors
+#       may be used to endorse or promote products derived from this software
+#       without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+# This script runs only from the biosbits VM.
+
+"""Tests for ACPI"""
+
+import acpi
+import bits
+import bits.mwait
+import struct
+import testutil
+import testsuite
+import time
+
+def register_tests():
+    testsuite.add_test("ACPI _MAT (Multiple APIC Table Entry) under Processor objects", test_mat, submenu="ACPI Tests")
+#    testsuite.add_test("ACPI _PSS (Pstate) table conformance tests", test_pss, submenu="ACPI Tests")
+#    testsuite.add_test("ACPI _PSS (Pstate) runtime tests", test_pstates, submenu="ACPI Tests")
+    testsuite.add_test("ACPI DSDT (Differentiated System Description Table)", test_dsdt, submenu="ACPI Tests")
+    testsuite.add_test("ACPI FACP (Fixed ACPI Description Table)", test_facp, submenu="ACPI Tests")
+    testsuite.add_test("ACPI HPET (High Precision Event Timer Table)", test_hpet, submenu="ACPI Tests")
+    testsuite.add_test("ACPI MADT (Multiple APIC Description Table)", test_apic, submenu="ACPI Tests")
+    testsuite.add_test("ACPI MPST (Memory Power State Table)", test_mpst, submenu="ACPI Tests")
+    testsuite.add_test("ACPI RSDP (Root System Description Pointer Structure)", test_rsdp, submenu="ACPI Tests")
+    testsuite.add_test("ACPI XSDT (Extended System Description Table)", test_xsdt, submenu="ACPI Tests")
+
+def test_mat():
+    cpupaths = acpi.get_cpupaths()
+    apic = acpi.parse_apic()
+    procid_apicid = apic.procid_apicid
+    uid_x2apicid = apic.uid_x2apicid
+    for cpupath in cpupaths:
+        # Find the ProcId defined by the processor object
+        processor = acpi.evaluate(cpupath)
+        # Find the UID defined by the processor object's _UID method
+        uid = acpi.evaluate(cpupath + "._UID")
+        mat_buffer = acpi.evaluate(cpupath + "._MAT")
+        if mat_buffer is None:
+            continue
+        # Process each _MAT subtable
+        mat = acpi._MAT(mat_buffer)
+        for index, subtable in enumerate(mat):
+            if subtable.subtype == acpi.MADT_TYPE_LOCAL_APIC:
+                if subtable.flags.bits.enabled:
+                    testsuite.test("{} Processor declaration ProcId = _MAT ProcId".format(cpupath), processor.ProcId == subtable.proc_id)
+                    testsuite.print_detail("{} ProcId ({:#02x}) != _MAT ProcId ({:#02x})".format(cpupath, processor.ProcId, subtable.proc_id))
+                    testsuite.print_detail("Processor Declaration: {}".format(processor))
+                    testsuite.print_detail("_MAT entry[{}]: {}".format(index, subtable))
+                    if testsuite.test("{} with local APIC in _MAT has local APIC in MADT".format(cpupath), processor.ProcId in procid_apicid):
+                        testsuite.test("{} ApicId derived using Processor declaration ProcId = _MAT ApicId".format(cpupath), procid_apicid[processor.ProcId] == subtable.apic_id)
+                        testsuite.print_detail("{} ApicId derived from MADT ({:#02x}) != _MAT ApicId ({:#02x})".format(cpupath, procid_apicid[processor.ProcId], subtable.apic_id))
+                        testsuite.print_detail("Processor Declaration: {}".format(processor))
+                        testsuite.print_detail("_MAT entry[{}]: {}".format(index, subtable))
+            if subtable.subtype == acpi.MADT_TYPE_LOCAL_X2APIC:
+                if subtable.flags.bits.enabled:
+                    if testsuite.test("{} with x2Apic in _MAT has _UID".format(cpupath), uid is not None):
+                        testsuite.test("{}._UID = _MAT UID".format(cpupath), uid == subtable.uid)
+                        testsuite.print_detail("{}._UID ({:#x}) != _MAT UID ({:#x})".format(cpupath, uid, subtable.uid))
+                        testsuite.print_detail("_MAT entry[{}]: {}".format(index, subtable))
+                    if testsuite.test("{} with _MAT x2Apic has x2Apic in MADT".format(cpupath), subtable.uid in uid_x2apicid):
+                        testsuite.test("{} x2ApicId derived from MADT using UID = _MAT x2ApicId".format(cpupath), uid_x2apicid[subtable.uid] == subtable.x2apicid)
+                        testsuite.print_detail("{} x2ApicId derived from MADT ({:#02x}) != _MAT x2ApicId ({:#02x})".format(cpupath, uid_x2apicid[subtable.uid], subtable.x2apicid))
+                        testsuite.print_detail("_MAT entry[{}]: {}".format(index, subtable))
+
+def test_pss():
+    uniques = acpi.parse_cpu_method("_PSS")
+    # We special-case None here to avoid a double-failure for CPUs without a _PSS
+    testsuite.test("_PSS must be identical for all CPUs", len(uniques) <= 1 or (len(uniques) == 2 and None in uniques))
+    for pss, cpupaths in uniques.iteritems():
+        if not testsuite.test("_PSS must exist", pss is not None):
+            testsuite.print_detail(acpi.factor_commonprefix(cpupaths))
+            testsuite.print_detail('No _PSS exists')
+            continue
+
+        if not testsuite.test("_PSS must not be empty", pss.pstates):
+            testsuite.print_detail(acpi.factor_commonprefix(cpupaths))
+            testsuite.print_detail('_PSS is empty')
+            continue
+
+        testsuite.print_detail(acpi.factor_commonprefix(cpupaths))
+        for index, pstate in enumerate(pss.pstates):
+            testsuite.print_detail("P[{}]: {}".format(index, pstate))
+
+        testsuite.test("_PSS must contain at most 16 Pstates", len(pss.pstates) <= 16)
+        testsuite.test("_PSS must have no duplicate Pstates", len(pss.pstates) == len(set(pss.pstates)))
+
+        frequencies = [p.core_frequency for p in pss.pstates]
+        testsuite.test("_PSS must list Pstates in descending order of frequency", frequencies == sorted(frequencies, reverse=True))
+
+        testsuite.test("_PSS must have Pstates with no duplicate frequencies", len(frequencies) == len(set(frequencies)))
+
+        dissipations = [p.power for p in pss.pstates]
+        testsuite.test("_PSS must list Pstates in descending order of power dissipation", dissipations == sorted(dissipations, reverse=True))
+
+def test_pstates():
+    """Execute and verify frequency for each Pstate in the _PSS"""
+    IA32_PERF_CTL = 0x199
+    with bits.mwait.use_hint(), bits.preserve_msr(IA32_PERF_CTL):
+        cpupath_procid = acpi.find_procid()
+        cpupath_uid = acpi.find_uid()
+        apic = acpi.parse_apic()
+        procid_apicid = apic.procid_apicid
+        uid_x2apicid = apic.uid_x2apicid
+        def cpupath_apicid(cpupath):
+            if procid_apicid is not None:
+                procid = cpupath_procid.get(cpupath, None)
+                if procid is not None:
+                    apicid = procid_apicid.get(procid, None)
+                    if apicid is not None:
+                        return apicid
+            if uid_x2apicid is not None:
+                uid = cpupath_uid.get(cpupath, None)
+                if uid is not None:
+                    apicid = uid_x2apicid.get(uid, None)
+                    if apicid is not None:
+                        return apicid
+            return bits.cpus()[0]
+
+        bclk = testutil.adjust_to_nearest(bits.bclk(), 100.0/12) * 1000000
+
+        uniques = acpi.parse_cpu_method("_PSS")
+        for pss, cpupaths in uniques.iteritems():
+            if not testsuite.test("_PSS must exist", pss is not None):
+                testsuite.print_detail(acpi.factor_commonprefix(cpupaths))
+                testsuite.print_detail('No _PSS exists')
+                continue
+
+            for n, pstate in enumerate(pss.pstates):
+                for cpupath in cpupaths:
+                    apicid = cpupath_apicid(cpupath)
+                    if apicid is None:
+                        print 'Failed to find apicid for cpupath {}'.format(cpupath)
+                        continue
+                    bits.wrmsr(apicid, IA32_PERF_CTL, pstate.control)
+
+                # Detecting Turbo frequency requires at least 2 pstates
+                # since turbo frequency = max non-turbo frequency + 1
+                turbo = False
+                if len(pss.pstates) >= 2:
+                    turbo = (n == 0 and pstate.core_frequency == (pss.pstates[1].core_frequency + 1))
+                    if turbo:
+                        # Needs to busywait, not sleep
+                        start = time.time()
+                        while (time.time() - start < 2):
+                            pass
+
+                for duration in (0.1, 1.0):
+                    frequency_data = bits.cpu_frequency(duration)
+                    # Abort the test if no cpu frequency is not available
+                    if frequency_data is None:
+                        continue
+                    aperf = frequency_data[1]
+                    aperf = testutil.adjust_to_nearest(aperf, bclk/2)
+                    aperf = int(aperf / 1000000)
+                    if turbo:
+                        if aperf >= pstate.core_frequency:
+                            break
+                    else:
+                        if aperf == pstate.core_frequency:
+                            break
+
+                if turbo:
+                    testsuite.test("P{}: Turbo measured frequency {} >= expected {} MHz".format(n, aperf, pstate.core_frequency), aperf >= pstate.core_frequency)
+                else:
+                    testsuite.test("P{}: measured frequency {} MHz == expected {} MHz".format(n, aperf, pstate.core_frequency), aperf == pstate.core_frequency)
+
+def test_psd_thread_scope():
+    uniques = acpi.parse_cpu_method("_PSD")
+    if not testsuite.test("_PSD (P-State Dependency) must exist for each processor", None not in uniques):
+        testsuite.print_detail(acpi.factor_commonprefix(uniques[None]))
+        testsuite.print_detail('No _PSD exists')
+        return
+    unique_num_dependencies = {}
+    unique_num_entries = {}
+    unique_revision = {}
+    unique_domain = {}
+    unique_coordination_type = {}
+    unique_num_processors = {}
+    for value, cpupaths in uniques.iteritems():
+        unique_num_dependencies.setdefault(len(value.dependencies), []).extend(cpupaths)
+        unique_num_entries.setdefault(value.dependencies[0].num_entries, []).extend(cpupaths)
+        unique_revision.setdefault(value.dependencies[0].revision, []).extend(cpupaths)
+        unique_domain.setdefault(value.dependencies[0].domain, []).extend(cpupaths)
+        unique_coordination_type.setdefault(value.dependencies[0].coordination_type, []).extend(cpupaths)
+        unique_num_processors.setdefault(value.dependencies[0].num_processors, []).extend(cpupaths)
+    def detail(d, fmt):
+        for value, cpupaths in sorted(d.iteritems(), key=(lambda (k,v): v)):
+            testsuite.print_detail(acpi.factor_commonprefix(cpupaths))
+            testsuite.print_detail(fmt.format(value))
+
+    testsuite.test('Dependency count for each processor must be 1', unique_num_dependencies.keys() == [1])
+    detail(unique_num_dependencies, 'Dependency count for each processor = {} (Expected 1)')
+    testsuite.test('_PSD.num_entries must be 5', unique_num_entries.keys() == [5])
+    detail(unique_num_entries, 'num_entries = {} (Expected 5)')
+    testsuite.test('_PSD.revision must be 0', unique_revision.keys() == [0])
+    detail(unique_revision, 'revision = {}')
+    testsuite.test('_PSD.coordination_type must be 0xFE (HW_ALL)', unique_coordination_type.keys() == [0xfe])
+    detail(unique_coordination_type, 'coordination_type = {:#x} (Expected 0xFE HW_ALL)')
+    testsuite.test('_PSD.domain must be unique (thread-scoped) for each processor', len(unique_domain) == len(acpi.get_cpupaths()))
+    detail(unique_domain, 'domain = {:#x} (Expected a unique value for each processor)')
+    testsuite.test('_PSD.num_processors must be 1', unique_num_processors.keys() == [1])
+    detail(unique_num_processors, 'num_processors = {} (Expected 1)')
+
+def test_table_checksum(data):
+    csum = sum(ord(c) for c in data) % 0x100
+    testsuite.test('ACPI table cumulative checksum must equal 0', csum == 0)
+    testsuite.print_detail("Cumulative checksum = {} (Expected 0)".format(csum))
+
+def test_apic():
+    data = acpi.get_table("APIC")
+    if data is None:
+        return
+    test_table_checksum(data)
+    apic = acpi.parse_apic()
+
+def test_dsdt():
+    data = acpi.get_table("DSDT")
+    if data is None:
+        return
+    test_table_checksum(data)
+
+def test_facp():
+    data = acpi.get_table("FACP")
+    if data is None:
+        return
+    test_table_checksum(data)
+    facp = acpi.parse_facp()
+
+def test_hpet():
+    data = acpi.get_table("HPET")
+    if data is None:
+        return
+    test_table_checksum(data)
+    hpet = acpi.parse_hpet()
+
+def test_mpst():
+    data = acpi.get_table("MPST")
+    if data is None:
+        return
+    test_table_checksum(data)
+    mpst = acpi.MPST(data)
+
+def test_rsdp():
+    data = acpi.get_table("RSD PTR ")
+    if data is None:
+        return
+
+    # Checksum the first 20 bytes per ACPI 1.0
+    csum = sum(ord(c) for c in data[:20]) % 0x100
+    testsuite.test('ACPI 1.0 table first 20 bytes cumulative checksum must equal 0', csum == 0)
+    testsuite.print_detail("Cumulative checksum = {} (Expected 0)".format(csum))
+
+    test_table_checksum(data)
+    rsdp = acpi.parse_rsdp()
+
+def test_xsdt():
+    data = acpi.get_table("XSDT")
+    if data is None:
+        return
+    test_table_checksum(data)
+    xsdt = acpi.parse_xsdt()

tests/functional/acpi-bits/bits-tests/testcpuid.py2

@@ -0,0 +1,87 @@
+# Copyright (c) 2012, Intel Corporation
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+#     * Redistributions of source code must retain the above copyright notice,
+#       this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above copyright notice,
+#       this list of conditions and the following disclaimer in the documentation
+#       and/or other materials provided with the distribution.
+#     * Neither the name of Intel Corporation nor the names of its contributors
+#       may be used to endorse or promote products derived from this software
+#       without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+# This script runs only from the biosbits VM.
+
+"""Tests and helpers for CPUID."""
+
+import bits
+import testsuite
+import testutil
+
+def cpuid_helper(function, index=None, shift=0, mask=~0, eax_mask=~0, ebx_mask=~0, ecx_mask=~0, edx_mask=~0):
+    if index is None:
+        index = 0
+        indexdesc = ""
+    else:
+        indexdesc = " index {0:#x}".format(index)
+
+    def find_mask(m):
+        if m == ~0:
+            return mask
+        return m
+    masks = map(find_mask, [eax_mask, ebx_mask, ecx_mask, edx_mask])
+
+    uniques = {}
+    for cpu in bits.cpus():
+        regs = bits.cpuid_result(*[(r >> shift) & m for r, m in zip(bits.cpuid(cpu, function, index), masks)])
+        uniques.setdefault(regs, []).append(cpu)
+
+    desc = ["CPUID function {:#x}{}".format(function, indexdesc)]
+
+    if shift != 0:
+        desc.append("Register values have been shifted by {}".format(shift))
+    if mask != ~0 or eax_mask != ~0 or ebx_mask != ~0 or ecx_mask != ~0 or edx_mask != ~0:
+        desc.append("Register values have been masked:")
+        shifted_masks = bits.cpuid_result(*[m << shift for m in masks])
+        desc.append("Masks:           eax={eax:#010x} ebx={ebx:#010x} ecx={ecx:#010x} edx={edx:#010x}".format(**shifted_masks._asdict()))
+
+    if len(uniques) > 1:
+        regvalues = zip(*uniques.iterkeys())
+        common_masks = bits.cpuid_result(*map(testutil.find_common_mask, regvalues))
+        common_values = bits.cpuid_result(*[v[0] & m for v, m in zip(regvalues, common_masks)])
+        desc.append('Register values are not unique across all logical processors')
+        desc.append("Common bits:     eax={eax:#010x} ebx={ebx:#010x} ecx={ecx:#010x} edx={edx:#010x}".format(**common_values._asdict()))
+        desc.append("Mask of common bits: {eax:#010x}     {ebx:#010x}     {ecx:#010x}     {edx:#010x}".format(**common_masks._asdict()))
+
+    for regs in sorted(uniques.iterkeys()):
+        cpus = uniques[regs]
+        desc.append("Register value:  eax={eax:#010x} ebx={ebx:#010x} ecx={ecx:#010x} edx={edx:#010x}".format(**regs._asdict()))
+        desc.append("On {0} CPUs: {1}".format(len(cpus), testutil.apicid_list(cpus)))
+
+    return uniques, desc
+
+def test_cpuid_consistency(text, function, index=None, shift=0, mask=~0, eax_mask=~0, ebx_mask=~0, ecx_mask=~0, edx_mask=~0):
+    uniques, desc = cpuid_helper(function, index, shift, mask, eax_mask, ebx_mask, ecx_mask, edx_mask)
+    desc[0] += " Consistency Check"
+    if text:
+        desc.insert(0, text)
+    status = testsuite.test(desc[0], len(uniques) == 1)
+    for line in desc[1:]:
+        testsuite.print_detail(line)
+    return status

tests/functional/meson.build

@@ -11,6 +11,7 @@ endif
 
 # Timeouts for individual tests that can be slow e.g. with debugging enabled
 test_timeouts = {
+  'acpi_bits' : 240,
   'netdev_ethtool' : 180,
   'ppc_40p' : 240,
   'ppc64_hv' : 1000,
@@ -96,6 +97,7 @@ tests_x86_64_system_quick = [
 ]
 
 tests_x86_64_system_thorough = [
+  'acpi_bits',
   'netdev_ethtool',
   'virtio_gpu',
 ]

tests/functional/test_acpi_bits.py

@@ -0,0 +1,410 @@
+#!/usr/bin/env python3
+#
+# Exercise QEMU generated ACPI/SMBIOS tables using biosbits,
+# https://biosbits.org/
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+#
+# Author:
+#  Ani Sinha <anisinha@redhat.com>
+
+# pylint: disable=invalid-name
+# pylint: disable=consider-using-f-string
+
+"""
+This is QEMU ACPI/SMBIOS functional tests using biosbits.
+Biosbits is available originally at https://biosbits.org/.
+This test uses a fork of the upstream bits and has numerous fixes
+including an upgraded acpica. The fork is located here:
+https://gitlab.com/qemu-project/biosbits-bits .
+"""
+
+import logging
+import os
+import platform
+import re
+import shutil
+import subprocess
+import tarfile
+import tempfile
+import time
+import zipfile
+
+from pathlib import Path
+from typing import (
+    List,
+    Optional,
+    Sequence,
+)
+from qemu.machine import QEMUMachine
+from unittest import skipIf
+from qemu_test import QemuBaseTest, Asset
+
+deps = ["xorriso", "mformat"] # dependent tools needed in the test setup/box.
+supported_platforms = ['x86_64'] # supported test platforms.
+
+# default timeout of 120 secs is sometimes not enough for bits test.
+BITS_TIMEOUT = 200
+
+def which(tool):
+    """ looks up the full path for @tool, returns None if not found
+        or if @tool does not have executable permissions.
+    """
+    paths=os.getenv('PATH')
+    for p in paths.split(os.path.pathsep):
+        p = os.path.join(p, tool)
+        if os.path.exists(p) and os.access(p, os.X_OK):
+            return p
+    return None
+
+def missing_deps():
+    """ returns True if any of the test dependent tools are absent.
+    """
+    for dep in deps:
+        if which(dep) is None:
+            return True
+    return False
+
+def supported_platform():
+    """ checks if the test is running on a supported platform.
+    """
+    return platform.machine() in supported_platforms
+
+class QEMUBitsMachine(QEMUMachine): # pylint: disable=too-few-public-methods
+    """
+    A QEMU VM, with isa-debugcon enabled and bits iso passed
+    using -cdrom to QEMU commandline.
+
+    """
+    def __init__(self,
+                 binary: str,
+                 args: Sequence[str] = (),
+                 wrapper: Sequence[str] = (),
+                 name: Optional[str] = None,
+                 base_temp_dir: str = "/var/tmp",
+                 debugcon_log: str = "debugcon-log.txt",
+                 debugcon_addr: str = "0x403",
+                 qmp_timer: Optional[float] = None):
+        # pylint: disable=too-many-arguments
+
+        if name is None:
+            name = "qemu-bits-%d" % os.getpid()
+        super().__init__(binary, args, wrapper=wrapper, name=name,
+                         base_temp_dir=base_temp_dir,
+                         qmp_timer=qmp_timer)
+        self.debugcon_log = debugcon_log
+        self.debugcon_addr = debugcon_addr
+        self.base_temp_dir = base_temp_dir
+
+    @property
+    def _base_args(self) -> List[str]:
+        args = super()._base_args
+        args.extend([
+            '-chardev',
+            'file,path=%s,id=debugcon' %os.path.join(self.base_temp_dir,
+                                                     self.debugcon_log),
+            '-device',
+            'isa-debugcon,iobase=%s,chardev=debugcon' %self.debugcon_addr,
+        ])
+        return args
+
+    def base_args(self):
+        """return the base argument to QEMU binary"""
+        return self._base_args
+
+@skipIf(not supported_platform() or missing_deps(),
+        'unsupported platform or dependencies (%s) not installed' \
+        % ','.join(deps))
+class AcpiBitsTest(QemuBaseTest): #pylint: disable=too-many-instance-attributes
+    """
+    ACPI and SMBIOS tests using biosbits.
+    """
+    # in slower systems the test can take as long as 3 minutes to complete.
+    timeout = BITS_TIMEOUT
+
+    # following are some standard configuration constants
+    # gitlab CI does shallow clones of depth 20
+    BITS_INTERNAL_VER = 2020
+    # commit hash must match the artifact tag below
+    BITS_COMMIT_HASH = 'c7920d2b'
+    # this is the latest bits release as of today.
+    BITS_TAG = "qemu-bits-10262023"
+
+    ASSET_BITS = Asset(("https://gitlab.com/qemu-project/"
+                        "biosbits-bits/-/jobs/artifacts/%s/"
+                        "download?job=qemu-bits-build" % BITS_TAG),
+                       '1b8dd612c6831a6b491716a77acc486666aaa867051cdc34f7ce169c2e25f487')
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._vm = None
+        self._workDir = None
+        self._baseDir = None
+
+        self._debugcon_addr = '0x403'
+        self._debugcon_log = 'debugcon-log.txt'
+        self.logger = self.log
+
+    def _print_log(self, log):
+        self.logger.info('\nlogs from biosbits follows:')
+        self.logger.info('==========================================\n')
+        self.logger.info(log)
+        self.logger.info('==========================================\n')
+
+    def copy_bits_config(self):
+        """ copies the bios bits config file into bits.
+        """
+        config_file = 'bits-cfg.txt'
+        bits_config_dir = os.path.join(self._baseDir, 'acpi-bits',
+                                       'bits-config')
+        target_config_dir = os.path.join(self._workDir,
+                                         'bits-%d' %self.BITS_INTERNAL_VER,
+                                         'boot')
+        self.assertTrue(os.path.exists(bits_config_dir))
+        self.assertTrue(os.path.exists(target_config_dir))
+        self.assertTrue(os.access(os.path.join(bits_config_dir,
+                                               config_file), os.R_OK))
+        shutil.copy2(os.path.join(bits_config_dir, config_file),
+                     target_config_dir)
+        self.logger.info('copied config file %s to %s',
+                         config_file, target_config_dir)
+
+    def copy_test_scripts(self):
+        """copies the python test scripts into bits. """
+
+        bits_test_dir = os.path.join(self._baseDir, 'acpi-bits',
+                                     'bits-tests')
+        target_test_dir = os.path.join(self._workDir,
+                                       'bits-%d' %self.BITS_INTERNAL_VER,
+                                       'boot', 'python')
+
+        self.assertTrue(os.path.exists(bits_test_dir))
+        self.assertTrue(os.path.exists(target_test_dir))
+
+        for filename in os.listdir(bits_test_dir):
+            if os.path.isfile(os.path.join(bits_test_dir, filename)) and \
+               filename.endswith('.py2'):
+                # all test scripts are named with extension .py2 so that
+                # avocado does not try to load them. These scripts are
+                # written for python 2.7 not python 3 and hence if avocado
+                # loaded them, it would complain about python 3 specific
+                # syntaxes.
+                newfilename = os.path.splitext(filename)[0] + '.py'
+                shutil.copy2(os.path.join(bits_test_dir, filename),
+                             os.path.join(target_test_dir, newfilename))
+                self.logger.info('copied test file %s to %s',
+                                 filename, target_test_dir)
+
+                # now remove the pyc test file if it exists, otherwise the
+                # changes in the python test script won't be executed.
+                testfile_pyc = os.path.splitext(filename)[0] + '.pyc'
+                if os.access(os.path.join(target_test_dir, testfile_pyc),
+                             os.F_OK):
+                    os.remove(os.path.join(target_test_dir, testfile_pyc))
+                    self.logger.info('removed compiled file %s',
+                                     os.path.join(target_test_dir,
+                                     testfile_pyc))
+
+    def fix_mkrescue(self, mkrescue):
+        """ grub-mkrescue is a bash script with two variables, 'prefix' and
+            'libdir'. They must be pointed to the right location so that the
+            iso can be generated appropriately. We point the two variables to
+            the directory where we have extracted our pre-built bits grub
+            tarball.
+        """
+        grub_x86_64_mods = os.path.join(self._workDir, 'grub-inst-x86_64-efi')
+        grub_i386_mods = os.path.join(self._workDir, 'grub-inst')
+
+        self.assertTrue(os.path.exists(grub_x86_64_mods))
+        self.assertTrue(os.path.exists(grub_i386_mods))
+
+        new_script = ""
+        with open(mkrescue, 'r', encoding='utf-8') as filehandle:
+            orig_script = filehandle.read()
+            new_script = re.sub('(^prefix=)(.*)',
+                                r'\1"%s"' %grub_x86_64_mods,
+                                orig_script, flags=re.M)
+            new_script = re.sub('(^libdir=)(.*)', r'\1"%s/lib"' %grub_i386_mods,
+                                new_script, flags=re.M)
+
+        with open(mkrescue, 'w', encoding='utf-8') as filehandle:
+            filehandle.write(new_script)
+
+    def generate_bits_iso(self):
+        """ Uses grub-mkrescue to generate a fresh bits iso with the python
+            test scripts
+        """
+        bits_dir = os.path.join(self._workDir,
+                                'bits-%d' %self.BITS_INTERNAL_VER)
+        iso_file = os.path.join(self._workDir,
+                                'bits-%d.iso' %self.BITS_INTERNAL_VER)
+        mkrescue_script = os.path.join(self._workDir,
+                                       'grub-inst-x86_64-efi', 'bin',
+                                       'grub-mkrescue')
+
+        self.assertTrue(os.access(mkrescue_script,
+                                  os.R_OK | os.W_OK | os.X_OK))
+
+        self.fix_mkrescue(mkrescue_script)
+
+        self.logger.info('using grub-mkrescue for generating biosbits iso ...')
+
+        try:
+            if os.getenv('V') or os.getenv('BITS_DEBUG'):
+                proc = subprocess.run([mkrescue_script, '-o', iso_file,
+                                       bits_dir],
+                                      stdout=subprocess.PIPE,
+                                      stderr=subprocess.STDOUT,
+                                      check=True)
+                self.logger.info("grub-mkrescue output %s" % proc.stdout)
+            else:
+                subprocess.check_call([mkrescue_script, '-o',
+                                      iso_file, bits_dir],
+                                      stderr=subprocess.DEVNULL,
+                                      stdout=subprocess.DEVNULL)
+        except Exception as e: # pylint: disable=broad-except
+            self.skipTest("Error while generating the bits iso. "
+                          "Pass V=1 in the environment to get more details. "
+                          + str(e))
+
+        self.assertTrue(os.access(iso_file, os.R_OK))
+
+        self.logger.info('iso file %s successfully generated.', iso_file)
+
+    def setUp(self): # pylint: disable=arguments-differ
+        super().setUp('qemu-system-')
+        self.logger = self.log
+
+        self._baseDir = Path(__file__).parent
+
+        # workdir could also be avocado's own workdir in self.workdir.
+        # At present, I prefer to maintain my own temporary working
+        # directory. It gives us more control over the generated bits
+        # log files and also for debugging, we may chose not to remove
+        # this working directory so that the logs and iso can be
+        # inspected manually and archived if needed.
+        self._workDir = tempfile.mkdtemp(prefix='acpi-bits-',
+                                         suffix='.tmp')
+        self.logger.info('working dir: %s', self._workDir)
+
+        prebuiltDir = os.path.join(self._workDir, 'prebuilt')
+        if not os.path.isdir(prebuiltDir):
+            os.mkdir(prebuiltDir, mode=0o775)
+
+        bits_zip_file = os.path.join(prebuiltDir, 'bits-%d-%s.zip'
+                                     %(self.BITS_INTERNAL_VER,
+                                       self.BITS_COMMIT_HASH))
+        grub_tar_file = os.path.join(prebuiltDir,
+                                     'bits-%d-%s-grub.tar.gz'
+                                     %(self.BITS_INTERNAL_VER,
+                                       self.BITS_COMMIT_HASH))
+
+        bitsLocalArtLoc = self.ASSET_BITS.fetch()
+        self.logger.info("downloaded bits artifacts to %s", bitsLocalArtLoc)
+
+        # extract the bits artifact in the temp working directory
+        with zipfile.ZipFile(bitsLocalArtLoc, 'r') as zref:
+            zref.extractall(prebuiltDir)
+
+        # extract the bits software in the temp working directory
+        with zipfile.ZipFile(bits_zip_file, 'r') as zref:
+            zref.extractall(self._workDir)
+
+        with tarfile.open(grub_tar_file, 'r', encoding='utf-8') as tarball:
+            tarball.extractall(self._workDir)
+
+        self.copy_test_scripts()
+        self.copy_bits_config()
+        self.generate_bits_iso()
+
+    def parse_log(self):
+        """parse the log generated by running bits tests and
+           check for failures.
+        """
+        debugconf = os.path.join(self._workDir, self._debugcon_log)
+        log = ""
+        with open(debugconf, 'r', encoding='utf-8') as filehandle:
+            log = filehandle.read()
+
+        matchiter = re.finditer(r'(.*Summary: )(\d+ passed), (\d+ failed).*',
+                                log)
+        for match in matchiter:
+            # verify that no test cases failed.
+            try:
+                self.assertEqual(match.group(3).split()[0], '0',
+                                 'Some bits tests seems to have failed. ' \
+                                 'Please check the test logs for more info.')
+            except AssertionError as e:
+                self._print_log(log)
+                raise e
+            else:
+                if os.getenv('V') or os.getenv('BITS_DEBUG'):
+                    self._print_log(log)
+
+    def tearDown(self):
+        """
+           Lets do some cleanups.
+        """
+        if self._vm:
+            self.assertFalse(not self._vm.is_running)
+        if not os.getenv('BITS_DEBUG') and self._workDir:
+            self.logger.info('removing the work directory %s', self._workDir)
+            shutil.rmtree(self._workDir)
+        else:
+            self.logger.info('not removing the work directory %s ' \
+                             'as BITS_DEBUG is ' \
+                             'passed in the environment', self._workDir)
+        super().tearDown()
+
+    def test_acpi_smbios_bits(self):
+        """The main test case implementation."""
+
+        iso_file = os.path.join(self._workDir,
+                                'bits-%d.iso' %self.BITS_INTERNAL_VER)
+
+        self.assertTrue(os.access(iso_file, os.R_OK))
+
+        self._vm = QEMUBitsMachine(binary=self.qemu_bin,
+                                   base_temp_dir=self._workDir,
+                                   debugcon_log=self._debugcon_log,
+                                   debugcon_addr=self._debugcon_addr)
+
+        self._vm.add_args('-cdrom', '%s' %iso_file)
+        # the vm needs to be run under icount so that TCG emulation is
+        # consistent in terms of timing. smilatency tests have consistent
+        # timing requirements.
+        self._vm.add_args('-icount', 'auto')
+        # currently there is no support in bits for recognizing 64-bit SMBIOS
+        # entry points. QEMU defaults to 64-bit entry points since the
+        # upstream commit bf376f3020 ("hw/i386/pc: Default to use SMBIOS 3.0
+        # for newer machine models"). Therefore, enforce 32-bit entry point.
+        self._vm.add_args('-machine', 'smbios-entry-point-type=32')
+
+        # enable console logging
+        self._vm.set_console()
+        self._vm.launch()
+
+
+        # biosbits has been configured to run all the specified test suites
+        # in batch mode and then automatically initiate a vm shutdown.
+        # Set timeout to BITS_TIMEOUT for SHUTDOWN event from bits VM at par
+        # with the avocado test timeout.
+        self._vm.event_wait('SHUTDOWN', timeout=BITS_TIMEOUT)
+        self._vm.wait(timeout=None)
+        self.logger.debug("Checking console output ...")
+        self.parse_log()
+
+if __name__ == '__main__':
+    QemuBaseTest.main()