diff --git a/src/ansys/fluent/core/services/datamodel_se.py b/src/ansys/fluent/core/services/datamodel_se.py index 6b1520ed80a..1c6a5404762 100644 --- a/src/ansys/fluent/core/services/datamodel_se.py +++ b/src/ansys/fluent/core/services/datamodel_se.py @@ -1577,7 +1577,7 @@ def _del_item(self, key: str) -> None: # On-deleted subscription objects are unsubscribed after the datamodel # object is deleted. self[key].add_on_deleted( - lambda _: self.service.subscriptions.unsubscribe_while_deleting( + lambda: self.service.subscriptions.unsubscribe_while_deleting( self.rules, se_path, "after" ) ) diff --git a/src/ansys/fluent/core/streaming_services/datamodel_event_streaming.py b/src/ansys/fluent/core/streaming_services/datamodel_event_streaming.py index 5c0ff5e1363..1627d8870d1 100644 --- a/src/ansys/fluent/core/streaming_services/datamodel_event_streaming.py +++ b/src/ansys/fluent/core/streaming_services/datamodel_event_streaming.py @@ -66,12 +66,12 @@ def _process_streaming(self, id, stream_begin_method, started_evt, *args, **kwar elif response.HasField("commandAttributeChangedEventResponse"): value = response.commandAttributeChangedEventResponse.value cb[1](_convert_variant_to_value(value)) - elif ( - response.HasField("modifiedEventResponse") - or response.HasField("deletedEventResponse") - or response.HasField("affectedEventResponse") - ): + elif response.HasField( + "modifiedEventResponse" + ) or response.HasField("affectedEventResponse"): cb[1](cb[0]) + elif response.HasField("deletedEventResponse"): + cb[1]() elif response.HasField("commandExecutedEventResponse"): command = response.commandExecutedEventResponse.command args = _convert_variant_to_value( diff --git a/tests/conftest.py b/tests/conftest.py index c6edd7b3835..14180cdf200 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,6 +16,8 @@ from ansys.fluent.core.utils.file_transfer_service import RemoteFileTransferStrategy from ansys.fluent.core.utils.fluent_version import FluentVersion +sys.path.append(Path(__file__).parent / "util") + def pytest_addoption(parser): parser.addoption( @@ -383,3 +385,16 @@ def periodic_rot_settings_session(new_solver_session): @pytest.fixture def disable_datamodel_cache(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(pyfluent, "DATAMODEL_USE_STATE_CACHE", False) + + +@pytest.fixture(params=["old", "new"]) +def datamodel_api_version_all(request, monkeypatch: pytest.MonkeyPatch) -> None: + if request.param == "new": + monkeypatch.setenv("REMOTING_NEW_DM_API", "1") + monkeypatch.setenv("REMOTING_MAPPED_NEW_DM_API", "1") + + +@pytest.fixture +def datamodel_api_version_new(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("REMOTING_NEW_DM_API", "1") + monkeypatch.setenv("REMOTING_MAPPED_NEW_DM_API", "1") diff --git a/tests/test_datamodel_api.py b/tests/test_datamodel_api.py new file mode 100644 index 00000000000..83be2a129e4 --- /dev/null +++ b/tests/test_datamodel_api.py @@ -0,0 +1,449 @@ +import time + +import pytest +from util import create_datamodel_root_in_server, create_root_using_datamodelgen + +from ansys.fluent.core.services.datamodel_se import ( + SubscribeEventError, + convert_path_to_se_path, +) +from ansys.fluent.core.utils.execution import timeout_loop + +rule_str = ( + "RULES:\n" + " STRING: X\n" + " default = ijk\n" + " END\n" + " SINGLETON: ROOT\n" + " members = A, B, D, G\n" + " commands= C\n" + " SINGLETON: A\n" + " members = X\n" + " x = $./X\n" + " END\n" + " OBJECT: B\n" + " members = X\n" + " END\n" + " SINGLETON: D\n" + " members = E, F, X\n" + " SINGLETON: E\n" + " members = X\n" + " END\n" + " SINGLETON: F\n" + " members = X\n" + " END\n" + " END\n" + " SINGLETON: G\n" + " members = H\n" + " DICT: H\n" + " END\n" + " END\n" + " COMMAND: C\n" + " arguments = X\n" + " x = $/A/X\n" + " END\n" + " END\n" + "END\n" +) + + +@pytest.mark.fluent_version(">=25.2") +def test_env_var_setting(datamodel_api_version_all, request, new_solver_session): + solver = new_solver_session + test_name = request.node.name + for var in ["REMOTING_NEW_DM_API", "REMOTING_MAPPED_NEW_DM_API"]: + # TODO: It might be possible to check the param value in the fixture + # instead of checking the test name here. + if test_name.endswith("[old]"): + assert solver.scheme_eval.scheme_eval(f'(getenv "{var}")') is None + elif test_name.endswith("[new]"): + assert solver.scheme_eval.scheme_eval(f'(getenv "{var}")') == "1" + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_child_created(datamodel_api_version_all, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + + called = 0 + created = [] + + def cb(obj): + nonlocal called + nonlocal created + called += 1 + created.append(convert_path_to_se_path(obj.path)) + + subscription = service.add_on_child_created(app_name, "/", "B", root, cb) + assert called == 0 + assert created == [] + service.set_state(app_name, "/", {"B:b": {"_name_": "b"}}) + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + assert created == ["/B:b"] + subscription.unsubscribe() + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_changed(datamodel_api_version_all, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called = 0 + state = None + called_obj = 0 + state_obj = None + + def cb(obj): + nonlocal called + nonlocal state + state = obj() + called += 1 + + def cb_obj(obj): + nonlocal called_obj + nonlocal state_obj + state_obj = obj() + called_obj += 1 + + subscription = service.add_on_changed(app_name, "/A/X", root.A.X, cb) + subscription_obj = service.add_on_changed(app_name, "/A", root.A, cb_obj) + assert called == 0 + assert state is None + assert called_obj == 0 + assert state_obj is None + service.set_state(app_name, "/A/X", "lmn") + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + assert state == "lmn" + assert called_obj == 1 + assert state_obj == {"X": "lmn"} + service.set_state(app_name, "/A/X", "abc") + timeout_loop(lambda: called == 2, timeout=5) + assert called == 2 + assert state == "abc" + assert called_obj == 2 + assert state_obj == {"X": "abc"} + subscription.unsubscribe() + subscription_obj.unsubscribe() + service.set_state(app_name, "/A/X", "xyz") + time.sleep(5) + assert called == 2 + assert state == "abc" + assert called_obj == 2 + assert state_obj == {"X": "abc"} + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_affected(datamodel_api_version_all, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called = 0 + + def cb(obj): + nonlocal called + called += 1 + + subscription = service.add_on_affected(app_name, "/D", root.D, cb) + assert called == 0 + service.set_state(app_name, "/D/X", "lmn") + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + service.set_state(app_name, "/D/E/X", "lmn") + timeout_loop(lambda: called == 2, timeout=5) + assert called == 2 + service.set_state(app_name, "/A/X", "lmn") + time.sleep(5) + assert called == 2 + subscription.unsubscribe() + service.set_state(app_name, "/D/E/X", "pqr") + time.sleep(5) + assert called == 2 + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_affected_at_type_path( + datamodel_api_version_all, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called = 0 + + def cb(obj): + nonlocal called + called += 1 + + subscription = service.add_on_affected_at_type_path( + app_name, "/D", "E", root.D.E, cb + ) + assert called == 0 + service.set_state(app_name, "/D/X", "lmn") + time.sleep(5) + assert called == 0 + service.set_state(app_name, "/D/E/X", "lmn") + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + service.set_state(app_name, "/D/F/X", "lmn") + time.sleep(5) + assert called == 1 + subscription.unsubscribe() + service.set_state(app_name, "/D/E/X", "pqr") + time.sleep(5) + assert called == 1 + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_deleted( + datamodel_api_version_all, request, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called = False + called_obj = False + + def cb(): + nonlocal called + called = True + + def cb_obj(): + nonlocal called_obj + called_obj = True + + service.set_state(app_name, "/", {"B:b": {"_name_": "b"}}) + subscription = service.add_on_deleted(app_name, "/B:b/X", root.B["b"].X, cb) + subscription_obj = service.add_on_deleted(app_name, "/B:b", root.B["b"], cb_obj) + assert not called + assert not called_obj + service.delete_object(app_name, "/B:b") + timeout_loop(lambda: called_obj, timeout=5) + test_name = request.node.name + # Note comment in StateEngine test testDataModelAPIOnDeleted + if test_name.endswith("[old]"): + assert called + elif test_name.endswith("[new]"): + assert not called + assert called_obj + subscription.unsubscribe() + subscription_obj.unsubscribe() + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_attribute_changed( + datamodel_api_version_all, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called = 0 + value = None + + def cb(val): + nonlocal called + nonlocal value + value = val + called += 1 + + subscription = service.add_on_attribute_changed(app_name, "/A", "x", root.A, cb) + assert called == 0 + assert value is None + service.set_state(app_name, "/A/X", "cde") + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + assert value == "cde" + service.set_state(app_name, "/A/X", "xyz") + timeout_loop(lambda: called == 2, timeout=5) + assert called == 2 + assert value == "xyz" + subscription.unsubscribe() + service.set_state(app_name, "/A/X", "abc") + time.sleep(5) + assert called == 2 + assert value == "xyz" + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_command_attribute_changed( + datamodel_api_version_all, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called = 0 + value = None + + def cb(val): + nonlocal called + nonlocal value + value = val + called += 1 + + subscription = service.add_on_command_attribute_changed( + app_name, "/", "C", "x", root.C, cb + ) + assert called == 0 + assert value is None + service.set_state(app_name, "/A/X", "cde") + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + assert value == "cde" + service.set_state(app_name, "/A/X", "xyz") + timeout_loop(lambda: called == 2, timeout=5) + assert called == 2 + # TODO: value is still "cde" in both old and new API + # assert value == "xyz" + subscription.unsubscribe() + service.set_state(app_name, "/A/X", "abc") + time.sleep(5) + assert called == 2 + # Commented out because of the issue above + # assert value == "xyz" + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_command_executed( + datamodel_api_version_all, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + executed = 0 + command = None + arguments = None + + def cb(obj, cmd, args): + nonlocal executed + nonlocal command + nonlocal arguments + command = cmd + arguments = args + executed += 1 + + # TODO: In C++ API, we don't need to pass the command name + subscription = service.add_on_command_executed(app_name, "/", "C", root, cb) + assert executed == 0 + assert command is None + assert arguments is None + service.execute_command(app_name, "/", "C", dict(X="abc")) + timeout_loop(lambda: executed == 1, timeout=5) + assert executed == 1 + assert command == "C" + assert arguments == {"X": "abc"} + subscription.unsubscribe() + service.execute_command(app_name, "/", "C", dict(X="uvw")) + time.sleep(5) + assert executed == 1 + assert command == "C" + assert arguments == {"X": "abc"} + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_get_state(datamodel_api_version_all, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + assert service.get_state(app_name, "/A/X") == "ijk" + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_set_state(datamodel_api_version_all, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + service.set_state(app_name, "/A/X", "new_val") + assert service.get_state(app_name, "/A/X") == "new_val" + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_update_dict(datamodel_api_version_all, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + service.update_dict(app_name, "/G/H", {"X": "abc"}) + assert service.get_state(app_name, "/G/H") == {"X": "abc"} + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_bad_input( + datamodel_api_version_all, request, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + test_name = request.node.name + new_api = test_name.endswith("[new]") + with pytest.raises(SubscribeEventError): + service.add_on_child_created(app_name, "", "", root, lambda _: None) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_child_created(app_name, "/BB", "B", root, lambda _: None) + with pytest.raises(SubscribeEventError): + service.add_on_child_created(app_name, "/", "A", root, lambda _: None) + with pytest.raises(SubscribeEventError): + service.add_on_child_created(app_name, "/", "BB", root, lambda _: None) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_changed(app_name, "/BB", root, lambda _: None) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_deleted(app_name, "/BB", root, lambda: None) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_affected(app_name, "/BB", root, lambda _: None) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_affected_at_type_path(app_name, "/BB", "B", root, lambda: None) + # TODO: not raised in the old API - issue + if new_api: + with pytest.raises(SubscribeEventError): + service.add_on_affected_at_type_path( + app_name, "/", "BB", root, lambda: None + ) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_attribute_changed( + app_name, "/BB", "isActive", root, lambda _: None + ) + with pytest.raises(SubscribeEventError): + service.add_on_attribute_changed(app_name, "/A", "", root, lambda _: None) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_command_attribute_changed( + app_name, "/BB", "C", "isActive", root, lambda _: None + ) + with pytest.raises(SubscribeEventError): + service.add_on_command_attribute_changed( + app_name, "/A", "CC", "", root, lambda _: None + ) + with pytest.raises(SubscribeEventError): + service.add_on_command_attribute_changed( + app_name, "/", "CC", "isActive", root, lambda _: None + ) + with pytest.raises(RuntimeError if new_api else SubscribeEventError): # TODO: issue + service.add_on_command_executed(app_name, "/BB", "C", root, lambda _: None) + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_static_info(datamodel_api_version_all, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rule_str, app_name) + service = solver._se_service + assert service.get_static_info(app_name) diff --git a/tests/test_datamodel_service.py b/tests/test_datamodel_service.py index 074bc925d24..289c5bc8a0e 100644 --- a/tests/test_datamodel_service.py +++ b/tests/test_datamodel_service.py @@ -3,17 +3,14 @@ from google.protobuf.json_format import MessageToDict import pytest +from util import create_datamodel_root_in_server, create_root_using_datamodelgen from ansys.api.fluent.v0 import datamodel_se_pb2 from ansys.api.fluent.v0.variant_pb2 import Variant import ansys.fluent.core as pyfluent from ansys.fluent.core import examples from ansys.fluent.core.services.datamodel_se import ( - PyCommand, - PyMenu, PyMenuGeneric, - PyNamedObjectContainer, - PyTextual, ReadOnlyObjectError, _convert_value_to_variant, _convert_variant_to_value, @@ -125,7 +122,7 @@ def test_add_on_deleted(new_meshing_session): meshing.workflow.InitializeWorkflow(WorkflowType="Watertight Geometry") data = [] _ = meshing.workflow.TaskObject["Import Geometry"].add_on_deleted( - lambda obj: data.append(convert_path_to_se_path(obj.path)) + lambda: data.append(True) ) assert data == [] meshing.workflow.InitializeWorkflow(WorkflowType="Fault-tolerant Meshing") @@ -543,47 +540,12 @@ def test_read_only_set_state(new_meshing_session): ) -class test_root(PyMenu): - def __init__(self, service, rules, path): - self.A = self.__class__.A(service, rules, path + [("A", "")]) - super().__init__(service, rules, path) - - class A(PyNamedObjectContainer): - class _A(PyMenu): - def __init__(self, service, rules, path): - self.B = self.__class__.B(service, rules, path + [("B", "")]) - self.X = self.__class__.X(service, rules, path + [("X", "")]) - self.C = self.__class__.C(service, rules, "C", path) - super().__init__(service, rules, path) - - class B(PyNamedObjectContainer): - class _B(PyMenu): - pass - - class X(PyTextual): - pass - - class C(PyCommand): - pass - - -def _create_datamodel_root(session, rules_str) -> PyMenu: - rules_file_name = "test.fdl" - session.scheme_eval.scheme_eval( - f'(with-output-to-file "{rules_file_name}" (lambda () (format "~a" "{rules_str}")))' - ) - session.scheme_eval.scheme_eval( - '(state/register-new-state-engine "test" "test.fdl")' - ) - session.scheme_eval.scheme_eval(f'(remove-file "{rules_file_name}")') - assert session.scheme_eval.scheme_eval('(state/find-root "test")') > 0 - return test_root(session._se_service, "test", []) - - @pytest.mark.fluent_version(">=24.2") def test_on_child_created_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] _ = root.A["A1"].add_on_child_created("B", lambda _: data.append(1)) @@ -601,11 +563,13 @@ def test_on_child_created_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_deleted_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] - _ = root.A["A1"].add_on_deleted(lambda _: data.append(1)) - root.A["A1"].add_on_deleted(lambda _: data.append(2)) + _ = root.A["A1"].add_on_deleted(lambda: data.append(1)) + root.A["A1"].add_on_deleted(lambda: data.append(2)) gc.collect() assert "/test/deleted/A:A1" in solver._se_service.subscriptions assert "/test/deleted/A:A1-1" in solver._se_service.subscriptions @@ -622,7 +586,9 @@ def test_on_deleted_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_changed_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] _ = root.A["A1"].X.add_on_changed(lambda _: data.append(1)) @@ -640,7 +606,9 @@ def test_on_changed_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_affected_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] _ = root.A["A1"].add_on_affected(lambda _: data.append(1)) @@ -658,7 +626,9 @@ def test_on_affected_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_affected_at_type_path_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] _ = root.A["A1"].add_on_affected_at_type_path("B", lambda _: data.append(1)) @@ -676,7 +646,9 @@ def test_on_affected_at_type_path_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_command_executed_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] _ = root.A["A1"].add_on_command_executed("C", lambda *args: data.append(1)) @@ -694,7 +666,9 @@ def test_on_command_executed_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_attribute_changed_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] _ = root.A["A1"].add_on_attribute_changed("isABC", lambda _: data.append(1)) @@ -714,7 +688,9 @@ def test_on_attribute_changed_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_command_attribute_changed_lifetime(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) root.A["A1"] = {} data = [] _ = root.A["A1"].add_on_command_attribute_changed( @@ -748,7 +724,9 @@ def test_on_command_attribute_changed_lifetime(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_affected_lifetime_with_delete_child_objects(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) pyfluent.logging.enable() root.A["A1"] = {} data = [] @@ -767,7 +745,9 @@ def test_on_affected_lifetime_with_delete_child_objects(new_solver_session): @pytest.mark.fluent_version(">=24.2") def test_on_affected_lifetime_with_delete_all_child_objects(new_solver_session): solver = new_solver_session - root = _create_datamodel_root(solver, test_rules) + app_name = "test" + create_datamodel_root_in_server(solver, test_rules, app_name) + root = create_root_using_datamodelgen(solver._se_service, app_name) pyfluent.logging.enable() root.A["A1"] = {} data = [] diff --git a/tests/test_mapped_api.py b/tests/test_mapped_api.py new file mode 100644 index 00000000000..6132bcc33f0 --- /dev/null +++ b/tests/test_mapped_api.py @@ -0,0 +1,785 @@ +import time + +import pytest +from util import create_datamodel_root_in_server, create_root_using_datamodelgen + +from ansys.fluent.core.services.datamodel_se import convert_path_to_se_path +from ansys.fluent.core.utils.execution import timeout_loop + +rules_str = ( + "RULES:\n" + " STRING: X\n" + " allowedValues = yes, no\n" + " logicalMapping = True, False\n" + " END\n" + " STRING: Y\n" + ' allowedValues = \\"1\\", \\"2\\", \\"3\\"\n' + ' default = \\"2\\"\n' + " isNumerical = True\n" + " END\n" + " INTEGER: Z\n" + " END\n" + " SINGLETON: ROOT\n" + " members = A\n" + " commands = C, D\n" + " SINGLETON: A\n" + " members = X, Y, Z\n" + " END\n" + " COMMAND: C\n" + " arguments = X\n" + " functionName = CFunc\n" + " END\n" + " COMMAND: D\n" + " arguments = X\n" + " functionName = CFunc\n" + " APIName = dd\n" + " END\n" + " END\n" + "END\n" +) + + +rules_str_caps = ( + "RULES:\n" + " STRING: X\n" + " allowedValues = Yes, No\n" + " default = No\n" + " logicalMapping = True, False\n" + " END\n" + " SINGLETON: ROOT\n" + " members = A\n" + " SINGLETON: A\n" + " members = X\n" + " END\n" + " END\n" + "END\n" +) + + +def get_static_info_value(static_info, type_path): + for p in type_path.removeprefix("/").split("/"): + static_info = static_info[p] + return static_info + + +def get_state_from_remote_app(session, app_name, type_path): + return session.scheme_eval.scheme_eval( + f'(state/object/get-state (state/object/find-child (state/find-root "{app_name}") "{type_path}"))' + ) + + +def get_error_state_message_from_remote_app(session, app_name, type_path): + return session.scheme_eval.scheme_eval( + f'(state/object/get-error-state-message (state/object/find-child (state/find-root "{app_name}") "{type_path}"))' + ) + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_bool_for_str_has_correct_type( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + static_info = service.get_static_info("test") + assert ( + get_static_info_value(static_info, "/singletons/A/parameters/X/type") + == "Logical" + ) + cmd_args = get_static_info_value(static_info, "/commands/C/commandinfo/args") + arg0 = cmd_args[0] + assert arg0["type"] == "Logical" + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_set_bool_for_str(datamodel_api_version_new, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + service.set_state(app_name, "/A/X", "yes") + assert service.get_state(app_name, "/A/X") is True + assert get_state_from_remote_app(solver, app_name, "/A/X") == "yes" + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_set_bool_nested_for_str( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + service.set_state(app_name, "/A", {"X": True}) + assert service.get_state(app_name, "/A/X") is True + assert get_error_state_message_from_remote_app(solver, app_name, "/A/X") is None + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_get_set_bool_for_str_with_flexible_strs_no_errors( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str_caps, app_name) + service = solver._se_service + service.set_state(app_name, "/A/X", True) + assert service.get_state(app_name, "/A/X") is True + assert get_error_state_message_from_remote_app(solver, app_name, "/A/X") is None + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_get_attrs_bool_for_str( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + assert service.get_attribute_value(app_name, "/A/Z", "allowedValues") is None + assert service.get_attribute_value(app_name, "/A/X", "allowedValues") is None + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_get_and_set_int_for_str( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + service.set_state(app_name, "/A/Y", 1) + assert service.get_state(app_name, "/A/Y") == 1 + assert get_error_state_message_from_remote_app(solver, app_name, "/A/Y") is None + + +# TODO: what are the equivalent of following tests in Python? +# testPopulateMappingAttrTablePaths +# testMapAPIStateToDM +# testMapDMStateToAPI +# testMapNestedAPIStateToDM +# testUpdateStateDictWithMapping + + +@pytest.mark.fluent_version(">=25.2") +def test_state_of_command_args_with_mapping( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + c_name = service.create_command_arguments(app_name, "/", "C") + with pytest.raises(RuntimeError): + service.set_state(app_name, f"/C:{c_name}/X", False) + assert service.get_state(app_name, f"/C:{c_name}") == {"X": None} + service.set_state(app_name, f"/C:{c_name}", {"X": False}) + assert service.get_state(app_name, f"/C:{c_name}") == {"X": False} + service.set_state(app_name, f"/C:{c_name}", {"X": True}) + assert service.get_state(app_name, f"/C:{c_name}") == {"X": True} + + +def register_external_function_in_remote_app(session, app_name, func_name): + session.scheme_eval.scheme_eval( + f'(state/register-external-fn "{app_name}" "{func_name}" (lambda (obj . args) (car args)) (cons "Variant" (list "ModelObject" "Variant")))' + ) + + +@pytest.mark.fluent_version(">=25.2") +def test_execute_command_with_args_mapping( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + register_external_function_in_remote_app(solver, app_name, "CFunc") + result = service.execute_command(app_name, "/", "C", {"X": True}) + assert result == "yes" + + +@pytest.mark.fluent_version(">=25.2") +def test_execute_command_with_args_and_path_mapping( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + register_external_function_in_remote_app(solver, app_name, "CFunc") + result = service.execute_command(app_name, "/", "dd", {"X": True}) + assert result == "yes" + + +@pytest.mark.fluent_version(">=25.2") +def test_execute_query_with_args_mapping(datamodel_api_version_new, new_solver_session): + rules_str = ( + "RULES:\n" + " STRING: X\n" + " allowedValues = yes, no\n" + " logicalMapping = True, False\n" + " END\n" + " SINGLETON: ROOT\n" + " queries = Q\n" + " QUERY: Q\n" + " arguments = X\n" + " functionName = QFunc\n" + " END\n" + " END\n" + "END\n" + ) + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + register_external_function_in_remote_app(solver, app_name, "QFunc") + result = service.execute_query(app_name, "/", "Q", {"X": True}) + assert result == "yes" + + +@pytest.mark.fluent_version(">=25.2") +def test_get_mapped_attr(datamodel_api_version_new, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + assert service.get_attribute_value(app_name, "/A/X", "allowedValues") is None + assert service.get_attribute_value(app_name, "/A/Y", "allowedValues") is None + assert service.get_attribute_value(app_name, "/A/Y", "min") == 1 + assert service.get_attribute_value(app_name, "/A/Y", "max") == 3 + assert service.get_attribute_value(app_name, "/A/Y", "default") == 2 + + +@pytest.mark.fluent_version(">=25.2") +def test_get_mapped_attr_defaults(datamodel_api_version_new, new_solver_session): + rules_str = ( + "RULES:\n" + " STRING: X\n" + " allowedValues = yes, no\n" + " default = no\n" + " logicalMapping = True, False\n" + " END\n" + " STRING: Y\n" + ' allowedValues = \\"1\\", \\"2\\", \\"3\\"\n' + ' default = \\"2\\"\n' + " isNumerical = True\n" + " END\n" + " INTEGER: Z\n" + " default = 42\n" + " END\n" + " SINGLETON: ROOT\n" + " members = A\n" + " SINGLETON: A\n" + " members = X, Y, Z\n" + " END\n" + " END\n" + "END\n" + ) + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + assert service.get_attribute_value(app_name, "/A/X", "default") is False + assert service.get_attribute_value(app_name, "/A/Y", "default") == 2 + assert service.get_attribute_value(app_name, "/A/Z", "default") == 42 + + +@pytest.mark.fluent_version(">=25.2") +def test_get_mapped_enum_attr(datamodel_api_version_new, new_solver_session): + rules_str = ( + "RULES:\n" + " STRING: X\n" + " allowedValues = ijk, lmn\n" + " default = lmn\n" + " enum = green, yellow\n" + " END\n" + " SINGLETON: ROOT\n" + " members = A\n" + " SINGLETON: A\n" + " members = X\n" + " END\n" + " END\n" + "END\n" + ) + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + assert service.get_attribute_value(app_name, "/A/X", "allowedValues") == [ + "green", + "yellow", + ] + assert service.get_attribute_value(app_name, "/A/X", "default") == "yellow" + + +@pytest.mark.fluent_version(">=25.2") +def test_get_mapped_dynamic_enum_attr(datamodel_api_version_new, new_solver_session): + rules_str = ( + "RULES:\n" + " LOGICAL: B\n" + " default = True\n" + " END\n" + " STRING: X\n" + ' allowedValues = IF($../B, (\\"ijk\\", \\"lmn\\"), (\\"ijk\\", \\"lmn\\", \\"opq\\"))\n' + " default = lmn\n" + ' enum = IF($../B, (\\"green\\", \\"yellow\\"), (\\"green\\", \\"yellow\\", \\"blue\\"))\n' + " END\n" + " SINGLETON: ROOT\n" + " members = A\n" + " SINGLETON: A\n" + " members = B, X\n" + " END\n" + " END\n" + "END\n" + ) + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + assert service.get_attribute_value(app_name, "/A/X", "allowedValues") == [ + "green", + "yellow", + ] + assert service.get_attribute_value(app_name, "/A/X", "default") == "yellow" + + +@pytest.mark.fluent_version(">=25.2") +def test_get_mapped_command_attr(datamodel_api_version_new, new_solver_session): + rules_str = ( + "RULES:\n" + " STRING: X\n" + " allowedValues = yes, no\n" + " default = no\n" + " logicalMapping = True, False\n" + " END\n" + " STRING: Y\n" + ' allowedValues = \\"1\\", \\"2\\", \\"3\\"\n' + ' default = \\"2\\"\n' + " isNumerical = True\n" + " END\n" + " INTEGER: Z\n" + " default = 42\n" + " END\n" + " SINGLETON: ROOT\n" + " commands = C\n" + " COMMAND: C\n" + " arguments = X, Y, Z\n" + " END\n" + " END\n" + "END\n" + ) + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + c_name = service.create_command_arguments(app_name, "/", "C") + # TODO: Attribute query at command argument level is not working + assert ( + service.get_attribute_value(app_name, f"/C:{c_name}", "X/allowedValues") is None + ) + assert ( + service.get_attribute_value(app_name, f"/C:{c_name}", "Y/allowedValues") is None + ) + assert service.get_attribute_value(app_name, f"/C:{c_name}", "Y/min") == 1 + assert service.get_attribute_value(app_name, f"/C:{c_name}", "Y/max") == 3 + assert service.get_attribute_value(app_name, f"/C:{c_name}", "X/default") is False + assert service.get_attribute_value(app_name, f"/C:{c_name}", "Y/default") == 2 + assert service.get_attribute_value(app_name, f"/C:{c_name}", "Z/default") == 42 + + +@pytest.mark.fluent_version(">=25.2") +def test_on_changed_is_mapped(datamodel_api_version_new, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + + called = 0 + state = None + called_obj = 0 + state_obj = None + + def on_changed(value): + nonlocal called + nonlocal state + state = value() + called += 1 + + def on_changed_obj(value): + nonlocal called_obj + nonlocal state_obj + state_obj = value() + called_obj += 1 + + subscription = service.add_on_changed(app_name, "/A/X", root.A.X, on_changed) + subscription_obj = service.add_on_changed(app_name, "/A", root.A, on_changed_obj) + + assert called == 0 + assert state is None + assert called_obj == 0 + assert state_obj is None + + service.set_state(app_name, "/A/X", True) + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + assert state is True + assert called_obj == 1 + assert state_obj == {"X": True, "Y": 2, "Z": None} + + service.set_state(app_name, "/A/X", False) + timeout_loop(lambda: called == 2, timeout=5) + assert called == 2 + assert state is False + assert called_obj == 2 + assert state_obj == {"X": False, "Y": 2, "Z": None} + + subscription.unsubscribe() + subscription_obj.unsubscribe() + + service.set_state(app_name, "/A/X", True) + time.sleep(5) + assert called == 2 + assert state is False + assert called_obj == 2 + assert state_obj == {"X": False, "Y": 2, "Z": None} + + +@pytest.mark.fluent_version(">=25.2") +def test_mapped_on_attribute_changed(datamodel_api_version_new, new_solver_session): + rules_str = ( + "RULES:\n" + " STRING: X\n" + " allowedValues = yes, no\n" + " default = $../Y\n" + " logicalMapping = True, False\n" + " END\n" + " STRING: Y\n" + " END\n" + " SINGLETON: ROOT\n" + " members = A\n" + " commands = C\n" + " SINGLETON: A\n" + " members = X, Y\n" + " END\n" + " COMMAND: C\n" + " arguments = X, Y\n" + " END\n" + " END\n" + "END\n" + ) + + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called = 0 + value = None + + def cb(val): + nonlocal called + nonlocal value + value = val + called += 1 + + subscription = service.add_on_attribute_changed( + app_name, "/A/X", "default", root.A.X, cb + ) + assert called == 0 + assert value is None + + service.set_state(app_name, "/A/Y", "no") + timeout_loop(lambda: called == 1, timeout=5) + assert called == 1 + assert value is False + + service.set_state(app_name, "/A/Y", "yes") + timeout_loop(lambda: called == 2, timeout=5) + assert called == 2 + assert value is True + + subscription.unsubscribe() + service.set_state(app_name, "/A/Y", "no") + time.sleep(5) + assert called == 2 + assert value is True + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_command_executed_mapped_args( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, rules_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + register_external_function_in_remote_app(solver, app_name, "CFunc") + executed = False + command = None + arguments = None + + def cb(obj, cmd, args): + nonlocal executed + nonlocal command + nonlocal arguments + command = cmd + arguments = args + executed = True + + subscription = service.add_on_command_executed(app_name, "/", "C", root, cb) + assert not executed + assert command is None + assert arguments is None + + service.execute_command(app_name, "/", "C", {"X": True}) + timeout_loop(lambda: executed, timeout=5) + assert executed + assert command == "C" + assert arguments == {"X": True} + + executed = False + command = None + arguments = None + + subscription.unsubscribe() + service.execute_command(app_name, "/", "C", {"X": False}) + time.sleep(5) + assert not executed + assert command is None + assert arguments is None + + +api_name_rules_str = ( + "RULES:\n" + " STRING: __X\n" + " allowedValues = yes, no\n" + " logicalMapping = True, False\n" + " attr1 = 42.0\n" + " APIName = xxx\n" + " END\n" + " STRING: __Y\n" + ' allowedValues = \\"1\\", \\"2\\", \\"3\\"\n' + ' default = \\"2\\"\n' + " isNumerical = True\n" + " APIName = yyy\n" + " END\n" + " INTEGER: Z\n" + " END\n" + " SINGLETON: ROOT\n" + " members = __A, B, __E\n" + " commands = __C, D\n" + " SINGLETON: __A\n" + " members = __X\n" + " APIName = aaa\n" + " END\n" + " OBJECT: B\n" + " members = __Y, Z\n" + " END\n" + " OBJECT: __E\n" + " members = __Y\n" + " APIName = eee\n" + " END\n" + " COMMAND: __C\n" + " arguments = __X\n" + " functionName = CFunc\n" + " APIName = ccc\n" + " END\n" + " COMMAND: D\n" + " arguments = __X\n" + " functionName = CFunc\n" + " END\n" + " END\n" + "END\n" +) + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_with_mapped_names(datamodel_api_version_new, new_solver_session): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + static_info = service.get_static_info(app_name) + assert ( + get_static_info_value(static_info, "/singletons/aaa/parameters/xxx/type") + == "Logical" + ) + assert ( + get_static_info_value(static_info, "/namedobjects/B/parameters/yyy/type") + == "Integer" + ) + assert ( + get_static_info_value(static_info, "/namedobjects/B/parameters/Z/type") + == "Integer" + ) + + command_args = [ + { + "helpstring": "", + "name": "xxx", + "type": "Logical", + } + ] + command_args = [sorted(x.items()) for x in command_args] + ccc_args = get_static_info_value( # noqa: F841 + static_info, "/commands/ccc/commandinfo/args" + ) + # TODO: helpstring is not being set + # assert command_args == [sorted(x.items()) for x in ccc_args] + d_args = get_static_info_value( # noqa: F841 + static_info, "/commands/D/commandinfo/args" + ) + # TODO: helpstring is not being returned + # assert command_args == [sorted(x.items()) for x in d_args] + + +# TODO: what are the equivalent of following tests in Python? +# testMapperMapDataModelPathToAPIPath +# testMapperMapAPIPathToDataModelPath +# testMapperMapDMValueToAPI + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_root_get_and_set_state_with_mapped_names( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + assert service.get_state(app_name, "/") == {"aaa": {"xxx": None}} + service.set_state(app_name, "/__A/__X", "yes") + assert service.get_state(app_name, "/") == {"aaa": {"xxx": True}} + service.set_state(app_name, "/", {"aaa": {"xxx": False}}) + assert service.get_state(app_name, "/") == {"aaa": {"xxx": False}} + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_root_get_attrs_with_mapped_names( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + assert service.get_attribute_value(app_name, "/aaa/xxx", "attr1") == 42.0 + service.set_state(app_name, "/", {"B:b": {}}) + assert service.get_attribute_value(app_name, "/B:b/yyy", "default") == 2 + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_cmd_args_op_with_mapped_names( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + c_name = service.create_command_arguments(app_name, "/", "ccc") + x_path_str = f"/__C:{c_name}/xxx" # noqa: F841 + # TODO: issue + # service.set_state(app_name, x_path_str, True) + service.set_state(app_name, f"/__C:{c_name}", {"xxx": True}) + assert service.get_state(app_name, f"/__C:{c_name}") == {"xxx": True} + assert service.get_attribute_value(app_name, f"/__C:{c_name}", "xxx/attr1") == 42.0 + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_rename_with_mapped_names( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + service.set_state(app_name, "/", {"B:b": {}}) + service.rename(app_name, "/B:b", "c") + service.set_state(app_name, "/", {"eee:e": {}}) + assert service.get_state(app_name, "/B:c/yyy") == 2 + service.rename(app_name, "/eee:e", "x") + assert service.get_state(app_name, "/eee:x/yyy") == 2 + + +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_delete_object_with_mapped_names( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + service.set_state(app_name, "/", {"B:b": {}}) + service.delete_object(app_name, "/B:b") + + +@pytest.mark.skip +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_created_on_changed_on_deleted_with_mapped_names( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + called_paths = [] + delete_count = 0 + changes = [] + + def create_cb(obj): + called_paths.append(convert_path_to_se_path(obj.path)) + + def delete_cb(): + nonlocal delete_count + delete_count += 1 + + def changed_cb(value): + changes.append(value()) + + service.add_on_child_created(app_name, "/", "eee", root, create_cb) + # TODO: fails at event streaming callback of on_child_created + # as the name "eee" is not available in the PyFluent side. + service.set_state(app_name, "/", {"eee:b": {}}) + service.set_state(app_name, "/", {"eee:c": {}}) + service.set_state(app_name, "/", {"B:d": {}}) + service.add_on_deleted(app_name, "/eee:b", root, delete_cb) + service.add_on_deleted(app_name, "/eee:c", root, delete_cb) + # TODO: Affected by name mangling of dunder members + service.add_on_changed(app_name, "/eee:b/yyy", root.__E["b"].__Y, changed_cb) + service.delete_object(app_name, "/eee:c") + service.set_state(app_name, "/", {"eee:b": {"yyy": 42}}) + assert called_paths == ["/eee:b", "/eee:c"] + assert delete_count == 1 + assert changes == [42] + + +@pytest.mark.skip +@pytest.mark.fluent_version(">=25.2") +def test_datamodel_api_on_changed_with_mapped_names( + datamodel_api_version_new, new_solver_session +): + solver = new_solver_session + app_name = "test" + create_datamodel_root_in_server(solver, api_name_rules_str, app_name) + service = solver._se_service + root = create_root_using_datamodelgen(service, app_name) + changes = [] + + def changed_cb(value): + changes.append(value()) + + service.set_state(app_name, "/", {"eee:b": {}}) + # TODO: Can't get this working due to name mangling of dunder members + service.add_on_changed(app_name, "/eee:b/yyy", root.__E["b"].__Y, changed_cb) + service.set_state(app_name, "/", {"eee:b": {"yyy": 42}}) + assert changes == [42] + + +# TODO: what are the equivalent of following tests in Python? +# testDataModelAPIWithNullCustomNameMapper +# testDataModelAPIWithAppendingCustomNameMapper +# testDataModelAPIWithSnakeyCustomNameMapper +# testDataModelAPIWithSnakeyCustomNameMapperAndMoreCamels diff --git a/tests/util/__init__.py b/tests/util/__init__.py new file mode 100644 index 00000000000..2be1c68ced9 --- /dev/null +++ b/tests/util/__init__.py @@ -0,0 +1,36 @@ +from pathlib import Path +from tempfile import TemporaryDirectory +import uuid + +from pytest import MonkeyPatch + +import ansys.fluent.core as pyfluent +from ansys.fluent.core.codegen import StaticInfoType, datamodelgen +from ansys.fluent.core.utils import load_module + + +def create_datamodel_root_in_server(session, rules_str, app_name) -> None: + rules_file_name = f"{uuid.uuid4()}.fdl" + session.scheme_eval.scheme_eval( + f'(with-output-to-file "{rules_file_name}" (lambda () (format "~a" "{rules_str}")))', + ) + session.scheme_eval.scheme_eval( + f'(state/register-new-state-engine "{app_name}" "{rules_file_name}")' + ) + session.scheme_eval.scheme_eval(f'(remove-file "{rules_file_name}")') + assert session.scheme_eval.scheme_eval(f'(state/find-root "{app_name}")') > 0 + + +def create_root_using_datamodelgen(service, app_name): + version = "252" + static_info = service.get_static_info(app_name) + with TemporaryDirectory() as temp_dir: + with MonkeyPatch.context() as m: + m.setattr(pyfluent, "CODEGEN_OUTDIR", Path(temp_dir)) + # TODO: Refactor datamdodelgen so we don't need to hardcode StaticInfoType + datamodelgen.generate( + version, static_infos={StaticInfoType.DATAMODEL_WORKFLOW: static_info} + ) + gen_file = Path(temp_dir) / f"datamodel_{version}" / "workflow.py" + module = load_module("datamodel", gen_file) + return module.Root(service, app_name, [])