Update run_process() implementation for udstool tests#467
Conversation
nemadhu
changed the title
nemadhu
changed the title
christophruethingbmw
left a comment
christophruethingbmw
left a comment
There was a problem hiding this comment.
Looks good to me
| ) | ||
| print(command) | ||
| output = helper.run_process(command) | ||
| output = run_process(command, "PositiveResponse") |
There was a problem hiding this comment.
Should we specify the timeout explicitly here? Otherwise we use the default (of I think 5s) and we are actually not aware what timeout we will have here?
| def run_process(cmd, output_contains, timeout=5): | ||
| """Run a shell command and wait until specific output appears. | ||
|
|
||
| The command is executed in ``/bin/sh`` using ``subprocess``. The function waits |
There was a problem hiding this comment.
/bin/sh vs /bin/bash below
| `` output_contains``. | ||
| """ | ||
| proc = subprocess.Popen( | ||
| ["/bin/bash", "-lc", cmd], |
There was a problem hiding this comment.
I would use a non-login shell here to isolate: ["/bin/bash", "-c", cmd],
| poller.register(proc.stdout, select.POLLIN) | ||
|
|
||
| output = [] | ||
| deadline = time.time() + timeout |
There was a problem hiding this comment.
Nit: you could use deadline = time.monotonic() + timeout here, then
try:
while True:
remaining = deadline - time.monotonic()
if remaining <= 0:
# raise timeout errro| remaining_ms = max(0, int((deadline - time.time()) * 1000)) | ||
| events = poller.poll(remaining_ms) | ||
| if not events: | ||
| break |
There was a problem hiding this comment.
This needs to be a continue, no? Otherwise we would quit in the first iteration if we haven't received any events yet
| if not events: | ||
| break | ||
| chunk = os.read(proc.stdout.fileno(), 4096).decode() | ||
| if not chunk: |
There was a problem hiding this comment.
Do we need to check if the process has finished?
if not chunk:
if proc.poll() is not None:
breakThere was a problem hiding this comment.
proc.poll() check is added after poller.poll()
| raise AssertionError(f"[TIMEOUT] Did not find {output_contains} within {timeout}s\n Output:\n{''.join(output)}") | ||
| finally: | ||
| proc.kill() | ||
| proc.wait() No newline at end of file |
There was a problem hiding this comment.
missing empty line at EOF
| typing_extensions==4.12.2 | ||
| udsoncan==1.23.1 | ||
| wrapt==1.16.0 | ||
| pexpect==4.9.0 |
There was a problem hiding this comment.
Revert the addition of pexpect
| poetry-core==2.2.1 | ||
| # via poetry | ||
| ptyprocess==0.7.0 | ||
| # via pexpect |
There was a problem hiding this comment.
restore the old lockfile to revert the addition of pexpect
3 participants
Update run_process() implementation for udstool tests
Adding run_process() helper using python subprocess
The helper function used to run udstool for test/pyTest/uds/test_udsToolRDBI.py took longer than the one second wait as per the previous implementation and the test was failing. Replacing the implementation of run_process() so as to fix this issue.