jenkins-bot has submitted this change. (
https://gerrit.wikimedia.org/r/c/pywikibot/core/+/1320208?usp=email )
Change subject: tests: Add TimeoutExpired to utils.execute result
......................................................................
tests: Add TimeoutExpired to utils.execute result
Bug: T433834
Change-Id: Ibd628eaae6b93618ef1d607d702e3ce3a65ce4e6
---
M tests/aspects.py
M tests/pwb_tests.py
M tests/script_tests.py
M tests/utils.py
4 files changed, 91 insertions(+), 20 deletions(-)
Approvals:
jenkins-bot: Verified
Xqt: Looks good to me, approved
diff --git a/tests/aspects.py b/tests/aspects.py
index 9486323..6ff4c77 100644
--- a/tests/aspects.py
+++ b/tests/aspects.py
@@ -54,6 +54,7 @@
AssertAPIErrorContextManager,
DryRequest,
DrySite,
+ ExecuteResult,
WarningSourceSkipContextManager,
execute_pwb,
skipping,
@@ -1552,15 +1553,18 @@
if self.orig_pywikibot_dir: # pragma: no cover
os.environ['PYWIKIBOT_DIR'] = self.orig_pywikibot_dir
- def execute(self, args: list[str], **kwargs):
+ def execute(self, args: list[str], **kwargs) -> ExecuteResult:
"""Run :func:`tests.utils.execute_pwb` with default site.
.. version-changed:: 9.1
pass all arguments to :func:`tests.utils.execute_pwb`; make
this method public.
+ .. version-changed:: 11.7
+ Return timeout information.
:param args: :mod:`pwb` warapper script arguments
:param kwargs: keyword arguments of :func:`tests.utils.execute_pwb`
+ :return: Result of :func:`tests.utils.execute_pwb`.
"""
site = self.get_site()
args.append(f'-site:{site.sitename}')
diff --git a/tests/pwb_tests.py b/tests/pwb_tests.py
index b4d4710..508a8cb 100755
--- a/tests/pwb_tests.py
+++ b/tests/pwb_tests.py
@@ -43,6 +43,9 @@
direct = execute([sys.executable, '-m', package_name])
vpwb = execute_pwb([script_path])
+
+ self.assertIsNone(direct['timeout'])
+ self.assertIsNone(vpwb['timeout'])
self.maxDiff = None
self.assertEqual(direct['stdout'], vpwb['stdout'])
@@ -112,6 +115,8 @@
'in 5.0 seconds; type CTRL-C to stop.',
]
stream = execute_pwb(['hello'], data_in=chr(3), timeout=10)
+ self.assertIsNone(stream['timeout'])
+
stderr = io.StringIO(stream['stderr'])
with self.subTest(line=0):
self.assertEqual(stderr.readline().strip(), result[0])
diff --git a/tests/script_tests.py b/tests/script_tests.py
index 07e6051..28fcf00 100755
--- a/tests/script_tests.py
+++ b/tests/script_tests.py
@@ -236,13 +236,23 @@
test_overrides['pywikibot.Site'] = 'lambda *a, **k: None'
# run the script
- result = execute_pwb(cmd, data_in=data_in, timeout=timeout,
- overrides=test_overrides)
+ result = execute_pwb(
+ cmd,
+ data_in=data_in,
+ timeout=timeout,
+ overrides=test_overrides,
+ )
err_result = result['stderr']
out_result = result['stdout']
stderr_other = err_result.splitlines()
+ if timeout_error := result['timeout']:
+ unittest_print(
+ f' timeout after {timeout_error.timeout} s',
+ end=' '
+ )
+
if result['exit_code'] == -9:
unittest_print(' killed', end=' ')
diff --git a/tests/utils.py b/tests/utils.py
index 63977c2..e3a328a 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -16,7 +16,7 @@
from contextlib import contextmanager, suppress
from pathlib import Path
from subprocess import PIPE, Popen, TimeoutExpired
-from typing import Any, NoReturn
+from typing import NoReturn, TypedDict
import pywikibot
from pywikibot import config
@@ -463,15 +463,37 @@
"""Ignore password changes."""
-def execute(command: list[str], *, data_in=None, timeout=None):
+class ExecuteResult(TypedDict):
+
+ """Result returned by execute()."""
+
+ exit_code: int
+ stdout: str
+ stderr: str
+ timeout: TimeoutExpired | None
+
+
+def execute(
+ command: list[str],
+ *,
+ data_in: Sequence[str] | None = None,
+ timeout: int | float | None = None,
+) -> ExecuteResult:
"""Execute a command and capture outputs.
.. version-changed:: 8.2
*error* parameter was removed.
.. version-changed:: 9.1
parameters except *command* are keyword only.
+ .. version-changed:: 11.7
+ Return timeout information.
- :param command: executable to run and arguments to use
+ :param command: executable to run and arguments to use.
+ :param data_in: Lines to send to the process via standard input.
+ :param timeout: Maximum number of seconds to wait for the process.
+ :return: Mapping containing the process exit code, captured stdout,
+ captured stderr, and the :exc:`subprocess.TimeoutExpired`
+ exception if the process timed out.
"""
env = os.environ.copy()
@@ -493,28 +515,43 @@
# Set EDITOR to an executable that ignores all arguments and does nothing.
env['EDITOR'] = 'break' if OSWIN32 else 'true'
- p = Popen(command, env=env, stdout=PIPE, stderr=PIPE,
- stdin=PIPE if data_in is not None else None)
+ p = Popen(
+ command,
+ env=env,
+ stdout=PIPE,
+ stderr=PIPE,
+ stdin=PIPE if data_in is not None else None
+ )
if data_in is not None:
data_in = data_in.encode(config.console_encoding)
+ timeout_error: TimeoutExpired | None = None
try:
- stdout_data, stderr_data = p.communicate(input=data_in,
- timeout=timeout)
- except TimeoutExpired:
+ stdout_data, stderr_data = p.communicate(
+ input=data_in,
+ timeout=timeout
+ )
+ except TimeoutExpired as e:
+ timeout_error = e
p.kill()
stdout_data, stderr_data = p.communicate()
- return {'exit_code': p.returncode,
- 'stdout': stdout_data.decode(config.console_encoding),
- 'stderr': stderr_data.decode(config.console_encoding)}
+ return {
+ 'exit_code': p.returncode,
+ 'stdout': stdout_data.decode(config.console_encoding),
+ 'stderr': stderr_data.decode(config.console_encoding),
+ 'timeout': timeout_error,
+ }
-def execute_pwb(args: list[str], *,
- data_in: Sequence[str] | None = None,
- timeout: int | float | None = None,
- overrides: dict[str, str] | None = None) -> dict[str, Any]:
+def execute_pwb(
+ args: list[str],
+ *,
+ data_in: Sequence[str] | None = None,
+ timeout: int | float | None = None,
+ overrides: dict[str, str] | None = None,
+) -> ExecuteResult:
"""Execute the pwb.py script and capture outputs.
.. version-changed:: 8.2
@@ -524,9 +561,18 @@
.. version-changed:: 10.4
coverage is used if running github actions and a temporary file
is used for overrides.
+ .. version-changed:: 11.7
+ The return value includes timeout information.
+
+ .. seealso:: :func:`execute`
:param args: list of arguments for pwb.py
+ :param data_in: Lines to send to the process via standard input.
+ :param timeout: Maximum number of seconds to wait for the process.
:param overrides: mapping of pywikibot symbols to test replacements
+ :return: Mapping containing the process exit code, captured stdout,
+ captured stderr, and the :exc:`subprocess.TimeoutExpired`
+ exception if the process timed out.
"""
tmp_path: Path | None = None
command = [sys.executable]
@@ -547,7 +593,10 @@
if use_coverage:
# Write overrides in temporary file
with tempfile.NamedTemporaryFile(
- 'w', suffix='.py', delete=False) as f:
+ 'w',
+ suffix='.py',
+ delete=False,
+ ) as f:
f.write(override_code)
tmp_path = Path(f.name)
command.append(f.name)
@@ -560,7 +609,10 @@
try:
# Run subprocess
result = execute(
- command=command + args, data_in=data_in, timeout=timeout)
+ command=command + args,
+ data_in=data_in,
+ timeout=timeout,
+ )
finally:
# delete temporary file if created
if tmp_path and tmp_path.exists():
--
To view, visit
https://gerrit.wikimedia.org/r/c/pywikibot/core/+/1320208?usp=email
To unsubscribe, or for help writing mail filters, visit
https://gerrit.wikimedia.org/r/settings?usp=email
Gerrit-MessageType: merged
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: Ibd628eaae6b93618ef1d607d702e3ce3a65ce4e6
Gerrit-Change-Number: 1320208
Gerrit-PatchSet: 2
Gerrit-Owner: Xqt <[email protected]>
Gerrit-Reviewer: Xqt <[email protected]>
Gerrit-Reviewer: jenkins-bot
_______________________________________________
Pywikibot-commits mailing list -- [email protected]
To unsubscribe send an email to [email protected]
