# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.

"""
Sigima Client Stub/Mock Server (Real Objects)
---------------------------------------------

This module provides a stub XML-RPC server that emulates DataLab's XML-RPC interface
for testing purposes using real Sigima objects. The stub server allows tests to run
without requiring a real DataLab instance.
"""

from __future__ import annotations

import os
import threading
import uuid
from contextlib import contextmanager
from socketserver import ThreadingMixIn
from typing import TYPE_CHECKING
from xmlrpc.client import Binary
from xmlrpc.server import SimpleXMLRPCServer

from guidata.env import execenv

from sigima.client import utils
from sigima.objects import ImageObj, SignalObj, create_image, create_signal

if TYPE_CHECKING:
    from collections.abc import Generator

# pylint: disable=invalid-name  # Allows short reference names like x, y, ...
# pylint: disable=duplicate-code


class MockGroup:
    """Mock group object."""

    def __init__(self, title: str, uuid_str: str | None = None):
        self.title = title
        self.uuid = uuid_str or str(uuid.uuid4())
        self.objects: list[str] = []  # List of object UUIDs


class ThreadingXMLRPCServer(ThreadingMixIn, SimpleXMLRPCServer):
    """Threading XML-RPC server to handle multiple requests."""

    daemon_threads = True
    allow_reuse_address = True


class DataLabStubServer:
    """Stub XML-RPC server emulating DataLab XML-RPC interface with real objects.

    This server provides mock implementations of all DataLab XML-RPC methods
    using real SignalObj and ImageObj instances for maximum compatibility.

    Args:
        port: Port to bind to. If 0, uses a random available port.
        verbose: If True, print verbose debug information.
    """

    def __init__(self, port: int = 0, verbose: bool = True) -> None:
        """Initialize the stub server.

        Args:
            port: Port to bind to. If 0, uses a random available port.
        """
        self.port = port
        self.verbose = verbose
        self.server: ThreadingXMLRPCServer | None = None
        self.server_thread: threading.Thread | None = None

        # Real Sigima object storage
        self.signals: dict[str, SignalObj] = {}  # uuid -> SignalObj
        self.images: dict[str, ImageObj] = {}  # uuid -> ImageObj
        self.signal_groups: dict[str, MockGroup] = {}  # uuid -> group
        self.image_groups: dict[str, MockGroup] = {}  # uuid -> group

        # Current state
        self.current_panel = "signal"  # "signal", "image", or "macro"
        self.selected_objects: list[str] = []  # list of UUIDs
        self.selected_groups: list[str] = []  # list of group UUIDs
        self.auto_refresh = True
        self.show_titles = True

        # Add default groups
        self._add_default_group("signal")
        self._add_default_group("image")

    def _add_default_group(self, panel: str) -> None:
        """Add default group for a panel."""
        group = MockGroup("Group 1")
        if panel == "signal":
            self.signal_groups[group.uuid] = group
        elif panel == "image":
            self.image_groups[group.uuid] = group

    def start(self) -> int:
        """Start the XML-RPC server.

        Returns:
            Port number the server is listening on
        """
        self.server = ThreadingXMLRPCServer(
            ("127.0.0.1", self.port), allow_none=True, logRequests=False
        )

        # Register all methods
        self._register_functions()

        self.port = self.server.server_address[1]

        # Start server in a separate thread
        self.server_thread = threading.Thread(
            target=self.server.serve_forever, daemon=True
        )
        self.server_thread.start()

        execenv.print(f"DataLab stub server started on port {self.port}")
        return self.port

    def stop(self) -> None:
        """Stop the XML-RPC server."""
        if self.server:
            self.server.shutdown()
            self.server.server_close()
            if self.server_thread:
                self.server_thread.join(timeout=1.0)
            execenv.print("DataLab stub server stopped")

    def _register_functions(self) -> None:
        """Register all XML-RPC functions."""
        # System introspection methods
        self.server.register_introspection_functions()

        # Basic server methods
        self.server.register_function(self.get_version, "get_version")
        self.server.register_function(self.close_application, "close_application")
        self.server.register_function(self.raise_window, "raise_window")

        # Panel management
        self.server.register_function(self.get_current_panel, "get_current_panel")
        self.server.register_function(self.set_current_panel, "set_current_panel")

        # Application control
        self.server.register_function(self.reset_all, "reset_all")
        self.server.register_function(self.toggle_auto_refresh, "toggle_auto_refresh")
        self.server.register_function(self.toggle_show_titles, "toggle_show_titles")

        # File operations
        self.server.register_function(self.save_to_h5_file, "save_to_h5_file")
        self.server.register_function(self.open_h5_files, "open_h5_files")
        self.server.register_function(self.import_h5_file, "import_h5_file")

        # Object operations
        self.server.register_function(self.add_signal, "add_signal")
        self.server.register_function(self.add_image, "add_image")
        self.server.register_function(self.get_object_titles, "get_object_titles")
        self.server.register_function(self.get_object_uuids, "get_object_uuids")
        self.server.register_function(self.get_object, "get_object")
        self.server.register_function(self.get_object_shapes, "get_object_shapes")
        self.server.register_function(self.delete_metadata, "delete_metadata")

        # Selection operations
        self.server.register_function(self.select_objects, "select_objects")
        self.server.register_function(self.select_groups, "select_groups")
        self.server.register_function(self.get_sel_object_uuids, "get_sel_object_uuids")
        self.server.register_function(self.delete_object, "delete_object")
        self.server.register_function(self.duplicate_object, "duplicate_object")
        self.server.register_function(self.copy_metadata, "copy_metadata")

        # Group operations
        self.server.register_function(self.add_group, "add_group")
        self.server.register_function(self.get_group_titles, "get_group_titles")
        self.server.register_function(
            self.get_group_titles_with_object_info, "get_group_titles_with_object_info"
        )
        self.server.register_function(self.move_up, "move_up")
        self.server.register_function(self.move_down, "move_down")
        self.server.register_function(self.delete_group, "delete_group")

        # Calculation operations
        self.server.register_function(self.calc, "calc")

        # Annotation operations
        self.server.register_function(
            self.add_annotations_from_items, "add_annotations_from_items"
        )
        self.server.register_function(self.add_label_with_title, "add_label_with_title")

    # Basic server methods
    def get_version(self) -> str:
        """Get DataLab version.

        Returns a valid PEP 440 version string for testing purposes.
        Since this is a stub server, we return "1.0.0" which is compliant
        with both PEP 440 and the minimum version requirement.
        """
        return "1.0.0"

    def close_application(self) -> None:
        """Close DataLab application."""
        # In stub mode, do nothing

    def raise_window(self) -> None:
        """Raise DataLab window."""
        # In stub mode, do nothing

    # Panel management
    def get_current_panel(self) -> str:
        """Get current panel name."""
        return self.current_panel

    def set_current_panel(self, panel: str) -> None:
        """Set current panel."""
        if panel in ("signal", "image", "macro"):
            self.current_panel = panel

    # Application control
    def reset_all(self) -> None:
        """Reset all data."""
        self.signals.clear()
        self.images.clear()
        self.selected_objects.clear()
        self.selected_groups.clear()

    def toggle_auto_refresh(self, state: bool) -> None:
        """Toggle auto refresh mode."""
        self.auto_refresh = state
        if self.verbose:
            execenv.print(f"[STUB] Auto-refresh set to: {state}")

    def toggle_show_titles(self, state: bool) -> None:
        """Toggle show titles mode."""
        self.show_titles = state
        if self.verbose:
            execenv.print(f"[STUB] Show titles set to: {state}")

    # File operations
    def save_to_h5_file(self, filename: str) -> None:
        """Save to a DataLab HDF5 file."""
        if self.verbose:
            execenv.print(f"[STUB] Simulating H5 file save to: {filename}")
        # In stub mode, just create a dummy text file to simulate the save operation
        # This avoids HDF5 dependencies and potential test failures
        try:
            with open(filename, "w", encoding="utf-8") as f:
                f.write("# DataLab stub file (for testing)\n")
                f.write(f"# Signals: {len(self.signals)}\n")
                f.write(f"# Images: {len(self.images)}\n")
                f.write("# This is a dummy file created by the stub server\n")
            if self.verbose:
                execenv.print(
                    f"[STUB] Successfully created dummy file with {len(self.signals)} "
                    f"signals and {len(self.images)} images"
                )
        except Exception as exc:  # pylint: disable=broad-except
            if self.verbose:
                execenv.print(f"[STUB] Failed to create dummy file: {exc}")
            # Ignore errors in stub mode

    # pylint: disable=unused-argument
    def open_h5_files(
        self,
        h5files: list[str] | None = None,
        import_all: bool | None = None,
        reset_all: bool | None = None,
    ) -> None:
        """Open a DataLab HDF5 file or import from any other HDF5 file."""
        if h5files is None:
            return

        if self.verbose:
            execenv.print(f"[STUB] Simulating H5 file loading: {h5files}")

        if reset_all:
            if self.verbose:
                execenv.print("[STUB] Resetting all data before loading")
            self.reset_all()

        # In stub mode, just simulate loading by creating dummy objects
        # This avoids complex HDF5 file parsing and potential test failures
        for _i, filename in enumerate(h5files):
            # Create a dummy signal for each file
            signal = create_signal(f"Loaded Signal from {os.path.basename(filename)}")
            self.signals[str(uuid.uuid4())] = signal
            if self.verbose:
                execenv.print(f"[STUB] Created dummy signal: {signal.title}")

            # Create a dummy image for each file
            image = create_image(f"Loaded Image from {os.path.basename(filename)}")
            self.images[str(uuid.uuid4())] = image
            if self.verbose:
                execenv.print(f"[STUB] Created dummy image: {image.title}")

    def import_h5_file(self, filename: str, reset_all: bool | None = None) -> None:
        """Open DataLab HDF5 browser to Import HDF5 file."""
        self.open_h5_files([filename], import_all=True, reset_all=reset_all)

    # Object operations
    # pylint: disable=unused-argument
    def add_signal(
        self,
        title: str,
        xbinary: Binary,
        ybinary: Binary,
        xunit: str = "",
        yunit: str = "",
        xlabel: str = "",
        ylabel: str = "",
        group_id: str = "",
        set_current: bool = True,
    ) -> bool:
        """Add signal data to DataLab."""
        xdata = utils.rpcbinary_to_array(xbinary)
        ydata = utils.rpcbinary_to_array(ybinary)

        # Create real SignalObj using factory function
        signal = create_signal(title, x=xdata, y=ydata)
        signal.xunit = xunit or ""
        signal.yunit = yunit or ""
        signal.xlabel = xlabel or ""
        signal.ylabel = ylabel or ""

        # Store signal
        obj_uuid = str(uuid.uuid4())
        self.signals[obj_uuid] = signal

        # Add to group if specified
        if group_id and group_id in self.signal_groups:
            self.signal_groups[group_id].objects.append(obj_uuid)

        return True

    # pylint: disable=unused-argument
    def add_image(
        self,
        title: str,
        zbinary: Binary,
        xunit: str = "",
        yunit: str = "",
        zunit: str = "",
        xlabel: str = "",
        ylabel: str = "",
        zlabel: str = "",
        group_id: str = "",
        set_current: bool = True,
    ) -> bool:
        """Add image data to DataLab."""
        data = utils.rpcbinary_to_array(zbinary)

        # Create real ImageObj using factory function
        image = create_image(title, data=data)
        image.xunit = xunit or ""
        image.yunit = yunit or ""
        image.zunit = zunit or ""
        image.xlabel = xlabel or ""
        image.ylabel = ylabel or ""
        image.zlabel = zlabel or ""

        # Store image
        obj_uuid = str(uuid.uuid4())
        self.images[obj_uuid] = image

        # Add to group if specified
        if group_id and group_id in self.image_groups:
            self.image_groups[group_id].objects.append(obj_uuid)

        return True

    def add_object(
        self, obj_data: list[str], group_id: str = "", set_current: bool = True
    ) -> bool:
        """Add object to stub server.

        Args:
            obj_serialized: Serialized signal or image object
            group_id: group id in which to add the object. Defaults to ""
            set_current: if True, set the added object as current
        """
        obj: SignalObj | ImageObj = utils.rpcjson_to_dataset(obj_data)
        if self.verbose:
            obj_str = "signal" if isinstance(obj, SignalObj) else "image"
            obj_uuid = str(uuid.uuid4())
            print(f"Added {obj_str} {obj.title} with UUID {obj_uuid}")
            if isinstance(obj, SignalObj):
                self.signals[obj_uuid] = obj
                if group_id and group_id in self.signal_groups:
                    self.signal_groups[group_id].objects.append(obj_uuid)
            else:
                self.images[obj_uuid] = obj
                if group_id and group_id in self.image_groups:
                    self.image_groups[group_id].objects.append(obj_uuid)

    def load_from_files(self, filenames: list[str]) -> None:
        """Load objects from files (stub implementation).

        Args:
            filenames: list of file names
        """
        if self.verbose:
            print(
                f"load_from_files called with {len(filenames)} files "
                "(stub - not implemented)"
            )

    def load_from_directory(self, path: str) -> None:
        """Load objects from directory (stub implementation).

        Args:
            path: directory path
        """
        if self.verbose:
            print(
                f"load_from_directory called with path: {path} (stub - not implemented)"
            )

    def get_object_titles(self, panel: str | None = None) -> list[str]:
        """Get object titles for panel."""
        panel = panel or self.current_panel
        if panel == "signal":
            return [signal.title for signal in self.signals.values()]
        if panel == "image":
            return [image.title for image in self.images.values()]
        return []

    # pylint: disable=unused-argument
    def get_object_uuids(
        self, panel: str | None = None, group: int | str | None = None
    ) -> list[str]:
        """Get object UUIDs for panel."""
        panel = panel or self.current_panel
        if panel == "signal":
            return list(self.signals.keys())
        if panel == "image":
            return list(self.images.keys())
        return []

    def get_object(self, uuid_str: str, panel: str | None = None) -> list[str] | None:
        """Get object by UUID, index, or title."""
        panel = panel or self.current_panel

        # Get the appropriate objects dictionary
        if panel == "signal":
            objects = self.signals
            object_list = list(self.signals.keys())
        elif panel == "image":
            objects = self.images
            object_list = list(self.images.keys())
        else:
            return None

        # Try to resolve uuid_str as UUID first
        if uuid_str in objects:
            obj = objects[uuid_str]
        else:
            # Try to resolve as 1-based index
            try:
                index = int(uuid_str) - 1  # Convert to 0-based
                if 0 <= index < len(object_list):
                    uuid_key = object_list[index]
                    obj = objects[uuid_key]
                else:
                    return None
            except (ValueError, TypeError):
                # Try to find by title
                obj = None
                for object_instance in objects.values():
                    if object_instance.title == uuid_str:
                        obj = object_instance
                        break
                if obj is None:
                    return None

        # Use standard serialization with real objects
        return utils.dataset_to_rpcjson(obj)

    # pylint: disable=unused-argument
    def get_object_shapes(
        self, uuid_str: str, panel: str | None = None
    ) -> list[dict] | None:
        """Get object shapes."""
        obj = self.signals.get(uuid_str) or self.images.get(uuid_str)
        if obj is None:
            return None
        return [roi.to_dict() for roi in obj.roi]

    def delete_metadata(self, uuid_str: str, key: str) -> bool:
        """Delete metadata entry for object."""
        obj = self.signals.get(uuid_str) or self.images.get(uuid_str)
        if obj is None:
            return False
        if key in obj.metadata:
            del obj.metadata[key]
            return True
        return False

    # Selection operations
    def select_objects(
        self, selection: list[int | str], panel: str | None = None
    ) -> None:
        """Select objects by indices or UUIDs."""
        panel = panel or self.current_panel
        if panel == "signal":
            uuids = list(self.signals.keys())
        elif panel == "image":
            uuids = list(self.images.keys())
        else:
            return

        selected_uuids = []
        for item in selection:
            if isinstance(item, str):
                # Item is a UUID
                if item in uuids:
                    selected_uuids.append(item)
            elif isinstance(item, int):
                # Item is a 1-based index
                index = item - 1  # Convert to 0-based
                if 0 <= index < len(uuids):
                    selected_uuids.append(uuids[index])

        self.selected_objects = selected_uuids

    # pylint: disable=unused-argument
    def get_sel_object_uuids(self, include_groups: bool = False) -> list[str]:
        """Return selected objects uuids."""
        return self.selected_objects.copy()

    def select_groups(self, selection: list[int], panel: str | None = None) -> None:
        """Select groups by indices."""
        panel = panel or self.current_panel
        if panel == "signal":
            group_uuids = list(self.signal_groups.keys())
        elif panel == "image":
            group_uuids = list(self.image_groups.keys())
        else:
            return

        self.selected_groups = [
            group_uuids[i] for i in selection if 0 <= i < len(group_uuids)
        ]

    def delete_object(self, uuid_str: str) -> bool:
        """Delete object by UUID."""
        if uuid_str in self.signals:
            del self.signals[uuid_str]
            return True
        if uuid_str in self.images:
            del self.images[uuid_str]
            return True
        return False

    def duplicate_object(self, uuid_str: str) -> str | None:
        """Duplicate object and return new UUID."""
        obj = self.signals.get(uuid_str) or self.images.get(uuid_str)
        if obj is None:
            return None

        # Create a copy using serialization/deserialization
        json_data = utils.dataset_to_rpcjson(obj)
        new_obj: SignalObj | ImageObj = utils.rpcjson_to_dataset(json_data)
        obj_uuid = str(uuid.uuid4())
        new_obj.title = f"{obj.title} (copy)"

        # Store the copy
        if isinstance(obj, SignalObj):
            self.signals[obj_uuid] = new_obj
        else:
            self.images[obj_uuid] = new_obj
        return obj_uuid

    def copy_metadata(self, src_uuid: str, dst_uuid: str) -> bool:
        """Copy metadata from source to destination object."""
        src_obj = self.signals.get(src_uuid) or self.images.get(src_uuid)
        dst_obj = self.signals.get(dst_uuid) or self.images.get(dst_uuid)

        if src_obj is None or dst_obj is None:
            return False

        dst_obj.metadata.update(src_obj.metadata)
        return True

    # Group operations
    # pylint: disable=unused-argument
    def add_group(
        self, title: str, panel: str | None = None, select: bool = False
    ) -> str:
        """Add group and return UUID."""
        panel = panel or self.current_panel
        group = MockGroup(title)

        if panel == "signal":
            self.signal_groups[group.uuid] = group
        elif panel == "image":
            self.image_groups[group.uuid] = group

        return group.uuid

    def get_group_titles_with_object_info(
        self,
    ) -> tuple[list[str], list[list[str]], list[list[str]]]:
        """Return groups titles and lists of inner objects uuids and titles."""
        panel = self.current_panel
        if panel == "signal":
            groups = self.signal_groups
            objects = self.signals
        elif panel == "image":
            groups = self.image_groups
            objects = self.images
        else:
            return ([], [], [])

        group_titles = []
        group_uuids_lists = []
        group_titles_lists = []

        for group in groups.values():
            group_titles.append(group.title)

            # Get objects in this group
            object_uuids = [uuid for uuid in group.objects if uuid in objects]
            object_titles = [
                objects[uuid].title for uuid in object_uuids if uuid in objects
            ]

            group_uuids_lists.append(object_uuids)
            group_titles_lists.append(object_titles)

        return (group_titles, group_uuids_lists, group_titles_lists)

    def get_group_titles(self, panel: str | None = None) -> list[str]:
        """Get group titles."""
        panel = panel or self.current_panel
        if panel == "signal":
            return [group.title for group in self.signal_groups.values()]
        if panel == "image":
            return [group.title for group in self.image_groups.values()]
        return []

    def move_up(self, uuid_str: str) -> bool:  # pylint: disable=unused-argument
        """Move object up in list."""
        # In stub mode, just return success
        return True

    def move_down(self, uuid_str: str) -> bool:  # pylint: disable=unused-argument
        """Move object down in list."""
        # In stub mode, just return success
        return True

    def delete_group(self, group_id: str, panel: str | None = None) -> bool:
        """Delete group."""
        panel = panel or self.current_panel
        if panel == "signal" and group_id in self.signal_groups:
            del self.signal_groups[group_id]
            return True
        if panel == "image" and group_id in self.image_groups:
            del self.image_groups[group_id]
            return True
        return False

    # Macro operations (stub implementations)
    def import_macro_from_file(self, filename: str) -> None:
        """Import macro from file (stub implementation).

        Args:
            filename: Filename
        """
        if self.verbose:
            print(f"import_macro_from_file called: {filename} (stub - not implemented)")

    def run_macro(self, number_or_title: int | str | None = None) -> None:
        """Run macro (stub implementation).

        Args:
            number_or_title: Number or title of the macro. Defaults to None.
        """
        if self.verbose:
            print(f"run_macro called: {number_or_title} (stub - not implemented)")

    def stop_macro(self, number_or_title: int | str | None = None) -> None:
        """Stop macro (stub implementation).

        Args:
            number_or_title: Number or title of the macro. Defaults to None.
        """
        if self.verbose:
            print(f"stop_macro called: {number_or_title} (stub - not implemented)")

    # Calculation operations
    def calc(self, name: str, param: list[str] | None = None) -> str | None:
        """Execute calculation and return result object UUID."""
        if self.verbose:
            execenv.print(
                f"[STUB] Simulating calculation '{name}' with params: {param}"
            )

        # In stub mode, just simulate by creating a dummy result object
        if not self.selected_objects:
            if self.verbose:
                execenv.print("[STUB] No objects selected for calculation")
            return None

        src_uuid = self.selected_objects[0]
        src_obj = self.signals.get(src_uuid) or self.images.get(src_uuid)

        if src_obj is None:
            if self.verbose:
                execenv.print(f"[STUB] Source object {src_uuid} not found")
            return None

        # Create a dummy result object based on source type
        if isinstance(src_obj, SignalObj):
            result = create_signal(f"{name}({src_obj.title})")
            obj_uuid = str(uuid.uuid4())
            self.signals[obj_uuid] = result
            if self.verbose:
                execenv.print(f"[STUB] Created dummy signal result: {result.title}")
            return obj_uuid
        if isinstance(src_obj, ImageObj):
            result = create_image(f"{name}({src_obj.title})")
            obj_uuid = str(uuid.uuid4())
            self.images[obj_uuid] = result
            if self.verbose:
                execenv.print(f"[STUB] Created dummy image result: {result.title}")
            return obj_uuid

        if self.verbose:
            execenv.print("[STUB] Unsupported object type for calculation")
        return None

    # Annotation operations
    # pylint: disable=unused-argument
    def add_annotations_from_items(
        self, items: list[str], refresh_plot: bool = True, panel: str | None = None
    ) -> None:
        """Add object annotations (annotation plot items)."""
        # In stub mode, just acknowledge the annotations
        # Real implementation would deserialize items and add to objects

    # pylint: disable=unused-argument
    def add_label_with_title(
        self, title: str | None = None, panel: str | None = None
    ) -> None:
        """Add label with title."""
        if self.verbose:
            execenv.print(f"[STUB] Simulating add label with title: {title}")
        # In stub mode, just acknowledge the label


@contextmanager
def datalab_stub_server(
    port: int = 0, verbose: bool = True
) -> Generator[int, None, None]:
    """Context manager for DataLab stub server.

    Args:
        port: Port to bind to. If 0, uses a random available port.
        verbose: If True, print verbose debug information.

    Yields:
        Port number the server is listening on
    """
    server = DataLabStubServer(port, verbose=verbose)
    try:
        actual_port = server.start()
        yield actual_port
    finally:
        server.stop()


def patch_simpleremoteproxy_for_stub() -> DataLabStubServer:
    """Patch SimpleRemoteProxy to connect to a stub server instead of real DataLab.

    This utility function:
    1. Creates and starts a DataLabStubServer instance
    2. Patches SimpleRemoteProxy.__connect_to_server to use the stub server port
    3. Returns the stub server instance for later cleanup

    Returns:
        The running DataLabStubServer instance that needs to be stopped later

    Example:
        >>> stub_server = patch_simpleremoteproxy_for_stub()
        >>> try:
        ...     # Your code using SimpleRemoteProxy here
        ...     proxy = SimpleRemoteProxy()
        ...     # proxy will connect to stub server automatically
        ... finally:
        ...     stub_server.stop()
    """
    # pylint: disable=import-outside-toplevel
    # Import directly from remote module to avoid circular dependency
    from sigima.client.remote import SimpleRemoteProxy

    # Store original method
    # pylint: disable=protected-access
    original_connect_to_server = SimpleRemoteProxy._SimpleRemoteProxy__connect_to_server

    # Start stub server
    stub_server_instance = DataLabStubServer(verbose=False)
    stub_port = stub_server_instance.start()

    # pylint: disable=unused-argument
    def patched_connect_to_server(self, port=None):
        """Patched connect that uses stub server port."""
        # Always use the stub server port, ignore the requested port
        original_connect_to_server(self, port=str(stub_port))

    # Apply the patch
    # pylint: disable=protected-access
    SimpleRemoteProxy._SimpleRemoteProxy__connect_to_server = patched_connect_to_server

    return stub_server_instance