# Copyright 2021-2025 Avaiga Private Limited # # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the # specific language governing permissions and limitations under the License. import json import logging import socket import time import typing as t import warnings from taipy.gui import Gui, Html, Markdown from taipy.gui._renderers.builder import _Builder from taipy.gui._warnings import TaipyGuiWarning from taipy.gui.utils._variable_directory import _reset_name_map from taipy.gui.utils.expr_var_name import _reset_expr_var_name class Helpers: @staticmethod def test_cleanup(): _Builder._reset_key() _reset_name_map() _reset_expr_var_name() @staticmethod def test_control_md(gui: Gui, md_string: str, expected_values: t.Union[str, t.List]): gui.add_page("test", Markdown(md_string, frame=None)) Helpers._test_control(gui, expected_values) @staticmethod def test_control_html(gui: Gui, html_string: str, expected_values: t.Union[str, t.List]): gui.add_page("test", Html(html_string, frame=None)) Helpers._test_control(gui, expected_values) @staticmethod def test_control_builder(gui: Gui, builder_page, expected_values: t.Union[str, t.List]): gui.add_page("test", builder_page) Helpers._test_control(gui, expected_values) @staticmethod def _test_control(gui: Gui, expected_values: t.Union[str, t.List]): gui.run(run_server=False, single_client=True, stylekit=False) client = gui._server.test_client() response = client.get("/taipy-jsx/test") assert response.status_code == 200, f"response.status_code {response.status_code} != 200" response_data = json.loads(response.get_data().decode("utf-8", "ignore")) assert isinstance(response_data, t.Dict), "response_data is not Dict" assert "jsx" in response_data, "jsx not in response_data" jsx = response_data["jsx"] logging.getLogger().debug(jsx) if isinstance(expected_values, str): assert jsx == expected_values, f"{jsx} != {expected_values}" elif isinstance(expected_values, list): for expected_value in expected_values: assert expected_value in jsx, f"{expected_value} not in {jsx}" @staticmethod def assert_outward_ws_message(received_message, type, varname, value): assert isinstance(received_message, dict) assert "name" in received_message and received_message["name"] == "message" assert "args" in received_message args = received_message["args"] assert "type" in args and args["type"] == type assert "payload" in args payload_arr = args["payload"] found_payload = False for payload in payload_arr: if "name" in payload and varname in payload["name"]: assert "payload" in payload and "value" in payload["payload"] and payload["payload"]["value"] == value found_payload = True logging.getLogger().debug(payload["payload"]["value"]) assert found_payload @staticmethod def assert_outward_simple_ws_message(received_message, type, varname, value): assert isinstance(received_message, dict) assert "name" in received_message and received_message["name"] == "message" assert "args" in received_message args = received_message["args"] assert "type" in args and args["type"] == type assert "name" in args and args["name"] == varname assert "payload" in args payload = args["payload"] assert "value" in payload and payload["value"] == value logging.getLogger().debug(payload["value"]) @staticmethod def assert_outward_ws_simple_message(received_message, aType, values): assert isinstance(received_message, dict) assert "name" in received_message and received_message["name"] == "message" assert "args" in received_message args = received_message["args"] assert "type" in args and args["type"] == aType for k, v in values.items(): assert k in args and args[k] == v logging.getLogger().debug(f"{k}: {args[k]}") @staticmethod def assert_outward_ws_multiple_message(received_message, type, array_len: int): assert isinstance(received_message, dict) assert "name" in received_message and received_message["name"] == "message" assert "args" in received_message args = received_message["args"] assert "type" in args and args["type"] == type assert "payload" in args payload = args["payload"] assert isinstance(payload, list) assert len(payload) == array_len logging.getLogger().debug(payload) @staticmethod def create_scope_and_get_sid(gui: Gui) -> str: sid = "test" gui._bindings()._get_or_create_scope(sid) return sid @staticmethod def port_check(): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(1) if s.connect_ex(("127.0.0.1", 5000)) == 0: s.close() return True else: s.close() return False @staticmethod def run_e2e(gui, **kwargs): kwargs["run_in_thread"] = True kwargs["single_client"] = True kwargs["run_browser"] = False kwargs["stylekit"] = kwargs.get("stylekit", False) with warnings.catch_warnings(record=True): gui.run(**kwargs) while not Helpers.port_check(): time.sleep(0.1) @staticmethod def run_e2e_multi_client(gui: Gui): with warnings.catch_warnings(record=True): gui.run(run_server=False, run_browser=False, single_client=False, stylekit=False) gui._server.run( host=gui._get_config("host", "127.0.0.1"), port=gui._get_config("port", 5000), client_url=gui._get_config("client_url", "http://localhost:{port}"), debug=False, use_reloader=False, flask_log=False, run_in_thread=True, allow_unsafe_werkzeug=False, notebook_proxy=False, port_auto_ranges=gui._get_config("port_auto_ranges", None), ) while not Helpers.port_check(): time.sleep(0.1) @staticmethod def get_taipy_warnings(warns: t.List[warnings.WarningMessage]) -> t.List[warnings.WarningMessage]: return [w for w in warns if issubclass(w.category, TaipyGuiWarning)]