mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-03-04 16:02:51 +01:00
fix(editors): VSCode widget removed
This commit is contained in:
@@ -56,7 +56,6 @@ _Widgets = {
|
||||
"ScatterWaveform": "ScatterWaveform",
|
||||
"SignalLabel": "SignalLabel",
|
||||
"TextBox": "TextBox",
|
||||
"VSCodeEditor": "VSCodeEditor",
|
||||
"Waveform": "Waveform",
|
||||
"WebConsole": "WebConsole",
|
||||
"WebsiteWidget": "WebsiteWidget",
|
||||
@@ -5529,12 +5528,6 @@ class TextBox(RPCBase):
|
||||
"""
|
||||
|
||||
|
||||
class VSCodeEditor(RPCBase):
|
||||
"""A widget to display the VSCode editor."""
|
||||
|
||||
...
|
||||
|
||||
|
||||
class Waveform(RPCBase):
|
||||
"""Widget for plotting waveforms."""
|
||||
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
def main(): # pragma: no cover
|
||||
from qtpy import PYSIDE6
|
||||
|
||||
if not PYSIDE6:
|
||||
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
|
||||
return
|
||||
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
|
||||
|
||||
from bec_widgets.widgets.editors.vscode.vs_code_editor_plugin import VSCodeEditorPlugin
|
||||
|
||||
QPyDesignerCustomWidgetCollection.addCustomWidget(VSCodeEditorPlugin())
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
@@ -1 +0,0 @@
|
||||
{'files': ['vscode.py']}
|
||||
@@ -1,57 +0,0 @@
|
||||
# Copyright (C) 2022 The Qt Company Ltd.
|
||||
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
|
||||
|
||||
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
|
||||
from qtpy.QtWidgets import QWidget
|
||||
|
||||
from bec_widgets.utils.bec_designer import designer_material_icon
|
||||
from bec_widgets.widgets.editors.vscode.vscode import VSCodeEditor
|
||||
|
||||
DOM_XML = """
|
||||
<ui language='c++'>
|
||||
<widget class='VSCodeEditor' name='vs_code_editor'>
|
||||
</widget>
|
||||
</ui>
|
||||
"""
|
||||
|
||||
|
||||
class VSCodeEditorPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._form_editor = None
|
||||
|
||||
def createWidget(self, parent):
|
||||
if parent is None:
|
||||
return QWidget()
|
||||
t = VSCodeEditor(parent)
|
||||
return t
|
||||
|
||||
def domXml(self):
|
||||
return DOM_XML
|
||||
|
||||
def group(self):
|
||||
return "BEC Developer"
|
||||
|
||||
def icon(self):
|
||||
return designer_material_icon(VSCodeEditor.ICON_NAME)
|
||||
|
||||
def includeFile(self):
|
||||
return "vs_code_editor"
|
||||
|
||||
def initialize(self, form_editor):
|
||||
self._form_editor = form_editor
|
||||
|
||||
def isContainer(self):
|
||||
return False
|
||||
|
||||
def isInitialized(self):
|
||||
return self._form_editor is not None
|
||||
|
||||
def name(self):
|
||||
return "VSCodeEditor"
|
||||
|
||||
def toolTip(self):
|
||||
return ""
|
||||
|
||||
def whatsThis(self):
|
||||
return self.toolTip()
|
||||
@@ -1,203 +0,0 @@
|
||||
import os
|
||||
import select
|
||||
import shlex
|
||||
import signal
|
||||
import socket
|
||||
import subprocess
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel
|
||||
from qtpy.QtCore import Signal, Slot
|
||||
|
||||
from bec_widgets.widgets.editors.website.website import WebsiteWidget
|
||||
|
||||
|
||||
class VSCodeInstructionMessage(BaseModel):
|
||||
command: Literal["open", "write", "close", "zenMode", "save", "new", "setCursor"]
|
||||
content: str = ""
|
||||
|
||||
|
||||
def get_free_port():
|
||||
"""
|
||||
Get a free port on the local machine.
|
||||
|
||||
Returns:
|
||||
int: The free port number
|
||||
"""
|
||||
sock = socket.socket()
|
||||
sock.bind(("", 0))
|
||||
port = sock.getsockname()[1]
|
||||
sock.close()
|
||||
return port
|
||||
|
||||
|
||||
class VSCodeEditor(WebsiteWidget):
|
||||
"""
|
||||
A widget to display the VSCode editor.
|
||||
"""
|
||||
|
||||
file_saved = Signal(str)
|
||||
|
||||
token = "bec"
|
||||
host = "127.0.0.1"
|
||||
|
||||
PLUGIN = True
|
||||
USER_ACCESS = []
|
||||
ICON_NAME = "developer_mode_tv"
|
||||
|
||||
def __init__(self, parent=None, config=None, client=None, gui_id=None, **kwargs):
|
||||
|
||||
self.process = None
|
||||
self.port = get_free_port()
|
||||
self._url = f"http://{self.host}:{self.port}?tkn={self.token}"
|
||||
super().__init__(parent=parent, config=config, client=client, gui_id=gui_id, **kwargs)
|
||||
self.start_server()
|
||||
self.bec_dispatcher.connect_slot(self.on_vscode_event, f"vscode-events/{self.gui_id}")
|
||||
|
||||
def start_server(self):
|
||||
"""
|
||||
Start the server.
|
||||
|
||||
This method starts the server for the VSCode editor in a subprocess.
|
||||
"""
|
||||
|
||||
env = os.environ.copy()
|
||||
env["BEC_Widgets_GUIID"] = self.gui_id
|
||||
env["BEC_REDIS_HOST"] = self.client.connector.host
|
||||
cmd = shlex.split(
|
||||
f"code serve-web --port {self.port} --connection-token={self.token} --accept-server-license-terms"
|
||||
)
|
||||
self.process = subprocess.Popen(
|
||||
cmd,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL,
|
||||
preexec_fn=os.setsid,
|
||||
env=env,
|
||||
)
|
||||
|
||||
os.set_blocking(self.process.stdout.fileno(), False)
|
||||
while self.process.poll() is None:
|
||||
readylist, _, _ = select.select([self.process.stdout], [], [], 1)
|
||||
if self.process.stdout in readylist:
|
||||
output = self.process.stdout.read(1024)
|
||||
if output and f"available at {self._url}" in output:
|
||||
break
|
||||
self.set_url(self._url)
|
||||
self.wait_until_loaded()
|
||||
|
||||
@Slot(str)
|
||||
def open_file(self, file_path: str):
|
||||
"""
|
||||
Open a file in the VSCode editor.
|
||||
|
||||
Args:
|
||||
file_path: The file path to open
|
||||
"""
|
||||
msg = VSCodeInstructionMessage(command="open", content=f"file://{file_path}")
|
||||
self.client.connector.raw_send(f"vscode-instructions/{self.gui_id}", msg.model_dump_json())
|
||||
|
||||
@Slot(dict, dict)
|
||||
def on_vscode_event(self, content, _metadata):
|
||||
"""
|
||||
Handle the VSCode event. VSCode events are received as RawMessages.
|
||||
|
||||
Args:
|
||||
content: The content of the event
|
||||
metadata: The metadata of the event
|
||||
"""
|
||||
|
||||
# the message also contains the content but I think is fine for now to just emit the file path
|
||||
if not isinstance(content["data"], dict):
|
||||
return
|
||||
if "uri" not in content["data"]:
|
||||
return
|
||||
if not content["data"]["uri"].startswith("file://"):
|
||||
return
|
||||
file_path = content["data"]["uri"].split("file://")[1]
|
||||
self.file_saved.emit(file_path)
|
||||
|
||||
@Slot()
|
||||
def save_file(self):
|
||||
"""
|
||||
Save the file in the VSCode editor.
|
||||
"""
|
||||
msg = VSCodeInstructionMessage(command="save")
|
||||
self.client.connector.raw_send(f"vscode-instructions/{self.gui_id}", msg.model_dump_json())
|
||||
|
||||
@Slot()
|
||||
def new_file(self):
|
||||
"""
|
||||
Create a new file in the VSCode editor.
|
||||
"""
|
||||
msg = VSCodeInstructionMessage(command="new")
|
||||
self.client.connector.raw_send(f"vscode-instructions/{self.gui_id}", msg.model_dump_json())
|
||||
|
||||
@Slot()
|
||||
def close_file(self):
|
||||
"""
|
||||
Close the file in the VSCode editor.
|
||||
"""
|
||||
msg = VSCodeInstructionMessage(command="close")
|
||||
self.client.connector.raw_send(f"vscode-instructions/{self.gui_id}", msg.model_dump_json())
|
||||
|
||||
@Slot(str)
|
||||
def write_file(self, content: str):
|
||||
"""
|
||||
Write content to the file in the VSCode editor.
|
||||
|
||||
Args:
|
||||
content: The content to write
|
||||
"""
|
||||
msg = VSCodeInstructionMessage(command="write", content=content)
|
||||
self.client.connector.raw_send(f"vscode-instructions/{self.gui_id}", msg.model_dump_json())
|
||||
|
||||
@Slot()
|
||||
def zen_mode(self):
|
||||
"""
|
||||
Toggle the Zen mode in the VSCode editor.
|
||||
"""
|
||||
msg = VSCodeInstructionMessage(command="zenMode")
|
||||
self.client.connector.raw_send(f"vscode-instructions/{self.gui_id}", msg.model_dump_json())
|
||||
|
||||
@Slot(int, int)
|
||||
def set_cursor(self, line: int, column: int):
|
||||
"""
|
||||
Set the cursor in the VSCode editor.
|
||||
|
||||
Args:
|
||||
line: The line number
|
||||
column: The column number
|
||||
"""
|
||||
msg = VSCodeInstructionMessage(command="setCursor", content=f"{line},{column}")
|
||||
self.client.connector.raw_send(f"vscode-instructions/{self.gui_id}", msg.model_dump_json())
|
||||
|
||||
def cleanup_vscode(self):
|
||||
"""
|
||||
Cleanup the VSCode editor.
|
||||
"""
|
||||
if not self.process or self.process.poll() is not None:
|
||||
return
|
||||
os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
|
||||
self.process.wait()
|
||||
|
||||
def cleanup(self):
|
||||
"""
|
||||
Cleanup the widget. This method is called from the dock area when the widget is removed.
|
||||
"""
|
||||
self.bec_dispatcher.disconnect_slot(self.on_vscode_event, f"vscode-events/{self.gui_id}")
|
||||
self.cleanup_vscode()
|
||||
return super().cleanup()
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
import sys
|
||||
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
widget = VSCodeEditor(gui_id="unknown")
|
||||
widget.show()
|
||||
app.exec_()
|
||||
widget.bec_dispatcher.disconnect_all()
|
||||
widget.client.shutdown()
|
||||
@@ -1,91 +0,0 @@
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from bec_widgets.widgets.editors.vscode.vscode import VSCodeEditor
|
||||
|
||||
from .client_mocks import mocked_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def vscode_widget(qtbot, mocked_client):
|
||||
with mock.patch("bec_widgets.widgets.editors.vscode.vscode.subprocess.Popen") as mock_popen:
|
||||
widget = VSCodeEditor(client=mocked_client)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
|
||||
|
||||
def test_vscode_widget(qtbot, vscode_widget):
|
||||
assert vscode_widget.process is not None
|
||||
assert vscode_widget._url == f"http://127.0.0.1:{vscode_widget.port}?tkn=bec"
|
||||
|
||||
|
||||
def test_start_server(qtbot, mocked_client):
|
||||
with mock.patch("bec_widgets.widgets.editors.vscode.vscode.os.killpg") as mock_killpg:
|
||||
with mock.patch("bec_widgets.widgets.editors.vscode.vscode.os.getpgid") as mock_getpgid:
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.editors.vscode.vscode.subprocess.Popen"
|
||||
) as mock_popen:
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.editors.vscode.vscode.select.select"
|
||||
) as mock_select:
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.editors.vscode.vscode.get_free_port"
|
||||
) as mock_get_free_port:
|
||||
mock_get_free_port.return_value = 12345
|
||||
mock_process = mock.Mock()
|
||||
mock_process.stdout.fileno.return_value = 1
|
||||
mock_process.poll.return_value = None
|
||||
mock_process.stdout.read.return_value = f"available at http://{VSCodeEditor.host}:{12345}?tkn={VSCodeEditor.token}"
|
||||
mock_popen.return_value = mock_process
|
||||
mock_select.return_value = [[mock_process.stdout], [], []]
|
||||
|
||||
widget = VSCodeEditor(client=mocked_client)
|
||||
widget.close()
|
||||
widget.deleteLater()
|
||||
|
||||
assert (
|
||||
mock.call(
|
||||
shlex.split(
|
||||
f"code serve-web --port {widget.port} --connection-token={widget.token} --accept-server-license-terms"
|
||||
),
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL,
|
||||
preexec_fn=os.setsid,
|
||||
env=mock.ANY,
|
||||
)
|
||||
in mock_popen.mock_calls
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patched_vscode_process(qtbot, vscode_widget):
|
||||
with mock.patch("bec_widgets.widgets.editors.vscode.vscode.os.killpg") as mock_killpg:
|
||||
mock_killpg.reset_mock()
|
||||
with mock.patch("bec_widgets.widgets.editors.vscode.vscode.os.getpgid") as mock_getpgid:
|
||||
mock_getpgid.return_value = 123
|
||||
vscode_widget.process = mock.Mock()
|
||||
yield vscode_widget, mock_killpg
|
||||
|
||||
|
||||
def test_vscode_cleanup(qtbot, patched_vscode_process):
|
||||
vscode_patched, mock_killpg = patched_vscode_process
|
||||
vscode_patched.process.pid = 123
|
||||
vscode_patched.process.poll.return_value = None
|
||||
vscode_patched.cleanup_vscode()
|
||||
mock_killpg.assert_called_once_with(123, 15)
|
||||
vscode_patched.process.wait.assert_called_once()
|
||||
|
||||
|
||||
def test_close_event_on_terminated_code(qtbot, patched_vscode_process):
|
||||
vscode_patched, mock_killpg = patched_vscode_process
|
||||
vscode_patched.process.pid = 123
|
||||
vscode_patched.process.poll.return_value = 0
|
||||
vscode_patched.cleanup_vscode()
|
||||
mock_killpg.assert_not_called()
|
||||
vscode_patched.process.wait.assert_not_called()
|
||||
Reference in New Issue
Block a user