From 6680b8a55ac68a31d17a74e819cff4344d9a7f91 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Parodi=2C=20Eugenio=20=F0=9F=8C=B6?= Date: Thu, 1 Jan 2026 14:47:26 +0100 Subject: [PATCH] chore: add tests --- docs/source/sphinx_modules/test_01.py | 20 +++---- ... copy.py => test.generic.010.daemon.02.py} | 0 .../test.generic.013.multiprocessing.01.py | 49 ++++++++++++++++ .../test.generic.013.multiprocessing.02.py | 56 +++++++++++++++++++ tests/t.generic/test.pytest.01.collector.py | 17 ++++-- 5 files changed, 128 insertions(+), 14 deletions(-) rename tests/t.generic/{test.generic.010.daemon.02 copy.py => test.generic.010.daemon.02.py} (100%) create mode 100755 tests/t.generic/test.generic.013.multiprocessing.01.py create mode 100755 tests/t.generic/test.generic.013.multiprocessing.02.py diff --git a/docs/source/sphinx_modules/test_01.py b/docs/source/sphinx_modules/test_01.py index e8343416..4c58c472 100644 --- a/docs/source/sphinx_modules/test_01.py +++ b/docs/source/sphinx_modules/test_01.py @@ -23,16 +23,16 @@ if True or TYPE_CHECKING: from sphinx.writers.html5 import HTML5Translator -def test_process_signature(app, what, name, obj, options, sig, ret_ann): - '''Callback function to provide new signatures.''' - print(f"{app=}\n{what=}\n{name=}\n{obj=}\n{options=}\n{sig=}\n{ret_ann=}") - return '\n'.join(["a","b","c"]), "Pipppo Pippero" - -def test_process_docstring(app, what, name, obj, options, lines:list[str]): - '''Callback function to provide overloaded signatures.''' - print(f"{app=}\n{what=}\n{name=}\n{obj=}\n{options=}\n{lines=}") - for i,line in enumerate(lines): - lines[i] = line.replace("Table","PIPPO") +# def test_process_signature(app, what, name, obj, options, sig, ret_ann): +# '''Callback function to provide new signatures.''' +# print(f"{app=}\n{what=}\n{name=}\n{obj=}\n{options=}\n{sig=}\n{ret_ann=}") +# return '\n'.join(["a","b","c"]), "Pipppo Pippero" + +# def test_process_docstring(app, what, name, obj, options, lines:list[str]): +# '''Callback function to provide overloaded signatures.''' +# print(f"{app=}\n{what=}\n{name=}\n{obj=}\n{options=}\n{lines=}") +# for i,line in enumerate(lines): +# lines[i] = line.replace("Table","PIPPO") class TermTkMethod(PyMethod): option_spec: ClassVar[OptionSpec] = PyMethod.option_spec.copy() diff --git a/tests/t.generic/test.generic.010.daemon.02 copy.py b/tests/t.generic/test.generic.010.daemon.02.py similarity index 100% rename from tests/t.generic/test.generic.010.daemon.02 copy.py rename to tests/t.generic/test.generic.010.daemon.02.py diff --git a/tests/t.generic/test.generic.013.multiprocessing.01.py b/tests/t.generic/test.generic.013.multiprocessing.01.py new file mode 100755 index 00000000..139183f6 --- /dev/null +++ b/tests/t.generic/test.generic.013.multiprocessing.01.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 + +# MIT License +# +# Copyright (c) 2025 Eugenio Parodi +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. + +import time, multiprocessing + +print(1) + +def task(): + print(2) + process = multiprocessing.current_process() + print(f"Task: {process.daemon=}") + for i in range(15): + print(f"{i=}",flush=True) + time.sleep(0.1) + +print(3) + +if __name__ == '__main__': + print(4) + process = multiprocessing.Process(target=task, daemon=True) + + process.start() + process.join() + + print(f"Main Process exit...") + + + diff --git a/tests/t.generic/test.generic.013.multiprocessing.02.py b/tests/t.generic/test.generic.013.multiprocessing.02.py new file mode 100755 index 00000000..5a0801c9 --- /dev/null +++ b/tests/t.generic/test.generic.013.multiprocessing.02.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 + +# MIT License +# +# Copyright (c) 2025 Eugenio Parodi +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. + +import time, multiprocessing + +print(1) + +def task(a,b,c,d,shared_results): + print(2,a,b,c,d,shared_results) + process = multiprocessing.current_process() + print(f"Task: {process.daemon=}") + for i in range(15): + print(f"{i=}",flush=True) + time.sleep(0.1) + shared_results['pippo'] = (5,6,7,8) + return 3,4,5,6 + +print(3) + +if __name__ == '__main__': + print(4) + manager = multiprocessing.Manager() + shared_results = manager.dict() + + process = multiprocessing.Process(target=task, daemon=True, args=(1,2,3,4,shared_results)) + + process.start() + process.join() + + print(shared_results) + + print(f"Main Process exit...") + + + diff --git a/tests/t.generic/test.pytest.01.collector.py b/tests/t.generic/test.pytest.01.collector.py index 42b986fd..c90d8656 100644 --- a/tests/t.generic/test.pytest.01.collector.py +++ b/tests/t.generic/test.pytest.01.collector.py @@ -13,14 +13,23 @@ class ResultCollector: "longrepr": str(report.longrepr) if report.failed else None }) + def pytest_itemcollected(self, item): + # Called during --collect-only + self.results.append({ + "nodeid": item.nodeid, + "outcome": "collected", + "duration": 0, + "longrepr": None + }) + def run_tests_and_collect(): collector = ResultCollector() - pytest.main(["tests/"], plugins=[collector]) + pytest.main(["-p no:terminal", "tests/"], plugins=[collector]) return collector.results def run_tests_and_collect_2(): collector = ResultCollector() - pytest.main(["--collect-only", "-q", "tests/"], plugins=[collector]) + pytest.main(["--collect-only", "-p no:terminal", "tests/"], plugins=[collector]) return collector.results if __name__ == "__main__": @@ -30,10 +39,10 @@ if __name__ == "__main__": if r['outcome'] == 'failed': print(" ↳ Error:", r['longrepr']) - print() + print("##################") results = run_tests_and_collect_2() for r in results: - print(f"{r['nodeid']}: {r['outcome']} ({r['duration']:.2f}s)") + print(f"{r['nodeid']}") if r['outcome'] == 'failed': print(" ↳ Error:", r['longrepr'])