108 lines
2.7 KiB
Python
108 lines
2.7 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
|
||
|
|
import os
|
||
|
|
import platform, random
|
||
|
|
import subprocess, string
|
||
|
|
import sys, io
|
||
|
|
from time import sleep
|
||
|
|
|
||
|
|
import defl
|
||
|
|
from defl import Assert, cl, log
|
||
|
|
from subprocess import Popen, PIPE
|
||
|
|
from defl.testing_ import Test, Tester, TestState
|
||
|
|
from random import randrange
|
||
|
|
from defl._userInput_ import UserInput, confirmContinue
|
||
|
|
|
||
|
|
tester = Tester(name=__name__)
|
||
|
|
|
||
|
|
def rand():
|
||
|
|
s = string.ascii_lowercase + ' '
|
||
|
|
return ''.join(random.sample(s, 20))
|
||
|
|
|
||
|
|
textSplit = [bytes(f'{i} {x}', encoding='utf8') for i, x in enumerate([rand() for x in range(10)])]
|
||
|
|
text = b'\n'.join(textSplit)
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def reader():
|
||
|
|
s = io.BufferedReader(io.BytesIO(text))
|
||
|
|
return s
|
||
|
|
|
||
|
|
def UserInTest(**kargs):
|
||
|
|
return UserInput(
|
||
|
|
**(dict(
|
||
|
|
message=f'[promp]',
|
||
|
|
preClearInputBuffer=False,
|
||
|
|
src=reader(),
|
||
|
|
decode=False,
|
||
|
|
) | kargs)
|
||
|
|
)
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def readerCheck():
|
||
|
|
Assert(reader().read()) == text
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def test_gather():
|
||
|
|
uin = UserInTest(singleLineMode=False)
|
||
|
|
res = uin.readGather(.1)
|
||
|
|
Assert(res) == text
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def test_gatherS():
|
||
|
|
uin = UserInTest(singleLineMode=True)
|
||
|
|
res = uin.readGather(.1)
|
||
|
|
Assert(res) == textSplit[0]
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def test_readLine():
|
||
|
|
uin = UserInTest(singleLineMode=False)
|
||
|
|
res = uin.readLine()
|
||
|
|
Assert(res) == textSplit # TODO
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def test_readLineS():
|
||
|
|
uin = UserInTest(singleLineMode=True)
|
||
|
|
res = uin.readLine()
|
||
|
|
Assert(res) == textSplit[0]
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def test_readChars():
|
||
|
|
uin = UserInTest(singleLineMode=False)
|
||
|
|
res = uin.readChars(40)
|
||
|
|
Assert(res) == text[:40]
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def test_readCharsS():
|
||
|
|
uin = UserInTest(singleLineMode=True)
|
||
|
|
res = uin.readChars(40)
|
||
|
|
Assert(res) == textSplit[0]
|
||
|
|
|
||
|
|
@tester.add()
|
||
|
|
def test_continue():
|
||
|
|
outStr = io.StringIO()
|
||
|
|
output = io.BufferedWriter(outStr)
|
||
|
|
src = io.BufferedReader(io.BytesIO(b'q\n'))
|
||
|
|
uin = confirmContinue(msg='test', src=src, _outputLog=output, preClearInputBuffer=False)
|
||
|
|
Assert(uin) == False
|
||
|
|
|
||
|
|
outStr = io.StringIO()
|
||
|
|
output = io.BufferedWriter(outStr)
|
||
|
|
src = io.BufferedReader(io.BytesIO(b'y\n'))
|
||
|
|
uin = confirmContinue(msg='test', src=src, _outputLog=output, preClearInputBuffer=False)
|
||
|
|
Assert(uin) == True
|
||
|
|
# Assert(outStr) == '' # TODO
|
||
|
|
|
||
|
|
# outStr = io.StringIO()
|
||
|
|
# output = io.BufferedWriter(outStr)
|
||
|
|
# src = io.BufferedReader(io.BytesIO(b'yy\n'))
|
||
|
|
# try:
|
||
|
|
# uin = confirmContinue(msg='test', src=src, _outputLog=output, preClearInputBuffer=False, timeout=.01)
|
||
|
|
# raise Exception()
|
||
|
|
# except defl.UserInputTimeoutError:
|
||
|
|
# ...
|
||
|
|
# res = uin.readChars(30)
|
||
|
|
# Assert(res) == textSplit[0]
|
||
|
|
|
||
|
|
log.info(tester.run())
|
||
|
|
tester.exitWithStatus()
|