I modified echo example a little bit to make it work with Windows 10. Since I can't really use socat so I used USB-UART adapter with shorted TX and RX connected to COM12. The problem is that the dut.write() is working but dut.expect() can't read anything.
I also verified that the same setup works on Ubuntu 20.04, I can both write and receive data.
import pytest
@pytest.mark.parametrize(
'port',
['COM12'],
indirect=True,
)
def test_echo_tcp(dut):
dut.write(b'aaa')
dut.expect('aaa') # will decode automatically
[pytest]
addopts = --embedded-services serial
# log related
log_cli = True
log_cli_level = INFO
log_cli_format = %(asctime)s %(levelname)s %(message)s
log_cli_date_format = %Y-%m-%d %H:%M:%S
================================================= test session starts =================================================
platform win32 -- Python 3.10.0, pytest-7.1.2, pluggy-1.0.0
rootdir: C:\rnd\pytest-rnd\first_test, configfile: pytest.ini
plugins: embedded-0.7.3
collected 1 item
test_echo.py::test_echo_tcp[COM12]
--------------------------------------------------- live log setup ----------------------------------------------------
2022-06-24 08:52:31 INFO Logs recorded under folder: C:\Users\user1\AppData\Local\Temp\pytest-embedded\2022-06-24_06-52-31\test_echo_tcp[COM12]
FAILED [100%]
====================================================== FAILURES =======================================================
________________________________________________ test_echo_tcp[COM12] _________________________________________________
self = <pytest_embedded_serial.dut.SerialDut object at 0x00000168C441F1F0>, pattern = 'aaa', expect_all = False
args = (), kwargs = {}, patterns = ['aaa'], res = [], wrapped_buffer_bytes = ''
debug_str = 'Not found "aaa"\nBytes in current buffer: \nPlease check the full log here: C:\\Users\\user1\\AppData\\Local\\Temp\\pytest-embedded\\2022-06-24_06-52-31\\test_echo_tcp[COM12]\\dut.log'
@functools.wraps(func) # noqa
def wrapper(
self, pattern, *args, expect_all: bool = False, **kwargs
) -> Union[Union[Match, AnyStr], List[Union[Match, AnyStr]]]:
patterns = to_list(pattern)
res = []
while patterns:
try:
> index = func(self, pattern, *args, **kwargs) # noqa
C:\Users\user1\AppData\Local\Programs\Python\Python310\lib\site-packages\pytest_embedded\dut.py:67:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <pytest_embedded_serial.dut.SerialDut object at 0x00000168C441F1F0>, pattern = 'aaa', kwargs = {}
@_pexpect_func # noqa
def expect(self, pattern, **kwargs) -> Match: # noqa
"""
Expect from `pexpect_proc`. All the arguments would pass to `pexpect.expect()`.
Returns:
AnyStr: if you're matching pexpect.EOF or pexpect.TIMEOUT to get all the current buffers.
Returns:
re.Match: if matched given string.
"""
> return self.pexpect_proc.expect(pattern, **kwargs)
C:\Users\user1\AppData\Local\Programs\Python\Python310\lib\site-packages\pytest_embedded\dut.py:109:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <pytest_embedded.log.PexpectProcess object at 0x00000168C441C550>, pattern = 'aaa', timeout = -1
searchwindowsize = -1, async_ = False, kw = {}, compiled_pattern_list = [re.compile(b'aaa', re.DOTALL)]
def expect(self, pattern, timeout=-1, searchwindowsize=-1, async_=False, **kw):
'''This seeks through the stream until a pattern is matched. The
pattern is overloaded and may take several types. The pattern can be a
StringType, EOF, a compiled re, or a list of any of those types.
Strings will be compiled to re types. This returns the index into the
pattern list. If the pattern was not a list this returns index 0 on a
successful match. This may raise exceptions for EOF or TIMEOUT. To
avoid the EOF or TIMEOUT exceptions add EOF or TIMEOUT to the pattern
list. That will cause expect to match an EOF or TIMEOUT condition
instead of raising an exception.
If you pass a list of patterns and more than one matches, the first
match in the stream is chosen. If more than one pattern matches at that
point, the leftmost in the pattern list is chosen. For example::
# the input is 'foobar'
index = p.expect(['bar', 'foo', 'foobar'])
# returns 1('foo') even though 'foobar' is a "better" match
Please note, however, that buffering can affect this behavior, since
input arrives in unpredictable chunks. For example::
# the input is 'foobar'
index = p.expect(['foobar', 'foo'])
# returns 0('foobar') if all input is available at once,
# but returns 1('foo') if parts of the final 'bar' arrive late
When a match is found for the given pattern, the class instance
attribute *match* becomes an re.MatchObject result. Should an EOF
or TIMEOUT pattern match, then the match attribute will be an instance
of that exception class. The pairing before and after class
instance attributes are views of the data preceding and following
the matching pattern. On general exception, class attribute
*before* is all data received up to the exception, while *match* and
*after* attributes are value None.
When the keyword argument timeout is -1 (default), then TIMEOUT will
raise after the default value specified by the class timeout
attribute. When None, TIMEOUT will not be raised and may block
indefinitely until match.
When the keyword argument searchwindowsize is -1 (default), then the
value specified by the class maxread attribute is used.
A list entry may be EOF or TIMEOUT instead of a string. This will
catch these exceptions and return the index of the list entry instead
of raising the exception. The attribute 'after' will be set to the
exception type. The attribute 'match' will be None. This allows you to
write code like this::
index = p.expect(['good', 'bad', pexpect.EOF, pexpect.TIMEOUT])
if index == 0:
do_something()
elif index == 1:
do_something_else()
elif index == 2:
do_some_other_thing()
elif index == 3:
do_something_completely_different()
instead of code like this::
try:
index = p.expect(['good', 'bad'])
if index == 0:
do_something()
elif index == 1:
do_something_else()
except EOF:
do_some_other_thing()
except TIMEOUT:
do_something_completely_different()
These two forms are equivalent. It all depends on what you want. You
can also just expect the EOF if you are waiting for all output of a
child to finish. For example::
p = pexpect.spawn('/bin/ls')
p.expect(pexpect.EOF)
print p.before
If you are trying to optimize for speed then see expect_list().
On Python 3.4, or Python 3.3 with asyncio installed, passing
``async_=True`` will make this return an :mod:`asyncio` coroutine,
which you can yield from to get the same result that this method would
normally give directly. So, inside a coroutine, you can replace this code::
index = p.expect(patterns)
With this non-blocking form::
index = yield from p.expect(patterns, async_=True)
'''
if 'async' in kw:
async_ = kw.pop('async')
if kw:
raise TypeError("Unknown keyword arguments: {}".format(kw))
compiled_pattern_list = self.compile_pattern_list(pattern)
> return self.expect_list(compiled_pattern_list,
timeout, searchwindowsize, async_)
C:\Users\user1\AppData\Local\Programs\Python\Python310\lib\site-packages\pexpect\spawnbase.py:343:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <pytest_embedded.log.PexpectProcess object at 0x00000168C441C550>, pattern_list = [re.compile(b'aaa', re.DOTALL)]
timeout = 30, searchwindowsize = -1, async_ = False, kw = {}
exp = <pexpect.expect.Expecter object at 0x00000168C441F760>
def expect_list(self, pattern_list, timeout=-1, searchwindowsize=-1,
async_=False, **kw):
'''This takes a list of compiled regular expressions and returns the
index into the pattern_list that matched the child output. The list may
also contain EOF or TIMEOUT(which are not compiled regular
expressions). This method is similar to the expect() method except that
expect_list() does not recompile the pattern list on every call. This
may help if you are trying to optimize for speed, otherwise just use
the expect() method. This is called by expect().
Like :meth:`expect`, passing ``async_=True`` will make this return an
asyncio coroutine.
'''
if timeout == -1:
timeout = self.timeout
if 'async' in kw:
async_ = kw.pop('async')
if kw:
raise TypeError("Unknown keyword arguments: {}".format(kw))
exp = Expecter(self, searcher_re(pattern_list), searchwindowsize)
if async_:
from ._async import expect_async
return expect_async(exp, timeout)
else:
> return exp.expect_loop(timeout)
C:\Users\user1\AppData\Local\Programs\Python\Python310\lib\site-packages\pexpect\spawnbase.py:372:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <pexpect.expect.Expecter object at 0x00000168C441F760>, timeout = -0.00274658203125
def expect_loop(self, timeout=-1):
"""Blocking expect"""
spawn = self.spawn
if timeout is not None:
end_time = time.time() + timeout
try:
idx = self.existing_data()
if idx is not None:
return idx
while True:
# No match at this point
if (timeout is not None) and (timeout < 0):
return self.timeout()
# Still have time left, so read more data
incoming = spawn.read_nonblocking(spawn.maxread, timeout)
if self.spawn.delayafterread is not None:
time.sleep(self.spawn.delayafterread)
idx = self.new_data(incoming)
# Keep reading until exception or return.
if idx is not None:
return idx
if timeout is not None:
timeout = end_time - time.time()
except EOF as e:
return self.eof(e)
except TIMEOUT as e:
> return self.timeout(e)
C:\Users\user1\AppData\Local\Programs\Python\Python310\lib\site-packages\pexpect\expect.py:181:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <pexpect.expect.Expecter object at 0x00000168C441F760>
err = TIMEOUT("<pytest_embedded.log.PexpectProcess object at 0x00000168C441C550>\nsearcher: searcher_re:\n 0: re.compile(b'aaa')")
def timeout(self, err=None):
spawn = self.spawn
spawn.before = spawn._before.getvalue()
spawn.after = TIMEOUT
index = self.searcher.timeout_index
if index >= 0:
spawn.match = TIMEOUT
spawn.match_index = index
return index
else:
spawn.match = None
spawn.match_index = None
msg = str(spawn)
msg += '\nsearcher: %s' % self.searcher
if err is not None:
msg = str(err) + '\n' + msg
exc = TIMEOUT(msg)
exc.__cause__ = None # in Python 3.x we can use "raise exc from None"
> raise exc
E pexpect.exceptions.TIMEOUT: <pytest_embedded.log.PexpectProcess object at 0x00000168C441C550>
E searcher: searcher_re:
E 0: re.compile(b'aaa')
E <pytest_embedded.log.PexpectProcess object at 0x00000168C441C550>
E searcher: searcher_re:
E 0: re.compile(b'aaa')
C:\Users\user1\AppData\Local\Programs\Python\Python310\lib\site-packages\pexpect\expect.py:144: TIMEOUT
The above exception was the direct cause of the following exception:
dut = <pytest_embedded_serial.dut.SerialDut object at 0x00000168C441F1F0>
@pytest.mark.parametrize(
'port',
['COM12'],
indirect=True,
)
def test_echo_tcp(dut):
dut.write(b'aaa')
> dut.expect('aaa')
test_echo.py:10:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <pytest_embedded_serial.dut.SerialDut object at 0x00000168C441F1F0>, pattern = 'aaa', expect_all = False
args = (), kwargs = {}, patterns = ['aaa'], res = [], wrapped_buffer_bytes = ''
debug_str = 'Not found "aaa"\nBytes in current buffer: \nPlease check the full log here: C:\\Users\\user1\\AppData\\Local\\Temp\\pytest-embedded\\2022-06-24_06-52-31\\test_echo_tcp[COM12]\\dut.log'
@functools.wraps(func) # noqa
def wrapper(
self, pattern, *args, expect_all: bool = False, **kwargs
) -> Union[Union[Match, AnyStr], List[Union[Match, AnyStr]]]:
patterns = to_list(pattern)
res = []
while patterns:
try:
index = func(self, pattern, *args, **kwargs) # noqa
except (pexpect.EOF, pexpect.TIMEOUT) as e:
wrapped_buffer_bytes = textwrap.shorten(
to_str(self.pexpect_proc.buffer),
width=200,
placeholder=f'... (total {len(self.pexpect_proc.buffer)} bytes)',
)
debug_str = (
f'Not found "{str(pattern)}"\n'
f'Bytes in current buffer: {wrapped_buffer_bytes}\n'
f'Please check the full log here: {self.logfile}'
)
> raise e.__class__(debug_str) from e
E pexpect.exceptions.TIMEOUT: Not found "aaa"
E Bytes in current buffer:
E Please check the full log here: C:\Users\user1\AppData\Local\Temp\pytest-embedded\2022-06-24_06-52-31\test_echo_tcp[COM12]\dut.log
C:\Users\user1\AppData\Local\Programs\Python\Python310\lib\site-packages\pytest_embedded\dut.py:79: TIMEOUT
------------------------------------------------- Captured log setup --------------------------------------------------
INFO root:dut.py:34 Logs recorded under folder: C:\Users\user1\AppData\Local\Temp\pytest-embedded\2022-06-24_06-52-31\test_echo_tcp[COM12]
=============================================== short test summary info ===============================================
FAILED test_echo.py::test_echo_tcp[COM12] - pexpect.exceptions.TIMEOUT: Not found "aaa"
================================================= 1 failed in 33.57s ==================================================
2022-06-24 08:52:31 aaa
![dependabot[bot] avatar](https://avatars.githubusercontent.com/in/29110?v=4 "dependabot[bot]")
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent