Fix FAT chain termination and harden SD driver error handling
This commit is contained in:
@@ -0,0 +1,49 @@
|
||||
# Проверки драйвера
|
||||
|
||||
Тесты исполняют собранный машинный код в Python-пакете `z80`.
|
||||
Данные диска синтетические; физические SD-карты и образы эмулятора
|
||||
не открываются и не изменяются.
|
||||
|
||||
Из `_sd` после сборки тела, обновления `body_syms.inc` и упаковки
|
||||
PackBits по инструкции в основном README драйвера:
|
||||
|
||||
```text
|
||||
sjasmplus.exe zc_sd_driver_body.asm --sym=body_syms_new.txt
|
||||
# При изменении тела обновить body_syms.inc и .pb до следующей команды.
|
||||
sjasmplus.exe zc_sd_driver.asm --sym=resident_syms_new.txt
|
||||
python tests/test_driver.py
|
||||
python tests/test_resident.py
|
||||
```
|
||||
|
||||
Нужны Python и пакет `z80`. Бинарники, символьные таблицы и PackBits
|
||||
должны относиться к одной сборке.
|
||||
|
||||
Покрытие (17 групп):
|
||||
|
||||
- все 128 смещений FAT32-записей, зеркалирование FAT и сохранение
|
||||
старших зарезервированных битов;
|
||||
- полное сравнение EOC, маскирование верхних четырёх битов,
|
||||
свободные/зарезервированные/повреждённые ссылки и ошибки чтения;
|
||||
- чтение и запись фрагментированных цепочек, размеры кластера
|
||||
1–128 секторов, несколько вызовов LOAD512/SAVE512, переходы страниц
|
||||
памяти с ненулевым начальным смещением;
|
||||
- выделение кластеров, расширение файла и каталога, отказ от продления
|
||||
повреждённой цепочки;
|
||||
- 32-битная арифметика LBA, границы тома, SDSC/SDHC-адресация;
|
||||
- MBR с разделом в каждом из четырёх слотов, BPB без MBR,
|
||||
геометрия FAT32 и базовые проверки BPB;
|
||||
- MKDIR/MKFILE, короткие и длинные имена, коллизии коротких имён,
|
||||
поиск без учёта регистра и отказ от создания дубликата;
|
||||
- FSInfo, упаковка даты/времени RTC в BCD и двоичном режиме;
|
||||
- значения регистров DMA и последовательности команд инициализации
|
||||
SDHC, SDv2 SDSC, legacy SD и MMC, включая повторную инициализацию
|
||||
и исчерпание попыток при постоянно занятой карте;
|
||||
- распаковка тела Z80-кодом, FENTRY, переименование, удаление,
|
||||
освобождение цепочки и сохранение IX в обёртке MKFILE.
|
||||
|
||||
Границы проверки: секторный SPI/DMA-обмен подменяется блоковым
|
||||
устройством в памяти. В отдельных тестах инициализации, DMA-регистров
|
||||
и RTC используются модели портов. Электрические характеристики,
|
||||
аппаратный тайминг, прерывания ZiFi и поведение при отключении питания
|
||||
этими тестами не проверяются. Тесты не доказывают транзакционность
|
||||
операций FAT или защиту от всех вариантов повреждённого тома.
|
||||
@@ -0,0 +1,362 @@
|
||||
"""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')
|
||||
@@ -20,9 +20,16 @@ class Driver:
|
||||
self.m = z80.Z80Machine()
|
||||
self.m.set_memory_block(0, BODY)
|
||||
self.disk = {}
|
||||
self.fail_reads = set()
|
||||
self.fail_writes = set()
|
||||
self.io = []
|
||||
self.ram = bytearray(256 * 16384)
|
||||
self.hooks = {SYMS[n]: n for n in
|
||||
('sdread_lba', 'sdread_lba_to', 'sdwrite_lba', 'write_fsinfo_hint')}
|
||||
('sdread_lba', 'sdread_lba_to', 'sdwrite_lba',
|
||||
'sdwrite_lba_from', 'write_fsinfo_hint')}
|
||||
self.hooks.update({SYMS[n]: n for n in
|
||||
('sdwrite_multi_start', 'sdwrite_multi_block', 'sdwrite_multi_stop')})
|
||||
self.multi_lba = None
|
||||
for address in [0xff00, *self.hooks]:
|
||||
self.m.set_breakpoint(address)
|
||||
self.put('reserved_sectors', 32)
|
||||
@@ -59,9 +66,28 @@ class Driver:
|
||||
hook = self.hooks.get(self.m.pc)
|
||||
if not hook:
|
||||
continue
|
||||
if hook != 'write_fsinfo_hint':
|
||||
failed = False
|
||||
if hook == 'sdwrite_multi_start':
|
||||
p = SYMS['lba_arg']
|
||||
self.multi_lba = int.from_bytes(self.m.memory[p:p+4], 'big')
|
||||
elif hook == 'sdwrite_multi_stop':
|
||||
self.multi_lba = None
|
||||
elif hook == 'sdwrite_multi_block':
|
||||
assert self.multi_lba is not None
|
||||
p = SYMS['sdbuf']
|
||||
self.disk[self.multi_lba] = bytearray(self.m.memory[p:p+512])
|
||||
self.io.append((hook, self.multi_lba))
|
||||
self.multi_lba += 1
|
||||
elif hook != 'write_fsinfo_hint':
|
||||
lba = int.from_bytes(self.m.memory[SYMS['lba_arg']:SYMS['lba_arg']+4], 'big')
|
||||
if hook == 'sdwrite_lba':
|
||||
self.io.append((hook, lba))
|
||||
failed = lba in (self.fail_reads if hook.startswith('sdread') else self.fail_writes)
|
||||
if failed:
|
||||
pass
|
||||
elif hook == 'sdwrite_lba_from':
|
||||
source = self.get('dma_xfer_page', 1)*16384 + (self.get('dma_xfer_off') & 16383)
|
||||
self.disk[lba] = self.ram[source:source+512]
|
||||
elif hook == 'sdwrite_lba':
|
||||
p = SYMS['sdbuf']
|
||||
self.disk[lba] = bytearray(self.m.memory[p:p+512])
|
||||
elif hook == 'sdread_lba':
|
||||
@@ -72,9 +98,11 @@ class Driver:
|
||||
else:
|
||||
dest = self.get('dma_xfer_page', 1)*16384 + (self.get('dma_xfer_off') & 16383)
|
||||
self.ram[dest:dest+512] = self.disk[lba]
|
||||
self.m.af = 0x0040 # A=0, Z=1 (successful sector I/O)
|
||||
self.m.af = 0x0100 if failed else 0x0040
|
||||
self.m.pc = int.from_bytes(self.m.memory[self.m.sp:self.m.sp+2], 'little')
|
||||
self.m.sp += 2
|
||||
if self.m.pc == 0xff00:
|
||||
return self.m.a
|
||||
raise AssertionError(f'{name}: execution did not return')
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Resident routines; build resident with --sym=resident_syms_new.txt first."""
|
||||
import re
|
||||
from test_fat_offsets import Driver, ROOT, SYMS, BODY
|
||||
|
||||
SYMS.update((name, int(value, 16)) for name, value in re.findall(
|
||||
r'^(\w+): EQU 0x([0-9A-Fa-f]+)$', (ROOT/'resident_syms_new.txt').read_text(), re.M))
|
||||
|
||||
|
||||
def driver():
|
||||
d = Driver()
|
||||
d.m.set_memory_block(0x2000, (ROOT/'zc_sd_driver.bin').read_bytes())
|
||||
return d
|
||||
|
||||
|
||||
def test_depacker():
|
||||
d = driver()
|
||||
d.m.set_memory_block(0, bytes(len(BODY)))
|
||||
assert d.call('drv_dos_swp') == 0
|
||||
assert bytes(d.m.memory[:len(BODY)]) == BODY
|
||||
assert bytes(d.m.memory[0x2000:0x2000+(ROOT/'zc_sd_driver.bin').stat().st_size]) == (ROOT/'zc_sd_driver.bin').read_bytes()
|
||||
|
||||
|
||||
def test_free_chain():
|
||||
for broken in (False, True):
|
||||
d = driver()
|
||||
chain = [319, 500, 320, 1023]
|
||||
for i, n in enumerate(chain):
|
||||
d.fat(n, chain[i+1] if i+1 < len(chain) else 0x0ffffff8)
|
||||
d.cluster(chain[0])
|
||||
if broken:
|
||||
d.fail_reads.add(49+chain[0]//128)
|
||||
assert d.call('core_free_chain') != 0
|
||||
assert not [x for x in d.io if x[0].startswith('sdwrite')]
|
||||
else:
|
||||
assert d.call('core_free_chain') == 0
|
||||
for n in chain:
|
||||
sector, offset = divmod(n*4, 512)
|
||||
assert d.disk[49+sector][offset:offset+4] == bytes(4)
|
||||
|
||||
|
||||
def test_file_abi():
|
||||
d = driver()
|
||||
d.put('rtc_cached', 1, 1)
|
||||
for s in range(49, 113):
|
||||
d.disk[s] = bytearray(512)
|
||||
d.fat(2, 0x0fffffff)
|
||||
d.disk[1017] = bytearray(512)
|
||||
d.disk[1018] = bytearray(512)
|
||||
d.put('root_cluster_lo', 2)
|
||||
d.put('next_search_valid', 1, 1)
|
||||
d.put('next_search_lo', 319)
|
||||
d.call('core_setroot')
|
||||
def arg(data):
|
||||
d.m.set_memory_block(0x9000, data+b'\0')
|
||||
d.m.hl = 0x9000
|
||||
arg(b'\0'+(1024).to_bytes(4, 'little')+b'Example long file.bin')
|
||||
d.m.ix = 0xabcd
|
||||
assert d.call('wrap_mkfile') == 0
|
||||
assert d.m.ix == 0xabcd
|
||||
arg(b'\0Example long file.bin')
|
||||
assert d.call('drv_fentry') == 1
|
||||
assert (d.m.de, d.m.hl) == (0, 1024)
|
||||
arg(b'\0Example long file.bin')
|
||||
d.m.set_memory_block(0x9200, b'Renamed long file.bin\0')
|
||||
d.m.de = 0x9200
|
||||
assert d.call('core_renam') == 1
|
||||
arg(b'\0Renamed long file.bin')
|
||||
assert d.call('drv_fentry') == 1
|
||||
arg(b'\0Example long file.bin')
|
||||
assert d.call('drv_fentry') == 0
|
||||
arg(b'\0Renamed long file.bin')
|
||||
assert d.call('core_delfl') == 1
|
||||
arg(b'\0Renamed long file.bin')
|
||||
assert d.call('drv_fentry') == 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
for test in (test_depacker, test_free_chain, test_file_abi):
|
||||
test()
|
||||
print('PASS:', test.__name__)
|
||||
Reference in New Issue
Block a user