def run_io_interrupt(_, target):
    test_harness.build_program(['io_interrupt.S'])
    result = test_harness.run_program(target)
    lines = result.split('\n')
    output = None

    for line in lines:
        start = line.find('!')
        if start != -1:
            output = line[start + 1:]

    if output is None:
        raise test_harness.TestException(
            'Could not find output string:\n' + result)

    # Make sure enough interrupts were triggered
    if output.count('*') < 2:
        raise test_harness.TestException(
            'Not enough interrupts triggered:\n' + result)

    # Make sure we see at least some of the base string printed after an
    # interrupt
    if output.find('*') >= len(output) - 1:
        raise test_harness.TestException(
            'No instances of interrupt return:\n' + result)

    # Remove all asterisks (interrupts) and make sure string is intact
    stripped = output.replace('*', '')
    if stripped != '0123456789:;<=>[email protected][\\]^_`' \
            'abcdefghijklmnopqrstuvwxyz' * 10:
        raise test_harness.TestException(
            'Base string does not match:\n' + stripped)

def send_host_interrupt(_, target):
    try:
        os.remove(SEND_PIPE_NAME)
    except OSError:
        pass    # Ignore if pipe doesn't exist

    test_harness.build_program(['send_host_interrupt.S'])

    os.mknod(SEND_PIPE_NAME, stat.S_IFIFO | 0o666)

    args = [test_harness.BIN_DIR + 'emulator',
            '-o', SEND_PIPE_NAME, test_harness.HEX_FILE]
    emulator_process = subprocess.Popen(args, stdout=subprocess.PIPE,
                                        stderr=subprocess.STDOUT)

    try:
        interrupt_pipe = os.open(SEND_PIPE_NAME, os.O_RDONLY | os.O_NONBLOCK)
        test_harness.TimedProcessRunner().communicate(emulator_process, 60)

        # Interrupts should be in pipe now
        interrupts = os.read(interrupt_pipe, 5)
        if interrupts != b'\x05\x06\x07\x08\x09':
            raise test_harness.TestException(
                'Did not receive proper host interrupts')
    finally:
        os.close(interrupt_pipe)
        os.unlink(SEND_PIPE_NAME)

def random_access_mmu_stress(_, target):
    test_harness.build_program(['random_access.S'])
    test_harness.run_program(
        target=target,
        dump_file='obj/vmem.bin',
        dump_base=DUMP_BASE,
        dump_length=MEMORY_SIZE * NUM_THREADS,
        timeout=240,
        flush_l2=True)

    # Check that threads have written proper values
    with open('obj/vmem.bin', 'rb') as memfile:
        for page_num in range(int(MEMORY_SIZE / PAGE_SIZE)):
            for thread_id in range(NUM_THREADS):
                for page_offset in range(0, PAGE_SIZE, 4):
                    val = memfile.read(4)
                    if len(val) < 4:
                        raise test_harness.TestException(
                            'output file is truncated')

                    num_val, = struct.unpack('<L', val)
                    va = page_num * PAGE_SIZE + \
                        page_offset + int(DUMP_BASE / 4)
                    expected = (thread_id << 24) | va
                    if num_val != expected:
                        raise test_harness.TestException(
                            'FAIL: mismatch @{:x} : got {:x} expected {:x}'.format((page_num * 4 + thread_id) * PAGE_SIZE,
                                                                                   num_val, expected))

def sdmmc_write(_, target):
    with open(SOURCE_BLOCK_DEV, 'wb') as fsimage:
        fsimage.write(b'\xcc' * 1536)

    test_harness.build_program(['sdmmc_write.c'])
    result = test_harness.run_program(
        target=target,
        block_device=SOURCE_BLOCK_DEV)
    if 'FAIL' in result:
        raise test_harness.TestException('Test failed ' + result)

    with open(SOURCE_BLOCK_DEV, 'rb') as fsimage:
        end_contents = fsimage.read()

    # Check contents. First block is not modified
    for index in range(512):
        if end_contents[index] != 0xcc:
            raise test_harness.TestException('mismatch at {} expected 0xcc got 0x{:02x}'
                .format(index, end_contents[index]))

    # Second block has a pattern in it
    for index in range(512):
        expected = (index ^ (index >> 3)) & 0xff
        if end_contents[index + 512] != expected:
            raise test_harness.TestException('mismatch at {} expected 0x{:02x} got 0x{:02x}'
                .format(index + 512, expected, end_contents[index + 512]))

    # Third block is not modified
    for index in range(512):
        if end_contents[index + 1024] != 0xcc:
            raise test_harness.TestException('mismatch at {} expected 0xcc got 0x{:02x}'
                .format(index + 1024, end_contents[index + 1024]))

def shared_memory(_, target):
    """See coprocessor.c for an explanation of this test"""

    test_harness.build_program(['coprocessor.c'])

    # Start the emulator
    memory_file = tempfile.NamedTemporaryFile()
    args = [test_harness.BIN_DIR + 'emulator', '-s',
            memory_file.name, test_harness.HEX_FILE]
    process = subprocess.Popen(args, stdout=subprocess.PIPE,
                               stderr=subprocess.STDOUT)

    try:
        # Hack: Need to wait for the emulator to create the shared memory
        # file and initialize it. There's currently no way for the emulator
        # to signal that this has completed, so just sleep a bit and hope
        # it's done.
        time.sleep(1.0)
        memory = mmap.mmap(memory_file.fileno(), 0)
        testvalues = [random.randint(0, 0xffffffff) for __ in range(10)]
        for value in testvalues:
            computed = sharedmem_transact(memory, value)
            if computed != (value ^ 0xffffffff):
                raise test_harness.TestException('Incorrect value from coprocessor expected ' +
                                                 hex(value ^ 0xffffffff) +
                                                 ' got ' + hex(computed))
    finally:
        process.kill()

def run_compiler_test(source_file, target):
    if target == 'host':
        subprocess.check_call(['cc', source_file, '-o', 'obj/a.out'],
                              stderr=subprocess.STDOUT)
        result = subprocess.check_output('obj/a.out')
        test_harness.check_result(source_file, result.decode())
    else:
        test_harness.build_program([source_file])
        result = test_harness.run_program(target)
        test_harness.check_result(source_file, result)

def kernel_globalinit(name):
    underscore = name.rfind('_')
    if underscore == -1:
        raise test_harness.TestException(
            'Internal error: unknown environment')

    environment = name[underscore + 1:]
    test_harness.build_program(['constructor.cpp'], image_type='user')
    result = test_harness.run_kernel(environment=environment, timeout=120)
    test_harness.check_result('constructor.cpp', result)

def kernel_ucf(name):
    underscore = name.rfind('_')
    if underscore == -1:
        raise test_harness.TestException(
            'Internal error: unknown environment')

    environment = name[underscore + 1:]
    test_harness.build_program(['user_copy_fault.c'], image_type='user')
    result = test_harness.run_kernel(environment=environment, timeout=120)
    test_harness.check_result('user_copy_fault.c', result)

def sdmmc_read(name, target):
    # Create random file
    with open(SOURCE_BLOCK_DEV, 'wb') as randfile:
        randfile.write(os.urandom(FILE_SIZE))

    test_harness.build_program(['sdmmc_read.c'])
    test_harness.run_program(
        target=target,
        block_device=SOURCE_BLOCK_DEV,
        dump_file=MEMDUMP,
        dump_base=0x200000,
        dump_length=FILE_SIZE,
        flush_l2=True)

    test_harness.assert_files_equal(SOURCE_BLOCK_DEV, MEMDUMP, 'file mismatch')

def sdmmc_read(name):
    # Create random file
    with open(SOURCE_BLOCK_DEV, 'wb') as randfile:
        randfile.write(os.urandom(FILE_SIZE))

    test_harness.build_program(['sdmmc_read.c'])
    test_harness.run_program(
        environment='emulator' if name.endswith('_emulator') else 'verilator',
        block_device=SOURCE_BLOCK_DEV,
        dump_file=MEMDUMP,
        dump_base=0x200000,
        dump_length=FILE_SIZE,
        flush_l2=True)

    test_harness.assert_files_equal(SOURCE_BLOCK_DEV, MEMDUMP, 'file mismatch')

def gdb_read_write_memory(_):
    hexfile = test_harness.build_program(['count.S'], image_type='raw')
    with EmulatorProcess(hexfile), DebugConnection() as conn:
        # Read program code at address 0. This should match values
        # in count.hex
        conn.expect('m0,10', '0004800700088007000c800700108007')

        # (address, data)
        tests = [
            (0x1000, '523c66b3'),
            (0x1234, '22'),
            (0x2242, '45f280397a5a3255fa19238693ff13c729'),
            (0x100000, '55483c091aac1e8c6db4bed1'),
            (0x200000, '16e1d56029e912a04121ce41a635155f3442355533703fafcb57f8295dd6330f82f9ffc40edb589fac1523665dc2f6e80c1e2de9718d253fcbce1c8a52c9dc21'),
        ]

        # Write memory
        for addr, data in tests:
            conn.expect('M' + hex(addr)[2:] + ',' +
                        hex(int(len(data) / 2))[2:] + ':' + data, 'OK')

        # Read and verify
        for addr, data in tests:
            conn.expect('m' + hex(addr)[2:] + ',' +
                        hex(int(len(data) / 2))[2:], data)

        # Try to write a bad address (out of range)
        # Doesn't return an error, test just ensures it
        # doesn't crash
        conn.expect('M10000000,4,12345678', 'OK')

        # Try to read a bad address (out of range)
        # As above, doesn't return error (returns 0xff...),
        # but ensure it doesn't crash.
        conn.expect('m10000000,4', 'ffffffff')

def filesystem(_, target):
    '''
    Filesystem tests. This creates a filesystem image with the test file fstest.txt
    in it, the compiles the program fs.c to perform operations on it. The program
    will print 'PASS' if it is successful.
    '''

    test_harness.build_program(['fs.c'])
    subprocess.check_output(
        [test_harness.BIN_DIR + 'mkfs', test_harness.WORK_DIR + '/fsimage.bin',
         'fstest.txt'], stderr=subprocess.STDOUT)
    result = test_harness.run_program(target=target,
                                      block_device=test_harness.WORK_DIR + '/fsimage.bin')
    if 'PASS' not in result or 'FAIL' in result:
        raise test_harness.TestException(
            'test program did not indicate pass\n' + result)

def filesystem(_):
    '''
    Filesystem tests. This creates a filesystem image with the test file fstest.txt
    in it, the compiles the program fs.c to perform operations on it. The program
    will print 'PASS' if it is successful.
    '''

    test_harness.build_program(['fs.c'])
    subprocess.check_output(
        [test_harness.PROJECT_TOP + '/bin/mkfs', 'obj/fsimage.bin',
         'fstest.txt'], stderr=subprocess.STDOUT)
    result = test_harness.run_program(environment='emulator',
                                      block_device='obj/fsimage.bin')
    if 'PASS' not in result:
        raise test_harness.TestException(
            'test program did not indicate pass\n' + result)

def lldb(_):
    """This mainly validates that LLDB is reading symbols correctly."""

    hexfile = test_harness.build_program(
        ['test_program.c'], opt_level='-O0', cflags=['-g'])
    with EmulatorProcess(hexfile) as conn:
        conn.send_command('file "obj/test.elf"')
        conn.send_command('gdb-remote 8000\n')
        response = conn.send_command(
            'breakpoint set --file test_program.c --line 27')
        if 'Breakpoint 1: where = test.elf`func2 + 96 at test_program.c:27' not in response:
            raise test_harness.TestException(
                'breakpoint: did not find expected value ' + response)

        conn.send_command('c')
        conn.wait_stop()

        expected_stack = [
            ('func2', 'test_program.c', 27),
            ('func1', 'test_program.c', 35),
            ('main', 'test_program.c', 41),
            ('do_main', '', 0)
        ]

        response = conn.send_command('bt')
        crawl = parse_stack_crawl(response)
        if crawl != expected_stack:
            raise test_harness.TestException(
                'stack crawl mismatch ' + str(crawl))

        response = conn.send_command('print value')
        if '= 67' not in response:
            raise test_harness.TestException(
                'print value: Did not find expected value ' + response)

        response = conn.send_command('print result')
        if '= 128' not in response:
            raise test_harness.TestException(
                'print result: Did not find expected value ' + response)

        # Up to previous frame
        conn.send_command('frame select --relative=1')

        response = conn.send_command('print a')
        if '= 12' not in response:
            raise test_harness.TestException(
                'print a: Did not find expected value ' + response)

        response = conn.send_command('print b')
        if '= 67' not in response:
            raise test_harness.TestException(
                'print b: Did not find expected value ' + response)

        conn.send_command('step')
        conn.wait_stop()

        response = conn.send_command('print result')
        if '= 64' not in response:
            raise test_harness.TestException(
                'print b: Did not find expected value ' + response)

def run_cosimulation_test(source_file):
    hexfile = test_harness.build_program([source_file])
    p1 = subprocess.Popen(
        verilator_args + ['+bin=' + hexfile], stdout=subprocess.PIPE)
    p2 = subprocess.Popen(
        emulator_args + [hexfile], stdin=p1.stdout, stdout=subprocess.PIPE)
    output = ''
    while True:
        got = p2.stdout.read(0x1000)
        if not got:
            break

        if verbose:
            print(str(got))
        else:
            output += str(got)

    p2.wait()
    time.sleep(1)  # Give verilator a chance to clean up
    p1.kill() 	# Make sure verilator has exited
    if p2.returncode:
        raise test_harness.TestException(
            'FAIL: cosimulation mismatch\n' + output)

    test_harness.assert_files_equal(VERILATOR_MEM_DUMP, EMULATOR_MEM_DUMP,
                                    'final memory contents to not match')

def run_cosimulation_test(source_file, *unused):
    random_seed = random.randint(0, 0xffffffff)
    if test_harness.DEBUG:
        print('random seed is {}'.format(random_seed))

    verilator_args = [
        test_harness.VSIM_PATH,
        '+trace',
        '+memdumpfile=' + VERILATOR_MEM_DUMP,
        '+memdumpbase=800000',
        '+memdumplen=400000',
        '+autoflushl2',
        '+verilator+rand+reset+2',
        '+verilator+seed+{}'.format(random_seed)
    ]

    # XXX this should probably be a command line option in test_harness.py
    if 'RANDSEED' in os.environ:
        verilator_args += ['+randseed=' + os.environ['RANDSEED']]

    emulator_args = [
        test_harness.EMULATOR_PATH,
        '-m',
        'cosim',
        '-d',
        EMULATOR_MEM_DUMP + ',0x800000,0x400000'
    ]

    if test_harness.DEBUG:
        emulator_args += ['-v']

    hexfile = test_harness.build_program([source_file])
    try:
        p1 = subprocess.Popen(
            verilator_args + ['+bin=' + hexfile], stdout=subprocess.PIPE)
        p2 = subprocess.Popen(
            emulator_args + [hexfile], stdin=p1.stdout, stdout=subprocess.PIPE)
        output = ''
        while True:
            got = p2.stdout.read(0x1000)
            if not got:
                break

            if test_harness.DEBUG:
                print(got.decode())
            else:
                output += got.decode()

        p2.wait()
        time.sleep(1)  # Give verilator a chance to clean up
    finally:
        p1.kill()
        p2.kill()

    if p2.returncode:
        raise test_harness.TestException(
            'FAIL: cosimulation mismatch\n' + output)

    test_harness.assert_files_equal(VERILATOR_MEM_DUMP, EMULATOR_MEM_DUMP,
                                    'final memory contents to not match')

def gdb_read_write_register(_, target):
    hexfile = test_harness.build_program(['register_values.S'])
    with EmulatorProcess(hexfile), DebugConnection() as conn:
        # Run code to load registers
        conn.expect('C', 'S05')

        # Check values set by program (remote GDB returns in swapped byte
        # order...)
        conn.expect('g1', '7d7f3e85')
        conn.expect('g20', 'f13403ef9d08309993f7819954ae4b3f7aeaa28f538fecbd95'
                    '36f59c6d7251269525ee70d26e8d34f48912639c86ae5dba426c83aa8455e1e2dbba4b41a4f321')

        tests = [
            (0, 'd3839b18'),
            (1, '7b53cc78'),
            (30, '0904c47d'),
            (32, 'aef331bc7dbd6f1d042be4d6f1e1649855d864387eb8f0fd49c205c37790'
                'd1874078516c1a05c74f67678456679ba7e05bb5aed7303c5aeeeba6e619accf702a'),
            (36, 'cb7e3668a97ef8ea55902658b62a682406f7206f75e5438ff95b4519fed1'
                'e73e16ce5a29b4385fa2560820f0c8f42227709387dbad3a8208b57c381e268ffe38'),
            (63, '9e2d89afb0633c2f64b2eb4fdbba4663401ee673753a66d6d899e4a4101a'
                'e4920b0b16f0e716e4f7d62d83b5784740c138ac6ab94fa14256ebb468e25f20e02f')
        ]

        for reg, value in tests:
            conn.expect('G' + hex(reg)[2:] + ',' + value, 'OK')

        for reg, value in tests:
            conn.expect('g' + hex(reg)[2:], value)

        # Read invalid register index
        conn.expect('g41', '')

        # Write invalid register index
        conn.expect('G41,12345678', '')

def gdb_register_info(_, target):
    hexfile = test_harness.build_program(['count.S'], image_type='raw')
    with EmulatorProcess(hexfile), DebugConnection() as conn:
        # Scalar registers
        for idx in range(28):
            regid = str(idx + 1)
            conn.expect('qRegisterInfo' + hex(idx + 1)[2:], 'name:s' + regid +
                        ';bitsize:32;encoding:uint;format:hex;'
                        'set:General Purpose Scalar Registers;gcc:' + regid +
                        ';dwarf:' + regid + ';')

        # These registers (sp, fp, ra) are special and have additional
        # information.
        names = ['fp', 'sp', 'ra']
        for idx, name in zip(range(28, 32), names):
            regid = str(idx + 1)
            conn.expect('qRegisterInfo' + hex(idx + 1)[2:], 'name:s' + regid +
                        ';bitsize:32;encoding:uint;format:hex;'
                        'set:General Purpose Scalar Registers;gcc:' + regid +
                        ';dwarf:' + regid + ';generic:' + name + ';')

        # Vector registers
        for idx in range(32, 63):
            regid = str(idx + 1)
            conn.expect('qRegisterInfo' + hex(idx + 1)[2:], 'name:v' + str(idx - 31) +
                        ';bitsize:512;encoding:uint;format:vector-uint32;'
                        'set:General Purpose Vector Registers;gcc:' + regid +
                        ';dwarf:' + regid + ';')

        conn.expect('qRegisterInfo65', '')

def gdb_breakpoint(_, target):
    """
    Validate stopping at a breakpoint and continuing after stopping.
    This sets two breakpoints
    """

    hexfile = test_harness.build_program(['count.S'], image_type='raw')
    with EmulatorProcess(hexfile), DebugConnection() as conn:
        # Set breakpoint
        conn.expect('Z0,0000000c', 'OK')

        # Set second breakpoint at next instruction
        conn.expect('Z0,00000010', 'OK')

        # Continue
        conn.expect('C', 'S05')

        # Read last signal
        conn.expect('?', 'S05')

        # Read PC register. Should be 0x000000c, but endian swapped
        conn.expect('g40', '0c000000')

        # Read s0, which should be 3
        conn.expect('g00', '03000000')

        # Continue again.
        conn.expect('C', 'S05')

        # Ensure the instruction it stopped at is
        # executed and it breaks on the next instruction
        conn.expect('g40', '10000000')

        # Read s0, which should be 4
        conn.expect('g00', '04000000')

def run_csmith_test(_, target):
    # Find version of csmith
    result = subprocess.check_output(['csmith', '-v']).decode()
    got = VERSION_RE.search(result)
    if not got:
        raise test_harness.TestException(
            'Could not determine csmith version ' + result)

    version_str = got.group('version')
    csmith_include = '-I/usr/local/include/csmith-' + version_str

    for x in range(100):
        source_file = 'test%04d.c' % x
        print('running ' + source_file)

        # Disable packed structs because we don't support unaligned accesses.
        # Disable longlong to avoid incompatibilities between 32-bit Nyuzi
        # and 64-bit hosts.
        subprocess.check_call(['csmith', '-o', source_file, '--no-longlong',
                               '--no-packed-struct'])

        # Compile and run on host
        subprocess.check_call(
            ['cc', '-w', source_file, '-o', test_harness.WORK_DIR + '/a.out', csmith_include])
        result = subprocess.check_output(
            test_harness.WORK_DIR + '/a.out').decode()

        got = CHECKSUM_RE.search(result)
        if not got:
            raise test_harness.TestException('no checksum in host output')

        host_checksum = int(got.group('checksum'), 16)
        print('host checksum %08x' % host_checksum)

        # Compile and run under emulator
        test_harness.build_program([source_file], cflags=[csmith_include])
        result = test_harness.run_program(target)
        got = CHECKSUM_RE.search(result)
        if not got:
            raise test_harness.TestException('no checksum in host output')

        emulator_checksum = int(got.group('checksum'), 16)
        print('emulator checksum %08x' % emulator_checksum)
        if host_checksum != emulator_checksum:
            raise test_harness.TestException('checksum mismatch')

        print('PASS')