Storm Dragon
71 lines
1.8 KiB
Python
71 lines
1.8 KiB
Python
from unittest.mock import Mock
|
|
|
|
from fenrirscreenreader.screenDriver.vcsaDriver import driver as VcsaDriver
|
|
|
|
|
|
class BulkReadFailure:
|
|
def __init__(self, chunks):
|
|
self.chunks = iter(chunks)
|
|
self.readline_calls = 0
|
|
|
|
def seek(self, offset):
|
|
assert offset == 0
|
|
|
|
def read(self):
|
|
raise OSError(22, "Invalid argument")
|
|
|
|
def readline(self, size):
|
|
assert size == 1
|
|
self.readline_calls += 1
|
|
if self.readline_calls > 3:
|
|
raise RuntimeError("read continued after EOF")
|
|
return next(self.chunks)
|
|
|
|
|
|
class TransientInvalidArgumentRead:
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
self.read_calls = 0
|
|
self.readline_calls = 0
|
|
|
|
def seek(self, offset):
|
|
assert offset == 0
|
|
|
|
def read(self):
|
|
self.read_calls += 1
|
|
if self.read_calls == 1:
|
|
raise OSError(22, "Invalid argument")
|
|
return self.payload
|
|
|
|
def readline(self, size):
|
|
self.readline_calls += 1
|
|
return b""
|
|
|
|
|
|
def build_vcsa_driver():
|
|
vcsa_driver = VcsaDriver.__new__(VcsaDriver)
|
|
vcsa_driver.env = {
|
|
"runtime": {
|
|
"DebugManager": Mock(),
|
|
}
|
|
}
|
|
vcsa_driver.read_retry_delay = 0
|
|
return vcsa_driver
|
|
|
|
|
|
def test_vcsa_read_fallback_stops_at_eof():
|
|
vcsa_driver = build_vcsa_driver()
|
|
screen_file = BulkReadFailure([b"a", b"b", b""])
|
|
|
|
assert vcsa_driver.read_file(screen_file) == b"ab"
|
|
assert screen_file.readline_calls == 3
|
|
|
|
|
|
def test_vcsa_read_retries_transient_invalid_argument():
|
|
vcsa_driver = build_vcsa_driver()
|
|
screen_file = TransientInvalidArgumentRead(b"\x01\x02\x00\x00x")
|
|
|
|
assert vcsa_driver.read_file(screen_file) == b"\x01\x02\x00\x00x"
|
|
assert screen_file.read_calls == 2
|
|
assert screen_file.readline_calls == 0
|