363 lines
14 KiB
Python
363 lines
14 KiB
Python
"""Broader machine-code regression tests; run after assembling the body.
|
|
|
|
Run from _sd: python tests/test_driver.py
|
|
Uses synthetic FAT32 sectors only; no physical card or emulator image is changed.
|
|
"""
|
|
import random
|
|
from test_fat_offsets import Driver, SYMS, test_offsets_and_links, test_write_and_allocation, test_fragmented_load
|
|
|
|
|
|
def u32(d, prefix):
|
|
return d.get(prefix+'_lo') | d.get(prefix+'_hi') << 16
|
|
|
|
|
|
def put32(d, prefix, value):
|
|
d.put(prefix+'_lo', value & 65535)
|
|
d.put(prefix+'_hi', value >> 16)
|
|
|
|
|
|
def test_eoc_boundaries():
|
|
valid = [2, 63, 64, 127, 128, 65535, 65536,
|
|
0x0ff7ffff, 0x0ff80000, 0x0fffffef]
|
|
for top in (0, 0xa0000000, 0xf0000000):
|
|
for value in valid + list(range(0x0ffffff8, 0x10000000)) + [0, 1] + list(range(0x0ffffff0, 0x0ffffff8)):
|
|
d = Driver()
|
|
d.fat(100, value | top)
|
|
d.cluster(100)
|
|
d.call('get_next_cluster')
|
|
expected = 0 if value in valid else (1 if value >= 0x0ffffff8 else 2)
|
|
assert d.get('chain_ended', 1) == expected, hex(value | top)
|
|
assert u32(d, 'cur_cluster') == (value if expected == 0 else 100)
|
|
|
|
|
|
def test_errors_do_not_extend_chain():
|
|
for broken in (0, 1, 0x0ffffff0, 0x0ffffff7, 'io'):
|
|
d = Driver()
|
|
d.fat(100, 101 if broken == 'io' else broken)
|
|
d.cluster(100)
|
|
d.call('stream_open')
|
|
d.put('stream_sec_in_cluster', 1, 1)
|
|
if broken == 'io':
|
|
d.fail_reads.add(49)
|
|
d.call('strm_advance')
|
|
assert d.get('stream_eoc', 1) == 2
|
|
before = len(d.io)
|
|
for routine in ('stream_write_sector', 'stream_write_sector_from', 'stream_read_sector_to'):
|
|
assert d.call(routine) != 0
|
|
assert len(d.io) == before, 'I/O after broken chain'
|
|
d = Driver()
|
|
d.cluster(8192) # first cluster outside a 64-sector FAT
|
|
d.call('get_next_cluster')
|
|
assert d.get('chain_ended', 1) == 2 and not d.io
|
|
for routine in ('core_load512', 'core_save512'):
|
|
d = Driver()
|
|
d.cluster(100)
|
|
d.call('stream_open')
|
|
d.fail_reads.add(1213)
|
|
d.fail_writes.add(1213)
|
|
d.m.c, d.m.hl, d.m.b = 32, 0x4000, 1
|
|
assert d.call(routine) == 15
|
|
assert (d.m.c, d.m.hl) == (32, 0x4000)
|
|
assert d.get('stream_sec_in_cluster', 1) == 0
|
|
|
|
|
|
def test_address_arithmetic():
|
|
rng = random.Random(589)
|
|
for _ in range(300):
|
|
d = Driver()
|
|
cluster = rng.randrange(2, 0x400000)
|
|
spc = rng.choice((1, 2, 4, 8, 16, 32, 64, 128))
|
|
data, base = rng.randrange(32, 200000), rng.randrange(0, 200000)
|
|
d.cluster(cluster)
|
|
d.put('sectors_per_cluster', spc, 1)
|
|
put32(d, 'data_start', data)
|
|
put32(d, 'addtop', base)
|
|
d.call('cluster_to_sector')
|
|
assert u32(d, 'fza_sector') == base+data+(cluster-2)*spc
|
|
maximum = rng.randrange(1, 0xffffffff)
|
|
put32(d, 'sd_lba_max', maximum)
|
|
for lba in (0, maximum-1, maximum, 0xffffffff):
|
|
d.m.hl, d.m.de = lba >> 16, lba & 65535
|
|
d.call('set_lba32')
|
|
d.call('lba_guard_check')
|
|
assert bool(d.m.f & 1) == (lba < maximum)
|
|
for block, lba in ((0, rng.randrange(0x400000)), (1, rng.randrange(0xffffffff))):
|
|
d.put('sd_block_addressed', block, 1)
|
|
d.m.hl, d.m.de = lba >> 16, lba & 65535
|
|
d.call('set_lba32')
|
|
d.call('compute_card_arg')
|
|
p = SYMS['card_arg']
|
|
assert int.from_bytes(d.m.memory[p:p+4], 'big') == lba*(1 if block else 512)
|
|
|
|
|
|
def boot_sector(spc=64):
|
|
b = bytearray(512)
|
|
for offset, size, value in ((11, 2, 512), (13, 1, spc), (14, 2, 32),
|
|
(16, 1, 2), (32, 4, 10000000), (36, 4, 59430),
|
|
(44, 4, 12345), (48, 2, 1)):
|
|
b[offset:offset+size] = value.to_bytes(size, 'little')
|
|
b[510:512] = b'\x55\xaa'
|
|
return b
|
|
|
|
|
|
def test_bpb_and_partition():
|
|
for spc in (1, 2, 4, 8, 16, 32, 64, 128):
|
|
d = Driver()
|
|
d.m.set_memory_block(SYMS['sdbuf'], boot_sector(spc))
|
|
assert d.call('verify_fat32') == 0
|
|
d.call('parse_bpb')
|
|
assert u32(d, 'data_start') == 32+2*59430
|
|
assert u32(d, 'root_dir') == 32+2*59430+(12345-2)*spc
|
|
assert u32(d, 'sd_lba_max') == 17+10000000
|
|
for offset, value in ((510, 0), (511, 0), (11, 1), (22, 1), (17, 1), (13, 0), (13, 3)):
|
|
d = Driver()
|
|
b = boot_sector()
|
|
b[offset] = value
|
|
d.m.set_memory_block(SYMS['sdbuf'], b)
|
|
assert d.call('verify_fat32') != 0, offset
|
|
for slot in range(4):
|
|
d = Driver()
|
|
mbr = bytearray(512)
|
|
mbr[510:512] = b'\x55\xaa'
|
|
mbr[446+slot*16+4] = 0x0c
|
|
mbr[446+slot*16+8:446+slot*16+12] = (70000).to_bytes(4, 'little')
|
|
d.disk[70000] = boot_sector()
|
|
d.m.set_memory_block(SYMS['sdbuf'], mbr)
|
|
assert d.call('detect_partition') == 0
|
|
assert u32(d, 'addtop') == 70000
|
|
d = Driver()
|
|
original = boot_sector()
|
|
d.m.set_memory_block(SYMS['sdbuf'], original)
|
|
assert d.call('detect_partition') == 0
|
|
assert u32(d, 'addtop') == 0
|
|
|
|
|
|
def test_stream_read_write_all_cluster_sizes():
|
|
rng = random.Random(731)
|
|
for spc in (1, 2, 4, 8, 16, 32, 64, 128):
|
|
d = Driver()
|
|
d.put('sectors_per_cluster', spc, 1)
|
|
chain = [319, 500, 320, 1023]
|
|
for i, n in enumerate(chain):
|
|
d.fat(n, chain[i+1] if i+1 < len(chain) else 0x0ffffff8)
|
|
for s in range(spc):
|
|
d.disk[1017+(n-2)*spc+s] = bytearray(512)
|
|
payload = rng.randbytes(len(chain)*spc*512)
|
|
start = 32*16384+4 # nonzero, even offset; exercise page overflow remainder
|
|
d.ram[start:start+len(payload)] = payload
|
|
d.cluster(chain[0])
|
|
d.call('stream_open')
|
|
d.m.c, d.m.hl = 32, 0x4004
|
|
remaining = len(payload)//512
|
|
while remaining:
|
|
count = min(remaining, 127)
|
|
d.m.b = count
|
|
assert d.call('core_save512') == 0
|
|
remaining -= count
|
|
on_disk = b''.join(d.disk[1017+(n-2)*spc+s] for n in chain for s in range(spc))
|
|
assert on_disk == payload, spc
|
|
d.ram[start:start+len(payload)] = bytes(len(payload))
|
|
d.cluster(chain[0])
|
|
d.call('stream_open')
|
|
d.m.c, d.m.hl = 32, 0x4004
|
|
remaining = len(payload)//512
|
|
while remaining:
|
|
count = min(remaining, 113)
|
|
d.m.b = count
|
|
assert d.call('core_load512') == 0
|
|
remaining -= count
|
|
assert d.ram[start:start+len(payload)] == payload, spc
|
|
d.cluster(chain[0])
|
|
d.call('stream_open')
|
|
for s in range(spc):
|
|
assert d.call('stream_read_sector') == 0
|
|
p = SYMS['sdbuf']
|
|
assert bytes(d.m.memory[p:p+512]) == payload[s*512:(s+1)*512]
|
|
|
|
|
|
def test_grow_fragmented_file():
|
|
d = Driver()
|
|
for n in range(256, 512):
|
|
d.fat(n, 0x0fffffff)
|
|
free = [320, 356, 383, 400]
|
|
for n in free:
|
|
d.fat(n, 0)
|
|
d.put('next_search_valid', 1, 1)
|
|
d.put('next_search_lo', 320)
|
|
d.cluster(319)
|
|
d.call('stream_open')
|
|
payload = random.Random(52).randbytes(10*512)
|
|
d.ram[32*16384:32*16384+len(payload)] = payload
|
|
d.m.c, d.m.hl, d.m.b = 32, 0x4000, 10
|
|
assert d.call('core_save512') == 0
|
|
chain = [319]+free
|
|
actual = b''.join(d.disk[1017+(n-2)*2+s] for n in chain for s in range(2))
|
|
assert actual == payload
|
|
for index, n in enumerate(chain):
|
|
sector, offset = divmod(n*4, 512)
|
|
value = int.from_bytes(d.disk[49+sector][offset:offset+4], 'little') & 0x0fffffff
|
|
assert value == (chain[index+1] if index+1 < len(chain) else 0x0fffffff)
|
|
assert d.disk[49+sector] == d.disk[113+sector]
|
|
|
|
|
|
def test_directories_and_long_names():
|
|
d = Driver()
|
|
d.put('rtc_cached', 1, 1)
|
|
d.put('fat_date', 0x5d47)
|
|
d.put('fat_time', 0x6d5d)
|
|
for sector in range(49, 113):
|
|
d.disk[sector] = bytearray(512)
|
|
d.fat(2, 0x0fffffff)
|
|
d.put('root_cluster_lo', 2)
|
|
d.call('core_setroot')
|
|
d.disk[1017] = bytearray(512)
|
|
d.disk[1018] = bytearray(512)
|
|
d.put('next_search_valid', 1, 1)
|
|
d.put('next_search_lo', 320)
|
|
def arg(data):
|
|
d.m.set_memory_block(0x9000, data + b'\0')
|
|
d.m.hl = 0x9000
|
|
arg(b'Long directory name')
|
|
assert d.call('core_mkdir') == 0
|
|
arg(b'long DIRECTORY name')
|
|
assert d.call('core_fentry') == 1
|
|
directory = u32(d, 'fbn_cluster')
|
|
assert directory == 320
|
|
first = d.disk[1017+(directory-2)*2]
|
|
assert first[:11] == b'. '
|
|
assert first[32:43] == b'.. '
|
|
assert int.from_bytes(first[58:60], 'little') == 2
|
|
d.call('core_setdir')
|
|
# Multiple colliding short names, LFN runs crossing sectors, directory growth.
|
|
names = [f'Long music filename number {i:02d}.pt3'.encode() for i in range(18)]
|
|
for name in names:
|
|
arg(b'\0'+(12345).to_bytes(4, 'little')+name)
|
|
assert d.call('core_mkfile') == 0, name
|
|
clusters = set()
|
|
for name in names:
|
|
arg(name.upper())
|
|
assert d.call('core_fentry') == 1, name
|
|
assert u32(d, 'fbn_size') == 12345
|
|
clusters.add(u32(d, 'fbn_cluster'))
|
|
assert len(clusters) == len(names)
|
|
before = {n: bytes(b) for n, b in d.disk.items()}
|
|
arg(b'\0'+(12345).to_bytes(4, 'little')+names[0])
|
|
assert d.call('core_mkfile') == 3
|
|
assert before == d.disk
|
|
|
|
|
|
def test_fsinfo():
|
|
for free_count in (0xffffffff, 100000, 65536):
|
|
d = Driver()
|
|
addr = SYMS['write_fsinfo_hint']
|
|
d.m.clear_breakpoint(addr)
|
|
del d.hooks[addr]
|
|
b = bytearray(512)
|
|
for offset, value in ((0, 0x41615252), (484, 0x61417272),
|
|
(488, free_count), (492, 65535)):
|
|
b[offset:offset+4] = value.to_bytes(4, 'little')
|
|
d.disk[18] = b
|
|
d.put('fsinfo_sector', 1)
|
|
d.call('read_fsinfo_hint')
|
|
assert d.get('fsinfo_valid', 1) == 1
|
|
assert u32(d, 'fsinfo_hint') == 65535
|
|
put32(d, 'next_search', 65536)
|
|
assert d.call('write_fsinfo_hint') == 0
|
|
assert int.from_bytes(d.disk[18][492:496], 'little') == 65536
|
|
assert int.from_bytes(d.disk[18][488:492], 'little') == (free_count if free_count == 0xffffffff else free_count-1)
|
|
|
|
|
|
def test_rtc_encoding():
|
|
for binary in (False, True):
|
|
d = Driver()
|
|
rtc = {0: 59, 2: 42, 4: 23, 7: 7, 8: 10, 9: 26}
|
|
selected = [0]
|
|
def output(port, value):
|
|
if port == 0xdff7:
|
|
selected[0] = value
|
|
def input_(port):
|
|
if port != 0xbff7:
|
|
return 0
|
|
if selected[0] == 11:
|
|
return 4 if binary else 0
|
|
value = rtc.get(selected[0], 0)
|
|
return value if binary else (value//10)*16+value%10
|
|
d.m.set_output_callback(output)
|
|
d.m.set_input_callback(input_)
|
|
d.call('rtc_to_fat_datetime')
|
|
assert d.get('fat_date') == ((2026-1980)<<9 | 10<<5 | 7)
|
|
assert d.get('fat_time') == (23<<11 | 42<<5 | 59//2)
|
|
|
|
|
|
def test_dma_registers():
|
|
for routine, base, control in (('dma_recv_ext', 0x1daf, 0x42), ('dma_send_ext', 0x1aaf, 0xc2)):
|
|
for offset in (0x4000, 0x4004, 0x7ffe):
|
|
d = Driver()
|
|
ports = {}
|
|
d.m.set_output_callback(lambda port, value: ports.__setitem__(port, value))
|
|
d.m.set_input_callback(lambda port: 0)
|
|
d.put('dma_xfer_off', offset)
|
|
d.put('dma_xfer_page', 32, 1)
|
|
d.call(routine)
|
|
assert ports == {base: offset & 255, base+256: (offset>>8)&63,
|
|
base+512: 32, 0x26af: 255, 0x28af: 0, 0x27af: control}
|
|
|
|
|
|
def test_sd_initialization():
|
|
from collections import deque
|
|
d = Driver()
|
|
for kind in ('sdhc', 'sdsc', 'legacy', 'mmc', 'busy'):
|
|
commands, frame, response = [], [], deque()
|
|
selected = [False]
|
|
def output(port, value):
|
|
if port & 255 == 0x77:
|
|
selected[0] = value == 1
|
|
if not selected[0]:
|
|
frame.clear()
|
|
response.clear()
|
|
if port & 255 != 0x57 or not selected[0]:
|
|
return
|
|
if not frame and not 0x40 <= value <= 0x7f:
|
|
return
|
|
frame.append(value)
|
|
if len(frame) != 6:
|
|
return
|
|
command = frame[0] & 63
|
|
commands.append((command, int.from_bytes(bytes(frame[1:5]), 'big')))
|
|
assert len(commands) < 16010, 'Initialization retry counter never expires'
|
|
answers = {0: [1], 8: [1, 0, 0, 1, 0xaa] if kind in ('sdhc', 'sdsc', 'busy') else [5],
|
|
55: [1], 41: [1 if kind == 'busy' else (5 if kind == 'mmc' else 0)], 1: [0], 16: [0],
|
|
58: [0, 0xc0 if kind == 'sdhc' else 0x80, 0xff, 0x80, 0]}
|
|
response.extend(answers[command])
|
|
frame.clear()
|
|
def input_(port):
|
|
return response.popleft() if response else 255
|
|
d.m.set_output_callback(output)
|
|
d.m.set_input_callback(input_)
|
|
result = d.call('sdinit')
|
|
if kind == 'busy':
|
|
assert result != 0
|
|
continue
|
|
assert result == 0, kind
|
|
assert d.get('sd_block_addressed', 1) == int(kind == 'sdhc'), kind
|
|
ids = [c for c, _ in commands]
|
|
if kind in ('sdhc', 'sdsc'):
|
|
assert ids.index(41) < ids.index(58)
|
|
if kind != 'sdhc':
|
|
assert (16, 512) in commands, (kind, commands)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
tests = [value for name, value in list(globals().items()) if name.startswith('test_')]
|
|
failures = []
|
|
for test in tests:
|
|
try:
|
|
test()
|
|
print('PASS:', test.__name__, flush=True)
|
|
except Exception as exc:
|
|
import traceback
|
|
traceback.print_exc()
|
|
failures.append(test.__name__)
|
|
assert not failures, failures
|
|
print(f'{len(tests)} test groups passed')
|