taipy/tests/gui/helpers.py

# 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 urllib.request
import warnings

import socketio

from taipy.gui import Gui, Html, Markdown
from taipy.gui._renderers.builder import _Builder
from taipy.gui._warnings import TaipyGuiWarning
from taipy.gui.servers.utils import get_server_type
from taipy.gui.utils._variable_directory import _reset_name_map
from taipy.gui.utils.expr_var_name import _reset_expr_var_name


class WebSocketTestClient:
    def __init__(self, url: str = "http://localhost:5000", auto_connect=True):
        self.url = url
        self.sio = socketio.Client()
        self.messages: t.List = []
        # Register event handlers
        self.sio.on("message", self._on_message)
        # Auto connect if specified
        if auto_connect:
            self.connect()

    def _on_message(self, data):
        self.messages.append(data)

    def connect(self):
        self.sio.connect(f"{self.url}/socket.io")

    def disconnect(self):
        self.sio.disconnect()

    def emit(self, *args, **kwargs):
        self.sio.emit(*args, **kwargs)
        self.sio.sleep(1)

    def get_received(self, timeout=1):
        self.sio.sleep(timeout)
        # wrap the messages in a dict to match the expected format
        wrapped_messages = []
        wrapped_messages.extend({"name": "message", "args": message} for message in self.messages)
        self.messages = []
        return wrapped_messages

    def get(self, url):
        with urllib.request.urlopen(self.url + url) as response:
            return response

    def get_sid(self) -> t.Optional[str]:
        return self.sio.get_sid()


class Helpers:
    @staticmethod
    def test_cleanup():
        _Builder._reset_key()
        _reset_name_map()
        _reset_expr_var_name()
        Gui._Gui__extensions.clear()
        Gui._Gui__shared_variables.clear()
        Gui._Gui__content_providers.clear()

    @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 = Helpers.get_response_data(response)
        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,
                server_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)]

    @staticmethod
    def get_response_data(response):
        if get_server_type() == "flask":
            return json.loads(response.get_data().decode("utf-8", "ignore"))
        elif get_server_type() == "fastapi":
            return response.json()
        return None

    @staticmethod
    def get_response_raw_data(response):
        if get_server_type() == "flask":
            return response.get_data().decode("utf-8", "ignore")
        elif get_server_type() == "fastapi":
            return response.text
        return None

    @staticmethod
    def get_socketio_test_client():
        return WebSocketTestClient()