def test_fpga_secure_bit_img_auth_kup(u_boot_console):

    test_net.test_net_dhcp(u_boot_console)
    test_net.test_net_setup_static(u_boot_console)

    f = u_boot_console.config.env.get('env__fpga_secure_readable_file', None)
    if not f:
        pytest.skip('No TFTP readable file to read')

    keyaddr = f.get('keyaddr', None)
    if not keyaddr:
      addr = u_boot_utils.find_ram_base(u_boot_console)
    expected_tftp = 'Bytes transferred = '
    keyfn = f['keyfn']
    output = u_boot_console.run_command('tftpboot %x %s' % (keyaddr, keyfn))
    assert expected_tftp in output

    addr = f.get('addr', None)
    if not addr:
      addr = u_boot_utils.find_ram_base(u_boot_console)
    expected_tftp = 'Bytes transferred = '
    fn = f['enckupfn']
    output = u_boot_console.run_command('tftpboot %x %s' % (addr, fn))
    assert expected_tftp in output

    expected_zynqmpsecure = 'Bitstream successfully loaded'
    output = u_boot_console.run_command('fpga loads 0 %x $filesize 0 1 %x' % (addr, keyaddr))
    assert expected_zynqmpsecure in output

def test_zynqmp_secure_boot_img_kup(u_boot_console):

    test_net.test_net_dhcp(u_boot_console)
    test_net.test_net_setup_static(u_boot_console)

    f = u_boot_console.config.env.get('env__zynqmp_secure_readable_file', None)
    if not f:
        pytest.skip('No TFTP readable file to read')

    keyaddr = f.get('keyaddr', None)
    if not keyaddr:
      addr = u_boot_utils.find_ram_base(u_boot_console)
    expected_tftp = 'Bytes transferred = '
    keyfn = f['keyfn']
    output = u_boot_console.run_command('tftpboot %x %s' % (keyaddr, keyfn))
    assert expected_tftp in output

    addr = f.get('addr', None)
    if not addr:
      addr = u_boot_utils.find_ram_base(u_boot_console)
    expected_tftp = 'Bytes transferred = '
    fn = f['enckupfn']
    output = u_boot_console.run_command('tftpboot %x %s' % (addr, fn))
    assert expected_tftp in output

    expected_zynqmpsecure = 'Verified image at'
    output = u_boot_console.run_command('zynqmp secure %x $filesize %x' % (addr, keyaddr))
    assert expected_zynqmpsecure in output
    output = u_boot_console.run_command('pri zynqmp_verified_img_addr')
    assert "Error" not in output

def test_tpm2_pcr_extend(u_boot_console):
    """Execute a TPM2_PCR_Extend command.

    Perform a PCR extension with a known hash in memory (zeroed since the board
    must have been rebooted).

    No authentication mechanism is used here, not protecting against packet
    replay, yet.
    """

    force_init(u_boot_console)
    ram = u_boot_utils.find_ram_base(u_boot_console)

    u_boot_console.run_command('tpm2 pcr_extend 0 0x%x' % ram)
    output = u_boot_console.run_command('echo $?')
    assert output.endswith('0')

    read_pcr = u_boot_console.run_command('tpm2 pcr_read 0 0x%x' % ram)
    output = u_boot_console.run_command('echo $?')
    assert output.endswith('0')
    assert 'f5 a5 fd 42 d1 6a 20 30 27 98 ef 6e d3 09 97 9b' in read_pcr
    assert '43 00 3d 23 20 d9 f0 e8 ea 98 31 a9 27 59 fb 4b' in read_pcr

    str = re.findall(r'\d+ known updates', read_pcr)[0]
    new_updates = int(re.findall(r'\d+', str)[0])
    assert (updates + 1) == new_updates

def test_zynq_aes_image(u_boot_console):

    zynq_aes_pre_commands(u_boot_console)
    test_net.test_net_dhcp(u_boot_console)
    test_net.test_net_setup_static(u_boot_console)

    f = u_boot_console.config.env.get('env__zynq_aes_readable_file', None)
    if not f:
        pytest.skip('No TFTP readable file to read')

    srcaddr = f.get('srcaddr', None)
    if not srcaddr:
      addr = u_boot_utils.find_ram_base(u_boot_console)

    expected_tftp = 'Bytes transferred = '
    fn = f['fn']
    output = u_boot_console.run_command('tftpboot %x %s' % (srcaddr, fn))
    assert expected_tftp in output

    dstaddr = f.get('dstaddr', None)
    if not dstaddr:
       pytest.skip('No dstaddr specified in env file to read')

    dstsize = f.get('dstlen', None)
    if not dstsize:
       pytest.skip('No dstlen specified in env file to read')

    expected_zynqaes = 'zynqaes [operation type] <srcaddr>'
    output = u_boot_console.run_command('zynqaes %x $filesize %x %x' % (srcaddr, dstaddr, dstsize))
    assert expected_zynqaes not in output

def test_env_import_whitelist_delete(state_test_env):

    """Test importing only a handful of env variables from an environment, with.
       deletion if a var A that is passed to env import is not in the
       environment to be imported.
    """
    c = state_test_env.u_boot_console
    ram_base = u_boot_utils.find_ram_base(state_test_env.u_boot_console)
    addr = '%08x' % ram_base

    set_var(state_test_env, 'foo1', 'bar1')
    set_var(state_test_env, 'foo2', 'bar2')
    set_var(state_test_env, 'foo3', 'bar3')

    c.run_command('env export %s' % addr)

    unset_var(state_test_env, 'foo1')
    set_var(state_test_env, 'foo2', 'test2')
    set_var(state_test_env, 'foo4', 'bar4')

    # no foo1 in current env, foo2 overridden, foo3 should be of the value
    # before exporting and foo4 should be empty.
    c.run_command('env import -d %s - foo1 foo2 foo4' % addr)

    validate_set(state_test_env, 'foo1', 'bar1')
    validate_set(state_test_env, 'foo2', 'bar2')
    validate_set(state_test_env, 'foo3', 'bar3')
    validate_empty(state_test_env, 'foo4')

    # Cleanup test environment
    unset_var(state_test_env, 'foo1')
    unset_var(state_test_env, 'foo2')
    unset_var(state_test_env, 'foo3')
    unset_var(state_test_env, 'foo4')

def test_net_tftpboot(u_boot_console):
    """Test the tftpboot command.

    A file is downloaded from the TFTP server, its size and optionally its
    CRC32 are validated.

    The details of the file to download are provided by the boardenv_* file;
    see the comment at the beginning of this file.
    """

    if not net_set_up:
        pytest.skip('Network not initialized')

    f = u_boot_console.config.env.get('env__net_tftp_readable_file', None)
    if not f:
        pytest.skip('No TFTP readable file to read')

    addr = u_boot_utils.find_ram_base(u_boot_console)
    fn = f['fn']
    output = u_boot_console.run_command('tftpboot %x %s' % (addr, fn))
    expected_text = 'Bytes transferred = '
    sz = f.get('size', None)
    if sz:
        expected_text += '%d' % sz
    assert expected_text in output

    expected_crc = f.get('crc32', None)
    if not expected_crc:
        return

    if u_boot_console.config.buildconfig.get('config_cmd_crc32', 'n') != 'y':
        return

    output = u_boot_console.run_command('crc32 %x $filesize' % addr)
    assert expected_crc in output

def test_tpm2_dam_parameters(u_boot_console):
    """Execute a TPM2_DictionaryAttackParameters command.

    Change Dictionary Attack Mitigation (DAM) parameters. Ask the TPM to change:
    - Max number of failed authentication before lockout: 3
    - Time before the failure counter is automatically decremented: 10 sec
    - Time after a lockout failure before it can be attempted again: 0 sec

    For an unknown reason, the DAM parameters must be changed before changing
    the authentication, otherwise the lockout will be engaged after the first
    failed authentication attempt.
    """

    force_init(u_boot_console)
    ram = u_boot_utils.find_ram_base(u_boot_console)

    # Set the DAM parameters to known values
    u_boot_console.run_command('tpm2 dam_parameters 3 10 0')
    output = u_boot_console.run_command('echo $?')
    assert output.endswith('0')

    # Check the values have been saved
    read_cap = u_boot_console.run_command('tpm2 get_capability 0x6 0x20f 0x%x 3' % ram)
    output = u_boot_console.run_command('echo $?')
    assert output.endswith('0')
    assert 'Property 0x0000020f: 0x00000003' in read_cap
    assert 'Property 0x00000210: 0x0000000a' in read_cap
    assert 'Property 0x00000211: 0x00000000' in read_cap

def test_qspi_write_twice(u_boot_console):
    qspi_pre_commands(u_boot_console)

    if not qspi_detected:
        pytest.skip('QSPI not detected')

    expected_write = "Written: OK"
    expected_read = "Read: OK"

    old_size = 0
    # TODO maybe add alignment and different start for pages
    for size in random.randint(4, page_size), random.randint(page_size, total_size), total_size:
        addr = u_boot_utils.find_ram_base(u_boot_console)
        size = size - old_size
        output = u_boot_console.run_command('crc32 %x %x' % (addr + total_size, size))
        m = re.search('==> (.+?)', output)
        if not m:
            pytest.fail("CRC32 failed")

        expected_crc32 = m.group(1)
        # print expected_crc32
        output = u_boot_console.run_command('sf write %x %x %x' % (addr + total_size, old_size, size))
        assert expected_write in output
        output = u_boot_console.run_command('sf read %x %x %x' % (addr + total_size + 10, old_size, size))
        assert expected_read in output
        output = u_boot_console.run_command('crc32 %x %x' % (addr + total_size + 10, size))
        assert expected_crc32 in output
        old_size = size

def qspi_write_continues(u_boot_console):

    qspi_erase_block(u_boot_console)
    expected_write = "Written: OK"
    expected_read = "Read: OK"
    addr = u_boot_utils.find_ram_base(u_boot_console)

    output = u_boot_console.run_command('crc32 %x %x' % (addr + 0x10000, total_size))
    m = re.search('==> (.+?)$', output)
    if not m:
        pytest.fail("CRC32 failed")
    expected_crc32 = m.group(1)
    # print expected_crc32

    timeout = 10000000
    old_size = 0
    for size in random.randint(4, page_size), random.randint(page_size, total_size), total_size:
	size = size & ~3
        size = size - old_size
        with u_boot_console.temporary_timeout(timeout):
            output = u_boot_console.run_command('sf write %x %x %x' % (addr + 0x10000 + old_size, old_size, size))
            assert expected_write in output
        old_size = size

    with u_boot_console.temporary_timeout(timeout):
        output = u_boot_console.run_command('sf read %x %x %x' % (addr + 0x10000 + total_size, 0, total_size))
        assert expected_read in output

    #u_boot_console.run_command('md %x' % (addr + 0x10000 + total_size))

    output = u_boot_console.run_command('crc32 %x %x' % (addr + 0x10000 + total_size, total_size))
    assert expected_crc32 in output

def test_mmc_fatload_fatwrite(u_boot_console):
    if not mmc_set_up:
        pytest.skip('No SD/MMC/eMMC controller available')

    if not devices:
        pytest.skip('No devices detected')

    part_detect = 0
    fs = "fat"
    for x in range(0, controllers):
        if devices[x]["detected"] == "yes":
            u_boot_console.run_command('mmc dev %d' % x)
            try:
                partitions = devices[x][fs]
            except:
                print ("No %s table on this device" % fs.upper());
                continue

            for part in partitions:
                part_detect = 1
                addr = u_boot_utils.find_ram_base(u_boot_console)
                devices[x]["addr_%d" % part] = addr
                size = random.randint(4, 1 * 1024 * 1024)
                devices[x]["size_%d" % part] = size
                # count CRC32
                output = u_boot_console.run_command('crc32 %x %x' % (addr, size))
                m = re.search('==> (.+?)', output)
                if not m:
                    pytest.fail("CRC32 failed")
                expected_crc32 = m.group(1)
                devices[x]["expected_crc32_%d" % part] = expected_crc32
                # do write

                file = '%s_%d' % ("uboot_test", size)
                devices[x]["file_%d" % part] = file
                output = u_boot_console.run_command("%swrite mmc %d:%s %x %s %x" % (fs, x, part, addr, file, size))
                assert "Unable to write" not in output
                assert "Error" not in output
                assert "overflow" not in output
                expected_text = "%d bytes written" % size
                assert expected_text in output

                alignment = int(u_boot_console.config.buildconfig.get('config_sys_cacheline_size', 128))
                offset = random.randrange(alignment, 1024, alignment)
                output = u_boot_console.run_command("%sload mmc %d:%s %x %s" % (fs, x, part, addr + offset, file))
                assert "Invalid FAT entry" not in output
                assert "Unable to read file" not in output
                assert "Misaligned buffer address" not in output
                expected_text = "%d bytes read" % size
                assert expected_text in output

                output = u_boot_console.run_command('crc32 %x $filesize' % (addr + offset))
                assert expected_crc32 in output

    if not part_detect:
        pytest.skip("No %s partition detected" % fs.upper())

def sf_prepare(u_boot_console, env__sf_config):
    """Check global state of the SPI Flash before running any test.

   Args:
        u_boot_console: A U-Boot console connection.
        env__sf_config: The single SPI Flash device configuration on which to
            run the tests.

    Returns:
        sf_params: a dictionary of SPI Flash parameters.
    """

    sf_params = {}
    sf_params['ram_base'] = u_boot_utils.find_ram_base(u_boot_console)

    probe_id = env__sf_config.get('id', 0)
    speed = env__sf_config.get('speed', 0)
    if isinstance(speed, int):
        sf_params['speed'] = speed
    else:
        assert len(speed) == 2, "If speed is a list, it must have 2 entries"
        sf_params['speed'] = random.randint(speed[0], speed[1])

    cmd = 'sf probe %d %d' % (probe_id, sf_params['speed'])

    output = u_boot_console.run_command(cmd)
    assert 'SF: Detected' in output, 'No Flash device available'

    m = re.search('page size (.+?) Bytes', output)
    assert m, 'SPI Flash page size not recognized'
    sf_params['page_size'] = int(m.group(1))

    m = re.search('erase size (.+?) KiB', output)
    assert m, 'SPI Flash erase size not recognized'
    sf_params['erase_size'] = int(m.group(1))
    sf_params['erase_size'] *= 1024

    m = re.search('total (.+?) MiB', output)
    assert m, 'SPI Flash total size not recognized'
    sf_params['total_size'] = int(m.group(1))
    sf_params['total_size'] *= 1024 * 1024

    assert 'offset' in env__sf_config, \
        '\'offset\' is required for this test.'
    sf_params['len'] = env__sf_config.get('len', sf_params['erase_size'])

    assert not env__sf_config['offset'] % sf_params['erase_size'], \
        'offset not multiple of erase size.'
    assert not sf_params['len'] % sf_params['erase_size'], \
        'erase length not multiple of erase size.'

    assert not (env__sf_config.get('writeable', False) and
                'crc32' in env__sf_config), \
        'Cannot check crc32 on writeable sections'

    return sf_params

def test_mmc_ext4load_ext4write(u_boot_console):
    if not mmc_set_up:
        pytest.skip('No SD/MMC/eMMC controller available')

    if not devices:
        pytest.skip('No devices detected')

    part_detect = 0
    fs = "ext4"
    for x in range(0, controllers):
        if devices[x]["detected"] == "yes":
            u_boot_console.run_command('mmc dev %d' % x)
            try:
                partitions = devices[x][fs]
            except:
                print ("No %s table on this device" % fs.upper());
                continue

            for part in partitions:
                part_detect = 1
                addr = u_boot_utils.find_ram_base(u_boot_console)
                devices[x]["addr_%d" % part] = addr
                size = random.randint(4, 1 * 1024 * 1024)
                devices[x]["size_%d" % part] = size
                # count CRC32
                output = u_boot_console.run_command('crc32 %x %x' % (addr, size))
                m = re.search('==> (.+?)', output)
                if not m:
                    pytest.fail("CRC32 failed")
                expected_crc32 = m.group(1)
                devices[x]["expected_crc32_%d" % part] = expected_crc32
                # do write

                file = '%s_%d' % ("uboot_test", size)
                devices[x]["file_%d" % part] = file
                output = u_boot_console.run_command("%swrite mmc %d:%s %x /%s %x" % (fs, x, part, addr, file, size))
                assert "Unable to write" not in output
                assert "Error" not in output
                # maybe not valid for ext4
                assert "overflow" not in output
                expected_text = "%d bytes written" % size
                assert expected_text in output

                # Read it back with 64 aligned offset
                # Todo read ALIGN
                offset = random.randrange(128, 1024, 128)
                output = u_boot_console.run_command("%sload mmc %d:%s %x /%s" % (fs, x, part, addr + offset, file))
                expected_text = "%d bytes read" % size
                assert expected_text in output

                output = u_boot_console.run_command('crc32 %x $filesize' % (addr + offset))
                assert expected_crc32 in output

    if not part_detect:
        pytest.skip("No %s partition detected" % fs.upper())

def test_env_import_whitelist_checksum_no_size(state_test_env):
    """Test that omitted ('-') size parameter with checksum validation fails the
       env import function when variables are passed as parameters.
    """
    c = state_test_env.u_boot_console
    ram_base = u_boot_utils.find_ram_base(state_test_env.u_boot_console)
    addr = '%08x' % ram_base

    with c.disable_check('error_notification'):
        response = c.run_command('env import -c %s - foo1 foo2 foo4' % addr)
    assert(response == '## Error: external checksum format must pass size')

def test_md_repeat(u_boot_console):
    """Test command repeat (via executing an empty command) operates correctly
    for "md"; the command must repeat and dump an incrementing address."""

    ram_base = u_boot_utils.find_ram_base(u_boot_console)
    addr_base = '%08x' % ram_base
    words = 0x10
    addr_repeat = '%08x' % (ram_base + (words * 4))
    u_boot_console.run_command('md %s %x' % (addr_base, words))
    response = u_boot_console.run_command('')
    expected_response = addr_repeat + ': '
    assert(expected_response in response)

def test_md(u_boot_console):
    """Test that md reads memory as expected, and that memory can be modified
    using the mw command."""

    ram_base = u_boot_utils.find_ram_base(u_boot_console)
    addr = '%08x' % ram_base
    val = 'a5f09876'
    expected_response = addr + ': ' + val
    u_boot_console.run_command('mw ' + addr + ' 0 10')
    response = u_boot_console.run_command('md ' + addr + ' 10')
    assert(not (expected_response in response))
    u_boot_console.run_command('mw ' + addr + ' ' + val)
    response = u_boot_console.run_command('md ' + addr + ' 10')
    assert(expected_response in response)

def qspi_write_twice(u_boot_console):

    addr = u_boot_utils.find_ram_base(u_boot_console)
    expected_write = "Written: OK"
    expected_read = "Read: OK"
    expected_erase = "Erased: OK"

    old_size = 0
    # TODO maybe add alignment and different start for pages
    for size in random.randint(4, page_size), random.randint(page_size, total_size), total_size:
        offset = random.randint(4, page_size)
        offset = offset & ~3
        size = size & ~3
        size = size - old_size
        output = u_boot_console.run_command('crc32 %x %x' % (addr + total_size, size))
        m = re.search('==> (.+?)$', output)
        if not m:
            pytest.fail("CRC32 failed")

        expected_crc32 = m.group(1)
        # print expected_crc32
        if old_size % page_size:
            old_size /= page_size
            old_size *= page_size

        if size % erase_size:
            erasesize = size/erase_size + 1
            erasesize *= erase_size

        eraseoffset = old_size/erase_size
        eraseoffset *= erase_size

        timeout = 100000000
        start = 0
        with u_boot_console.temporary_timeout(timeout):
            output = u_boot_console.run_command('sf erase %x %x' % (eraseoffset, erasesize))
            assert expected_erase in output

        with u_boot_console.temporary_timeout(timeout):
            output = u_boot_console.run_command('sf write %x %x %x' % (addr + total_size, old_size, size))
            assert expected_write in output
        with u_boot_console.temporary_timeout(timeout):
            output = u_boot_console.run_command('sf read %x %x %x' % (addr + total_size + offset, old_size, size))
            assert expected_read in output
        output = u_boot_console.run_command('crc32 %x %x' % (addr + total_size + offset, size))
        assert expected_crc32 in output
        old_size = size

def test_qspi_boot_images(u_boot_console):
    qspi_pre_commands(u_boot_console)
    if not qspi_detected:
        pytest.skip('QSPI not detected')

    if not test_net.net_set_up:
        pytest.skip('Network not initialized')

    test_net.test_net_dhcp(u_boot_console)
    test_net.test_net_setup_static(u_boot_console)
    test_net.test_net_tftpboot(u_boot_console)

    f = u_boot_console.config.env.get('env__net_tftp_readable_file', None)
    if not f:
        pytest.skip('No TFTP readable file to read')

    addr = f.get('addr', None)
    if not addr:
      addr = u_boot_utils.find_ram_base(u_boot_console)

    map = 0x0
    temp = 0x50000
    expected_write = "OK"
    output = u_boot_console.run_command('imxtract %x [email protected] %x' % (addr, temp))
    assert expected_write in output

    expected_erase = "Erased: OK"
    output = u_boot_console.run_command('sf erase %x +$filesize ' % map)
    assert expected_erase in output

    expected_write = "Written: OK"
    output = u_boot_console.run_command('sf write %x %x $filesize' % (temp, map))
    assert expected_write in output

    map = u_boot_console.config.buildconfig.get('config_sys_spi_u_boot_offs', "0x1000")
    map = int(map, 16)
    expected_write = "OK"
    output = u_boot_console.run_command('imxtract %x [email protected] %x' % (addr, temp))
    assert expected_write in output

    expected_erase = "Erased: OK"
    output = u_boot_console.run_command('sf erase %x +$filesize ' % map)
    assert expected_erase in output

    expected_write = "Written: OK"
    output = u_boot_console.run_command('sf write %x %x $filesize' % (temp, map))
    assert expected_write in output

def test_tpm2_get_capability(u_boot_console):
    """Execute a TPM_GetCapability command.

    Display one capability. In our test case, let's display the default DAM
    lockout counter that should be 0 since the CLEAR:
    - TPM_CAP_TPM_PROPERTIES = 0x6
    - TPM_PT_LOCKOUT_COUNTER (1st parameter) = PTR_VAR + 14

    There is no expected default values because it would depend on the chip
    used. We can still save them in order to check they have changed later.
    """

    force_init(u_boot_console)
    ram = u_boot_utils.find_ram_base(u_boot_console)

    read_cap = u_boot_console.run_command('tpm2 get_capability 0x6 0x20e 0x200 1') #0x%x 1' % ram)
    output = u_boot_console.run_command('echo $?')
    assert output.endswith('0')
    assert 'Property 0x0000020e: 0x00000000' in read_cap

def qspi_read_twice(u_boot_console):

    expected_read = "Read: OK"

    # TODO maybe add alignment and different start for pages
    for size in random.randint(4, page_size), random.randint(4, total_size), total_size:
        addr = u_boot_utils.find_ram_base(u_boot_console)
        # FIXME using 0 is failing for me
        output = u_boot_console.run_command('sf read %x 0 %x' % (addr + total_size, size))
        assert expected_read in output
        output = u_boot_console.run_command('crc32 %x %x' % (addr + total_size, size))
        m = re.search('==> (.+?)', output)
        if not m:
            pytest.fail("CRC32 failed")
        expected_crc32 = m.group(1)
        output = u_boot_console.run_command('sf read %x 0 %x' % (addr + total_size + 10, size))
        assert expected_read in output
        output = u_boot_console.run_command('crc32 %x %x' % (addr + total_size + 10, size))
        assert expected_crc32 in output

def test_tpm2_pcr_read(u_boot_console):
    """Execute a TPM2_PCR_Read command.

    Perform a PCR read of the 0th PCR. Must be zero.
    """

    force_init(u_boot_console)
    ram = u_boot_utils.find_ram_base(u_boot_console)

    read_pcr = u_boot_console.run_command('tpm2 pcr_read 0 0x%x' % ram)
    output = u_boot_console.run_command('echo $?')
    assert output.endswith('0')

    # Save the number of PCR updates
    str = re.findall(r'\d+ known updates', read_pcr)[0]
    global updates
    updates = int(re.findall(r'\d+', str)[0])

    # Check the output value
    assert 'PCR #0 content' in read_pcr
    assert '00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00' in read_pcr