-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Co-authored-by: Mohamed Koubaa <koubaa@github.com> Co-authored-by: pyansys-ci-bot <92810346+pyansys-ci-bot@users.noreply.github.com>
- Loading branch information
1 parent
999a671
commit 97cb3b0
Showing
5 changed files
with
308 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1 @@ | ||
add background app class |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,106 @@ | ||
# Copyright (C) 2022 - 2024 ANSYS, Inc. and/or its affiliates. | ||
# SPDX-License-Identifier: MIT | ||
# | ||
# | ||
# 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. | ||
|
||
"""Class for running Mechanical on a background thread.""" | ||
|
||
import atexit | ||
import threading | ||
import time | ||
import typing | ||
|
||
import ansys.mechanical.core as mech | ||
from ansys.mechanical.core.embedding.poster import Poster | ||
import ansys.mechanical.core.embedding.utils as utils | ||
|
||
|
||
def _exit(background_app: "BackgroundApp"): | ||
"""Stop the thread serving the Background App.""" | ||
background_app.stop() | ||
atexit.unregister(_exit) | ||
|
||
|
||
class BackgroundApp: | ||
"""Background App.""" | ||
|
||
__app: mech.App = None | ||
__app_thread: threading.Thread = None | ||
__stopped: bool = False | ||
__stop_signaled: bool = False | ||
__poster: Poster = None | ||
|
||
def __init__(self, **kwargs): | ||
"""Construct an instance of BackgroundApp.""" | ||
if BackgroundApp.__app_thread == None: | ||
BackgroundApp.__app_thread = threading.Thread( | ||
target=self._start_app, kwargs=kwargs, daemon=True | ||
) | ||
BackgroundApp.__app_thread.start() | ||
|
||
while BackgroundApp.__poster is None: | ||
time.sleep(0.05) | ||
continue | ||
else: | ||
assert ( | ||
not BackgroundApp.__stopped | ||
), "Cannot initialize a BackgroundApp once it has been stopped!" | ||
|
||
def new(): | ||
BackgroundApp.__app.new() | ||
|
||
self.post(new) | ||
|
||
atexit.register(_exit, self) | ||
|
||
@property | ||
def app(self) -> mech.App: | ||
"""Get the App instance of the background thread. | ||
It is not meant to be used aside from passing to methods using `post`. | ||
""" | ||
return BackgroundApp.__app | ||
|
||
def post(self, callable: typing.Callable): | ||
"""Post callable method to the background app thread.""" | ||
assert not BackgroundApp.__stopped, "Cannot use background app after stopping it." | ||
return BackgroundApp.__poster.post(callable) | ||
|
||
def stop(self) -> None: | ||
"""Stop the background app thread.""" | ||
if BackgroundApp.__stopped: | ||
return | ||
BackgroundApp.__stop_signaled = True | ||
while True: | ||
time.sleep(0.05) | ||
if BackgroundApp.__stopped: | ||
break | ||
|
||
def _start_app(self, **kwargs) -> None: | ||
BackgroundApp.__app = mech.App(**kwargs) | ||
BackgroundApp.__poster = BackgroundApp.__app.poster | ||
while True: | ||
if BackgroundApp.__stop_signaled: | ||
break | ||
try: | ||
utils.sleep(40) | ||
except: | ||
pass | ||
BackgroundApp.__stopped = True |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,104 @@ | ||
# Copyright (C) 2022 - 2024 ANSYS, Inc. and/or its affiliates. | ||
# SPDX-License-Identifier: MIT | ||
# | ||
# | ||
# 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. | ||
|
||
"""Miscellaneous embedding tests""" | ||
import os | ||
import sys | ||
import typing | ||
|
||
import pytest | ||
|
||
|
||
def _run_background_app_test_process( | ||
rootdir: str, run_subprocess, pytestconfig, testname: str, pass_expected: bool = None | ||
) -> typing.Tuple[bytes, bytes]: | ||
"""Run the process and return stdout and stderr after it finishes.""" | ||
version = pytestconfig.getoption("ansys_version") | ||
script = os.path.join(rootdir, "tests", "scripts", "background_app_test.py") | ||
stdout, stderr = run_subprocess( | ||
[sys.executable, script, version, testname], None, pass_expected | ||
) | ||
return stdout, stderr | ||
|
||
|
||
def _assert_success(stdout: str, pass_expected: bool) -> bool: | ||
"""Check whether the process ran to completion from its stdout | ||
Duplicate of the `_assert_success` function in test_logger.py | ||
""" | ||
|
||
if pass_expected: | ||
assert "@@success@@" in stdout | ||
else: | ||
assert "@@success@@" not in stdout | ||
|
||
|
||
def _run_background_app_test( | ||
run_subprocess, rootdir: str, pytestconfig, testname: str, pass_expected: bool = True | ||
) -> str: | ||
"""Test stderr logging using a subprocess. | ||
Also ensure that the subprocess either passes or fails based on pass_expected | ||
Returns the stderr of the subprocess as a string. | ||
""" | ||
subprocess_pass_expected = pass_expected | ||
if pass_expected == True and os.name != "nt": | ||
subprocess_pass_expected = False | ||
stdout, stderr = _run_background_app_test_process( | ||
rootdir, run_subprocess, pytestconfig, testname, subprocess_pass_expected | ||
) | ||
if not subprocess_pass_expected: | ||
stdout = stdout.decode() | ||
_assert_success(stdout, pass_expected) | ||
stderr = stderr.decode() | ||
return stderr | ||
|
||
|
||
@pytest.mark.embedding_scripts | ||
def test_background_app_multiple_instances(rootdir, run_subprocess, pytestconfig): | ||
"""Multiple instances of background app can be used.""" | ||
stderr = _run_background_app_test( | ||
run_subprocess, rootdir, pytestconfig, "multiple_instances", True | ||
) | ||
assert "Project 1" in stderr | ||
assert "Project 2" in stderr | ||
assert "Foo 3" in stderr | ||
assert "Project 4" in stderr | ||
|
||
|
||
@pytest.mark.embedding_scripts | ||
def test_background_app_use_stopped(rootdir, run_subprocess, pytestconfig): | ||
"""Multiple instances of background app cannot be used after an instance is stopped.""" | ||
stderr = _run_background_app_test( | ||
run_subprocess, rootdir, pytestconfig, "test_background_app_use_stopped", False | ||
) | ||
assert "Cannot use background app after stopping it" in stderr | ||
|
||
|
||
@pytest.mark.embedding_scripts | ||
def test_background_app_initialize_stopped(rootdir, run_subprocess, pytestconfig): | ||
"""Multiple instances of background app cannot be used after an instance is stopped.""" | ||
stderr = _run_background_app_test( | ||
run_subprocess, rootdir, pytestconfig, "test_background_app_initialize_stopped", False | ||
) | ||
assert "Cannot initialize a BackgroundApp once it has been stopped!" in stderr |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,96 @@ | ||
# Copyright (C) 2022 - 2024 ANSYS, Inc. and/or its affiliates. | ||
# SPDX-License-Identifier: MIT | ||
# | ||
# | ||
# 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. | ||
|
||
"""Test cases for background app.""" | ||
|
||
import sys | ||
import time | ||
|
||
from ansys.mechanical.core.embedding.background import BackgroundApp | ||
|
||
|
||
def _print_to_stderr(*args): | ||
msg = " ".join(map(str, args)) + "\n" | ||
sys.stderr.write(msg) | ||
|
||
|
||
def multiple_instances(version): | ||
"""Use multiple instances of BackgroundApp.""" | ||
s = BackgroundApp(version=version) | ||
|
||
def func(): | ||
return s.app.DataModel.Project.Name | ||
|
||
_print_to_stderr(s.post(func), "1") | ||
time.sleep(0.03) | ||
_print_to_stderr(s.post(func), "2") | ||
|
||
def new(): | ||
s.app.new() | ||
s.app.DataModel.Project.Name = "Foo" | ||
return s.app.DataModel.Project.Name | ||
|
||
_print_to_stderr(s.post(new), "3") | ||
|
||
del s | ||
s = BackgroundApp(version=version) | ||
_print_to_stderr(s.post(func), "4") | ||
|
||
s.stop() | ||
|
||
|
||
def test_background_app_use_stopped(version): | ||
"""Stop background app then try to use it.""" | ||
s = BackgroundApp(version=version) | ||
|
||
def func(): | ||
return s.app.DataModel.Project.Name | ||
|
||
s.post(func) | ||
s.stop() | ||
s.post(func) | ||
|
||
|
||
def test_background_app_initialize_stopped(version): | ||
"""Stop background app then try to use it.""" | ||
s = BackgroundApp(version=version) | ||
|
||
def func(): | ||
return s.app.DataModel.Project.Name | ||
|
||
s.post(func) | ||
s.stop() | ||
del s | ||
s = BackgroundApp(version=version) | ||
|
||
|
||
if __name__ == "__main__": | ||
version = sys.argv[1] | ||
test_name = sys.argv[2] | ||
tests = { | ||
"multiple_instances": multiple_instances, | ||
"test_background_app_use_stopped": test_background_app_use_stopped, | ||
"test_background_app_initialize_stopped": test_background_app_initialize_stopped, | ||
} | ||
tests[test_name](int(version)) | ||
print("@@success@@") | ||
sys.exit(0) |