Watch feature and tests

jupyter_builder/extension_commands/watch.py

@@ -7,6 +7,7 @@
 
 from ..base_extension_app import BaseExtensionApp
 from ..federated_extensions import watch_labextension
+from ..core_path import default_core_path
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 
@@ -19,7 +20,8 @@ class WatchLabExtensionApp(BaseExtensionApp):
     source_map = Bool(False, config=True, help="Generate source maps")
 
     core_path = Unicode(
-        os.path.join(HERE, "staging"),
+        # os.path.join(HERE, "staging"),
+        default_core_path(),
         config=True,
         help="Directory containing core application package.json file",
     )

jupyter_builder/federated_extensions.py

@@ -227,7 +227,8 @@ def watch_labextension(
     path, labextensions_path, logger=None, development=False, source_map=False, core_path=None
 ):
     """Watch a labextension in a given path"""
-    core_path = osp.join(HERE, "staging") if core_path is None else str(Path(core_path).resolve())
+    # core_path = osp.join(HERE, "staging") if core_path is None else str(Path(core_path).resolve())
+    core_path = default_core_path() if core_path is None else str(Path(core_path).resolve())
     ext_path = str(Path(path).resolve())
 
     if logger:

jupyter_builder/jlpm.py

@@ -0,0 +1,43 @@
+"""A Jupyter-aware wrapper for the yarn package manager"""
+
+import os
+
+# Copyright (c) Jupyter Development Team.
+# Distributed under the terms of the Modified BSD License.
+import sys
+
+from jupyterlab_server.process import subprocess, which
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+YARN_PATH = os.path.join(HERE, "staging", "yarn.js")
+
+
+def execvp(cmd, argv):
+    """Execvp, except on Windows where it uses Popen.
+
+    The first argument, by convention, should point to the filename
+    associated with the file being executed.
+
+    Python provides execvp on Windows, but its behavior is problematic
+    (Python bug#9148).
+    """
+    cmd = which(cmd)
+    if os.name == "nt":
+        import signal
+        import sys
+
+        p = subprocess.Popen([cmd] + argv[1:])
+        # Don't raise KeyboardInterrupt in the parent process.
+        # Set this after spawning, to avoid subprocess inheriting handler.
+        signal.signal(signal.SIGINT, signal.SIG_IGN)
+        p.wait()
+        sys.exit(p.returncode)
+    else:
+        os.execvp(cmd, argv)  # noqa S606
+
+
+def main(argv=None):
+    """Run node and return the result."""
+    # Make sure node is available.
+    argv = argv or sys.argv[1:]
+    execvp("node", ["node", YARN_PATH, *argv])

jupyter_builder/main.py

@@ -10,8 +10,7 @@
 
 from jupyter_builder.extension_commands.build import BuildLabExtensionApp
 from jupyter_builder.extension_commands.develop import DevelopLabExtensionApp
-
-# from .commands.watch import WatchLabExtensionApp
+from jupyter_builder.extension_commands.watch import WatchLabExtensionApp
 
 
 class LabExtensionApp(JupyterApp):
@@ -36,7 +35,7 @@ class LabExtensionApp(JupyterApp):
         # "check": (CheckLabExtensionsApp, "Check labextension(s)"),
         "develop": (DevelopLabExtensionApp, "(developer) Develop labextension(s)"),
         "build": (BuildLabExtensionApp, "(developer) Build labextension"),
-        # "watch": (WatchLabExtensionApp, "(developer) Watch labextension"),
+        "watch": (WatchLabExtensionApp, "(developer) Watch labextension"),
     }
 
     def start(self):

tests/test_tpl.py

@@ -2,7 +2,11 @@
 # Distributed under the terms of the Modified BSD License.
 
 import os
-from subprocess import run
+import platform
+import signal
+import subprocess
+import time
+from subprocess import run, Popen
 from pathlib import Path
 
 
@@ -27,6 +31,7 @@ def helper(dest):
     log.touch()
 
 
+# ---------------------- BUILD TESTS --------------------------------------
 def test_files_build(tmp_path):
     extension_folder = tmp_path / "ext"
     extension_folder.mkdir()
@@ -81,3 +86,83 @@ def test_files_build_development(tmp_path):
     for filename in expected_files:
         filepath = os.path.join(folder_path, filename)
         assert os.path.exists(filepath), f"File {filename} does not exist in {folder_path}!"
+
+
+# --------------------------------- WATCH TESTS ---------------------------------------
+
+
+def list_files_in_static(directory):
+    """List all filenames in the specified directory."""
+    return {f.name for f in Path(directory).glob("*")}
+
+
+def test_watch_functionality(tmp_path):
+    extension_folder = tmp_path / "ext"
+    extension_folder.mkdir()
+    helper(str(extension_folder))
+
+    env = os.environ.copy()
+    env.update({"YARN_ENABLE_IMMUTABLE_INSTALLS": "false"})
+    run(
+        ["jlpm", "install"],
+        cwd=extension_folder,
+        check=True,
+        env=env,
+    )
+    run(["jlpm", "run", "build"], cwd=extension_folder, check=True)
+
+    # Path to the TypeScript file to change
+    index_ts_path = extension_folder / "src/index.ts"
+
+    static_dir = extension_folder / "myextension/labextension/static"
+
+    # Ensure the TypeScript file exists
+    assert index_ts_path.exists(), f"File {index_ts_path} does not exist!"
+
+    # List filenames in static directory before change
+    initial_files = list_files_in_static(static_dir)
+
+    is_windows = platform.system() == "Windows"
+    if is_windows:
+        kwargs = {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP}
+    else:
+        kwargs = {}
+
+    watch_process = Popen(
+        ["jupyter-builder", "watch", str(extension_folder)], cwd=extension_folder, **kwargs
+    )
+
+    # This sleep time makes sure that the comment is added only after the watch process is running.
+    time.sleep(20)
+
+    try:
+        # Add a comment to the TypeScript file to trigger watch
+        with index_ts_path.open("a") as f:
+            f.write("// Test comment to trigger watch\n")
+
+        # Wait for watch process to detect change and rebuild
+        time.sleep(15)  # Adjust this time if needed
+
+        # List filenames in static directory after change
+        final_files = list_files_in_static(static_dir)
+
+        # Compare the initial and final file lists
+        assert initial_files != final_files, (
+            " No changes detected in the static directory."
+            "Watch process may not have triggered correctly!"
+        )
+
+    finally:
+        watch_process.terminate()
+        # Give some time to terminate the process cleanly
+        time.sleep(5)
+        if watch_process.poll() is None:
+            watch_process.kill()
+
+    # Note: The ideal process of termination is given below, but does not work
+    #     if is_windows:
+    #         watch_process.send_signal(signal.CTRL_C_EVENT)
+
+    #     else:
+    #         watch_process.send_signal(signal.SIGINT)
+    #     watch_process.wait()