diff --git a/.github/workflows/test_local_integration.yaml b/.github/workflows/test_local_integration.yaml index 6d605787a6..b8dec682bf 100644 --- a/.github/workflows/test_local_integration.yaml +++ b/.github/workflows/test_local_integration.yaml @@ -49,7 +49,7 @@ jobs: run: shell: bash -l {0} steps: - - name: 'Checkout Infrastructure' + - name: "Checkout Infrastructure" uses: actions/checkout@main with: fetch-depth: 0 @@ -172,7 +172,7 @@ jobs: # create environment file envsubst < .env.tpl > .env # run playwright pytest tests in headed mode with the chromium browser - xvfb-run pytest --browser chromium + xvfb-run pytest --browser chromium --slowmo 300 --headed - name: Save Cypress screenshots and videos if: always() diff --git a/tests/common/handlers.py b/tests/common/handlers.py new file mode 100644 index 0000000000..51964d3ac5 --- /dev/null +++ b/tests/common/handlers.py @@ -0,0 +1,342 @@ +import logging +import re +import time + +from playwright.sync_api import expect + +logger = logging.getLogger() + + +class JupyterLab: + def __init__(self, navigator): + logger.debug(">>> Starting notebook manager...") + self.nav = navigator + self.page = self.nav.page + + def reset_workspace(self): + """Reset the JupyterLab workspace.""" + logger.debug(">>> Resetting JupyterLab workspace") + + # Check for and handle kernel popup + logger.debug(">>> Checking for kernel popup") + if self._check_for_kernel_popup(): + self._handle_kernel_popup() + + # Shutdown all kernels + logger.debug(">>> Shutting down all kernels") + self._shutdown_all_kernels() + + # Navigate back to root folder and close all tabs + logger.debug(">>> Navigating to root folder and closing all tabs") + self._navigate_to_root_folder() + logger.debug(">>> Closing all tabs") + self._close_all_tabs() + + # Ensure theme and launcher screen + logger.debug(">>> Ensuring theme and launcher screen") + self._assert_theme_and_launcher() + + def set_environment(self, kernel): + """Set environment for a Jupyter notebook.""" + if not self._check_for_kernel_popup(): + self._trigger_kernel_change_popup() + + self._handle_kernel_popup(kernel) + self._wait_for_kernel_label(kernel) + + def write_file(self, filepath, content): + """Write a file to the Nebari instance filesystem.""" + logger.debug(f">>> Writing file to {filepath}") + self._open_terminal() + self._execute_terminal_commands( + [f"cat <{filepath}", content, "EOF", f"ls {filepath}"] + ) + time.sleep(2) + + def _check_for_kernel_popup(self): + """Check if the kernel popup is open.""" + logger.debug(">>> Checking for kernel popup") + self.page.wait_for_load_state() + time.sleep(3) + visible = self.page.get_by_text("Select KernelStart a new").is_visible() + logger.debug(f">>> Kernel popup visible: {visible}") + return visible + + def _handle_kernel_popup(self, kernel=None): + """Handle kernel popup by selecting the appropriate kernel or dismissing the popup.""" + if kernel: + self._select_kernel(kernel) + else: + self._dismiss_kernel_popup() + + def _dismiss_kernel_popup(self): + """Dismiss the kernel selection popup.""" + logger.debug(">>> Dismissing kernel popup") + no_kernel_button = self.page.get_by_role("dialog").get_by_role( + "button", name="No Kernel" + ) + if no_kernel_button.is_visible(): + no_kernel_button.click() + else: + try: + self.page.get_by_role("button", name="Cancel").click() + except Exception: + raise ValueError("Unable to escape kernel selection dialog.") + + def _shutdown_all_kernels(self): + """Shutdown all running kernels.""" + logger.debug(">>> Shutting down all kernels") + kernel_menu = self.page.get_by_role("menuitem", name="Kernel") + kernel_menu.click() + shut_down_all = self.page.get_by_role("menuitem", name="Shut Down All Kernels…") + logger.debug( + f">>> Shut down all kernels visible: {shut_down_all.is_visible()} enabled: {shut_down_all.is_enabled()}" + ) + if shut_down_all.is_visible() and shut_down_all.is_enabled(): + shut_down_all.click() + self.page.get_by_role("button", name="Shut Down All").click() + else: + logger.debug(">>> No kernels to shut down") + + def _navigate_to_root_folder(self): + """Navigate back to the root folder in JupyterLab.""" + logger.debug(">>> Navigating to root folder") + self.page.get_by_title(f"/home/{self.nav.username}", exact=True).locator( + "path" + ).click() + + def _close_all_tabs(self): + """Close all open tabs in JupyterLab.""" + logger.debug(">>> Closing all tabs") + self.page.get_by_text("File", exact=True).click() + self.page.get_by_role("menuitem", name="Close All Tabs", exact=True).click() + + if self.page.get_by_text("Save your work", exact=True).is_visible(): + self.page.get_by_role( + "button", name="Discard changes to file", exact=True + ).click() + + def _assert_theme_and_launcher(self): + """Ensure that the theme is set to JupyterLab Dark and Launcher screen is visible.""" + expect( + self.page.get_by_text( + "Set Preferred Dark Theme: JupyterLab Dark", exact=True + ) + ).to_be_hidden() + self.page.get_by_title("VS Code [↗]").wait_for(state="visible") + + def _open_terminal(self): + """Open a new terminal in JupyterLab.""" + self.page.get_by_text("File", exact=True).click() + self.page.get_by_text("New", exact=True).click() + self.page.get_by_role("menuitem", name="Terminal").get_by_text( + "Terminal" + ).click() + + def _execute_terminal_commands(self, commands): + """Execute a series of commands in the terminal.""" + for command in commands: + self.page.get_by_role("textbox", name="Terminal input").fill(command) + self.page.get_by_role("textbox", name="Terminal input").press("Enter") + time.sleep(0.5) + + +class Notebook(JupyterLab): + def __init__(self, navigator): + logger.debug(">>> Starting notebook manager...") + self.nav = navigator + self.page = self.nav.page + + def _open_notebook(self, notebook_name): + """Open a notebook in JupyterLab.""" + self.page.get_by_text("File", exact=True).click() + self.page.locator("#jp-mainmenu-file").get_by_text("Open from Path…").click() + + expect(self.page.get_by_text("Open PathPathCancelOpen")).to_be_visible() + + # Fill notebook name into the textbox and click Open + self.page.get_by_placeholder("/path/relative/to/jlab/root").fill(notebook_name) + self.page.get_by_role("button", name="Open").click() + if self.page.get_by_text("Could not find path:").is_visible(): + self.page.get_by_role("button", name="Dismiss").click() + raise ValueError(f"Notebook {notebook_name} not found") + + # make sure that this notebook is one currently selected + expect(self.page.get_by_role("tab", name=notebook_name)).to_be_visible() + + def _run_all_cells(self): + """Run all cells in a Jupyter notebook.""" + self.page.get_by_role("menuitem", name="Run").click() + run_all_cells = self.page.locator("#jp-mainmenu-run").get_by_text( + "Run All Cells", exact=True + ) + if run_all_cells.is_visible(): + run_all_cells.click() + else: + self.page.get_by_text("Restart the kernel and run").click() + # Check if restart popup is visible + restart_popup = self.page.get_by_text("Restart Kernel?") + if restart_popup.is_visible(): + restart_popup.click() + self.page.get_by_role("button", name="Confirm Kernel Restart").click() + + def _wait_for_commands_completion( + self, timeout: float, completion_wait_time: float + ): + """ + Wait for commands to finish running + + Parameters + ---------- + timeout: float + Time in seconds to wait for the expected output text to appear. + completion_wait_time: float + Time in seconds to wait between checking for expected output text. + """ + elapsed_time = 0.0 + still_visible = True + start_time = time.time() + while elapsed_time < timeout: + running = self.nav.page.get_by_text("[*]").all() + still_visible = any(list(map(lambda r: r.is_visible(), running))) + if not still_visible: + break + elapsed_time = time.time() - start_time + time.sleep(completion_wait_time) + if still_visible: + raise ValueError( + f"Timeout Waited for commands to finish, " + f"but couldn't finish in {timeout} sec" + ) + + def _get_outputs(self): + output_elements = self.nav.page.query_selector_all(".jp-OutputArea-output") + text_content = [element.text_content().strip() for element in output_elements] + return text_content + + def run_notebook(self, notebook_name, kernel): + """Run a notebook in JupyterLab.""" + # Open the notebook + logger.debug(f">>> Opening notebook: {notebook_name}") + self._open_notebook(notebook_name) + + # Set environment + logger.debug(f">>> Setting environment for kernel: {kernel}") + self.set_environment(kernel=kernel) + + # Run all cells + logger.debug(">>> Running all cells") + self._run_all_cells() + + # Wait for commands to finish running + logger.debug(">>> Waiting for commands to finish running") + self._wait_for_commands_completion(timeout=300, completion_wait_time=5) + + # Get the outputs + logger.debug(">>> Gathering outputs") + outputs = self._get_outputs() + + return outputs + + def _trigger_kernel_change_popup(self): + """Trigger the kernel change popup. (expects a notebook to be open)""" + self.page.get_by_role("menuitem", name="Kernel").click() + kernel_menu = self.page.get_by_role("menuitem", name="Change Kernel…") + if kernel_menu.is_visible(): + kernel_menu.click() + self.page.get_by_text("Select KernelStart a new").wait_for(state="visible") + logger.debug(">>> Kernel popup is visible") + else: + pass + + def _select_kernel(self, kernel): + """Select a kernel from the popup.""" + logger.debug(f">>> Selecting kernel: {kernel}") + + self.page.get_by_role("dialog").get_by_label("", exact=True).fill(kernel) + + # List of potential selectors + selectors = [ + self.page.get_by_role("cell", name=re.compile(kernel, re.IGNORECASE)).nth( + 1 + ), + self.page.get_by_role("cell", name=re.compile(kernel, re.IGNORECASE)).first, + self.page.get_by_text(kernel, exact=True).nth(1), + ] + + # Try each selector until one is visible and clickable + # this is done due to the different ways the kernel can be displayed + # as part of the new extension + for selector in selectors: + if selector.is_visible(): + selector.click() + logger.debug(f">>> Kernel {kernel} selected") + return + + # If none of the selectors match, dismiss the popup and raise an error + self._dismiss_kernel_popup() + raise ValueError(f"Kernel {kernel} not found in the list of kernels") + + def _wait_for_kernel_label(self, kernel): + """Wait for the kernel label to be visible.""" + kernel_label_loc = self.page.get_by_role("button", name=kernel) + if not kernel_label_loc.is_visible(): + kernel_label_loc.wait_for(state="attached") + logger.debug(f">>> Kernel label {kernel} is now visible") + + +class CondaStore(JupyterLab): + def __init__(self, navigator): + self.page = navigator.page + self.nav = navigator + + def _open_conda_store_service(self): + self.page.get_by_text("Services", exact=True).click() + self.page.get_by_text("Environment Management").click() + expect(self.page.get_by_role("tab", name="conda-store")).to_be_visible() + time.sleep(2) + + def _open_new_environment_tab(self): + self.page.get_by_label("Create a new environment in").click() + expect(self.page.get_by_text("Create Environment")).to_be_visible() + + def _assert_user_namespace(self): + expect( + self.page.get_by_role("button", name=f"{self.nav.username} Create a new") + ).to_be_visible() + + def _get_shown_namespaces(self): + _envs = self.page.locator("#environmentsScroll").get_by_role("button") + _env_contents = [env.text_content() for env in _envs.all()] + # Remove the "New" entry from each namespace "button" text + return [ + namespace.replace(" New", "") + for namespace in _env_contents + if namespace != " New" + ] + + def _assert_logged_in(self): + login_button = self.page.get_by_role("button", name="Log in") + if login_button.is_visible(): + login_button.click() + # wait for page to reload + self.page.wait_for_load_state() + time.sleep(2) + # A reload is required as conda-store "created" a new page once logged in + self.page.reload() + self.page.wait_for_load_state() + self._open_conda_store_service() + else: + # In this case logout should already be visible + expect(self.page.get_by_role("button", name="Logout")).to_be_visible() + self._assert_user_namespace() + + def conda_store_ui(self): + logger.debug(">>> Opening Conda Store UI") + self._open_conda_store_service() + + logger.debug(">>> Assert user is logged in") + self._assert_logged_in() + + logger.debug(">>> Opening new environment tab") + self._open_new_environment_tab() diff --git a/tests/common/navigator.py b/tests/common/navigator.py index f846d9a545..04e019a7a6 100644 --- a/tests/common/navigator.py +++ b/tests/common/navigator.py @@ -1,125 +1,48 @@ -import contextlib -import datetime as dt import logging import re -import time import urllib +from abc import ABC +from pathlib import Path from playwright.sync_api import expect, sync_playwright logger = logging.getLogger() -class Navigator: - """Base class for Nebari Playwright testing. This provides setup and - teardown methods that all tests will need and some other generally useful - methods such as clearing the workspace. Specific tests such has "Run a - notebook" are included in separate classes which use an instance of - this class. - - The Navigator class and the associated test classes are design to be able - to run either standalone, or inside of pytest. This makes it easy to - develop new tests, but also have them fully prepared to be - included as part of the test suite. - - Parameters - ---------- - nebari_url: str - Nebari URL to access for testing, e.g. "https://{nebari_url} - username: str - Login username for Nebari. For Google login, this will be email address. - password: str - Login password for Nebari. For Google login, this will be the Google - password. - auth: str - Authentication type of this Nebari instance. Options are "google" and - "password". - headless: bool - (Optional) Run the tests in headless mode (without visuals). Defaults - to False. - slow_mo: int - (Optional) Additional milliseconds to add to each Playwright command, - creating the effect of running the tests in slow motion so they are - easier for humans to follow. Defaults to 0. - browser: str - (Optional) Browser on which to run tests. Options are "chromium", - "webkit", and "firefox". Defaults to "chromium". - instance_name: str - (Optional) Server instance type on which to run tests. Options are - based on the configuration of the Nebari instance. Defaults to - "small-instance". Note that special characters (such as parenthesis) - will need to be converted to dashes. Check the HTML element to get the - exact structure. - video_dir: None or str - (Optional) Directory in which to save videos. If None, no video will - be saved. Defaults to None. +class NavigatorMixin(ABC): + """ + A mixin class providing common setup and teardown functionalities for Playwright navigators. """ def __init__( self, - nebari_url, - username, - password, - auth, headless=False, slow_mo=0, browser="chromium", - instance_name="small-instance", video_dir=None, + video_name_prefix=None, ): - self.nebari_url = nebari_url - self.username = username - self.password = password - self.auth = auth - self.initialized = False self.headless = headless self.slow_mo = slow_mo - self.browser = browser - self.instance_name = instance_name + self.browser_name = browser self.video_dir = video_dir + self.video_name_prefix = video_name_prefix + self.initialized = False + self.setup() - self.setup( - browser=self.browser, - headless=self.headless, - slow_mo=self.slow_mo, - ) - self.wait_for_server_spinup = 300_000 # 5 * 60 * 1_000 # 5 minutes in ms - - @property - def initialize(self): - """Ensure that the Navigator is setup and ready for testing.""" - if not self.initialized: - self.setup( - browser=self.browser, - headless=self.headless, - slow_mo=self.slow_mo, - ) - - def setup(self, browser, headless, slow_mo): - """Initial setup for running playwright. Starts playwright, creates - the browser object, a new browser context, and a new page object. - - Parameters - ---------- - browser: str - Browser on which to run tests. Options are "chromium", - "webkit", and "firefox". - headless: bool - Run the tests in headless mode (without visuals) if True - slow_mo: int - Additional milliseconds to add to each Playwright command, - creating the effect of running the tests in slow motion so they are - easier for humans to follow. - """ + def setup(self): + """Setup Playwright browser and context.""" logger.debug(">>> Setting up browser for Playwright") - self.playwright = sync_playwright().start() try: - self.browser = getattr(self.playwright, browser).launch( - headless=headless, slow_mo=slow_mo + self.browser = getattr(self.playwright, self.browser_name).launch( + headless=self.headless, slow_mo=self.slow_mo ) except AttributeError: - raise RuntimeError(f"{browser} browser is not recognized.") from None + raise RuntimeError( + f"{self.browser_name} browser is not recognized." + ) from None + self.context = self.browser.new_context( ignore_https_errors=True, record_video_dir=self.video_dir, @@ -127,26 +50,65 @@ def setup(self, browser, headless, slow_mo): self.page = self.context.new_page() self.initialized = True + def _rename_test_video_path(self, video_path): + """Rename the test video file to the test unique identifier.""" + video_file_name = ( + f"{self.video_name_prefix}.mp4" if self.video_name_prefix else None + ) + if video_file_name and video_path: + Path.rename(video_path, Path(self.video_dir) / video_file_name) + def teardown(self) -> None: - """Shut down and close playwright. This is important to ensure that - no leftover processes are left running in the background.""" - self.context.close() - self.browser.close() # Make sure to close, so that videos are saved. - self.playwright.stop() - logger.debug(">>> Teardown complete.") - - def login(self) -> None: - """Login to nebari deployment using the auth method on the class.""" + """Teardown Playwright browser and context.""" + if self.initialized: + # Rename the video file to the test unique identifier + current_video_path = self.page.video.path() + self._rename_test_video_path(current_video_path) + + self.context.close() + self.browser.close() + self.playwright.stop() + logger.debug(">>> Teardown complete.") + self.initialized = False + + def __enter__(self): + """Enter the runtime context related to this object.""" + return self + + def __exit__(self, exc_type, exc_value, traceback): + """Exit the runtime context related to this object.""" + self.teardown() + + +class LoginNavigator(NavigatorMixin): + """ + A navigator class to handle login operations for Nebari. + """ + + def __init__(self, nebari_url, username, password, auth="password", **kwargs): + super().__init__(**kwargs) + self.nebari_url = nebari_url + self.username = username + self.password = password + self.auth = auth + + def login(self): + """Login to Nebari deployment using the provided authentication method.""" + login_methods = { + "google": self._login_google, + "password": self._login_password, + } try: - return { - "google": self.login_google, - "password": self.login_password, - }[self.auth]() + login_methods[self.auth]() except KeyError: - raise ValueError(f"Auth type of {self.auth} is invalid.") from None + raise ValueError(f"Auth type {self.auth} is invalid.") - def login_google(self) -> None: - """Go to a nebari deployment, login via Google""" + def logout(self): + """Logout from Nebari deployment.""" + self.page.get_by_role("button", name="Logout").click() + self.page.wait_for_load_state + + def _login_google(self): logger.debug(">>> Sign in via Google and start the server") self.page.goto(self.nebari_url) expect(self.page).to_have_url(re.compile(f"{self.nebari_url}*")) @@ -156,278 +118,81 @@ def login_google(self) -> None: self.page.get_by_role("textbox", name="Email or phone").fill(self.username) self.page.get_by_role("button", name="Next").click() self.page.get_by_role("textbox", name="Enter your password").fill(self.password) - - self.page.wait_for_load_state("networkidle") self.page.get_by_role("button", name="Next").click() - - # let the page load self.page.wait_for_load_state("networkidle") - def login_password(self) -> None: - """Go to a nebari deployment, login via Username/Password, and start - a new server. - """ + def _login_password(self): logger.debug(">>> Sign in via Username/Password") self.page.goto(self.nebari_url) expect(self.page).to_have_url(re.compile(f"{self.nebari_url}*")) self.page.get_by_role("button", name="Sign in with Keycloak").click() self.page.get_by_label("Username").fill(self.username) - self.page.get_by_label("Password").click() self.page.get_by_label("Password").fill(self.password) self.page.get_by_role("button", name="Sign In").click() + self.page.wait_for_load_state() - # let the page load - self.page.wait_for_load_state("networkidle") + # Redirect to hub control panel + self.page.goto(urllib.parse.urljoin(self.nebari_url, "hub/home")) + expect(self.page.get_by_role("button", name="Logout")).to_be_visible() + + +class ServerManager(LoginNavigator): + """ + Manages server operations such as starting and stopping a Nebari server. + """ + + def __init__( + self, instance_name="small-instance", wait_for_server_spinup=300_000, **kwargs + ): + super().__init__(**kwargs) + self.instance_name = instance_name + self.wait_for_server_spinup = wait_for_server_spinup + + def start_server(self): + """Start a Nebari server, handling different UI states.""" + self.login() - def start_server(self) -> None: - """Start a nebari server. There are several different web interfaces - possible in this process depending on if you already have a server - running or not. In order for this to work, wait for the page to load, - we look for html elements that exist when no server is running, if - they aren't visible, we check for an existing server start option. - """ - # wait for the page to load logout_button = self.page.get_by_text("Logout", exact=True) logout_button.wait_for(state="attached", timeout=90000) - # if the server is already running start_locator = self.page.get_by_role("button", name="My Server", exact=True) if start_locator.is_visible(): start_locator.click() - - # if server is not yet running - start_locator = self.page.get_by_role("button", name="Start My Server") - if start_locator.is_visible(): - start_locator.click() + else: + start_locator = self.page.get_by_role("button", name="Start My Server") + if start_locator.is_visible(): + start_locator.click() server_options = self.page.get_by_role("heading", name="Server Options") if server_options.is_visible(): - # select instance type (this will fail if this instance type is not - # available) self.page.locator(f"#profile-item-{self.instance_name}").click() self.page.get_by_role("button", name="Start").click() - # wait for server spinup - self.page.wait_for_url( - urllib.parse.urljoin(self.nebari_url, f"user/{self.username}/*"), - wait_until="networkidle", - timeout=180000, - ) - - # the jupyter page loads independent of network activity so here - # we wait for the File menu to be available on the page, a proxy for - # the jupyterlab page being loaded. + self.page.wait_for_url(re.compile(f".*user/{self.username}/.*"), timeout=180000) file_locator = self.page.get_by_text("File", exact=True) - file_locator.wait_for( - timeout=self.wait_for_server_spinup, - state="attached", - ) + file_locator.wait_for(state="attached", timeout=self.wait_for_server_spinup) - logger.debug(">>> Sign in complete.") + logger.debug(">>> Profile Spawn complete.") - def _check_for_kernel_popup(self): - """Is the kernel popup currently open? - - Returns - ------- - True if the kernel popup is open. - """ - self.page.wait_for_load_state("networkidle") - time.sleep(3) - visible = self.page.get_by_text("Select Kernel", exact=True).is_visible() - return visible - - def reset_workspace(self): - """Reset the Jupyterlab workspace. - - * Closes all Tabs & handle possible popups for saving changes, - * make sure any kernel popups are dealt with - * reset file browser is reset to root - * Finally, ensure that the Launcher screen is showing - """ - logger.info(">>> Reset JupyterLab workspace") - - # server is already running and there is no popup - popup = self._check_for_kernel_popup() - - # server is on running and there is a popup - if popup: - self._set_environment_via_popup(kernel=None) - - # go to Kernel menu - kernel_menuitem = self.page.get_by_role("menuitem", name="Kernel", exact=True) - kernel_menuitem.click() - # shut down multiple running kernels - with contextlib.suppress(Exception): - shut_down_all = self.page.get_by_text( - "Shut Down All Kernels...", exact=True - ) - shut_down_all.wait_for(timeout=300, state="attached") - shut_down_all.click() - - # shut down kernel if only one notebook is running - kernel_menuitem.click() - with contextlib.suppress(Exception): - shut_down_current = self.page.get_by_text("Shut Down Kernel", exact=True) - shut_down_current.wait_for(timeout=300, state="attached") - shut_down_current.click() - - # go back to root folder - self.page.get_by_title(f"/home/{self.username}", exact=True).locator( - "path" - ).click() - - # go to File menu + def stop_server(self): + """Stops the Nebari server via the Hub Control Panel.""" self.page.get_by_text("File", exact=True).click() - # close all tabs - self.page.get_by_role("menuitem", name="Close All Tabs", exact=True).click() - - # there may be a popup to save your work, don't save - if self.page.get_by_text("Save your work", exact=True).is_visible(): - self.page.get_by_role("button", name="Discard", exact=True).click() - - # wait to ensure that the Launcher is showing - self.page.get_by_text("VS Code [↗]", exact=True).wait_for( - timeout=3000, state="attached" - ) - - def _set_environment_via_popup(self, kernel=None): - """Set the environment kernel on a jupyter notebook via the popup - dialog box. If kernel is `None`, `No Kernel` is selected and the - popup is dismissed. - - Attributes - ---------- - kernel: str or None - (Optional) name of conda environment to set. Defaults to None. - - """ - if kernel is None: - # close dialog (deal with the two formats of this dialog) - try: - cancel_button = self.page.get_by_text("Cancel", exact=True) - if cancel_button.is_visible(): - cancel_button.click() - else: - self.page.mouse.click(0, 0) - except Exception: - self.page.locator("div").filter(has_text="No KernelSelect").get_by_role( - "button", name="No Kernel" - ).wait_for(timeout=300, state="attached") - else: - # set the environment - # failure here indicates that the environment doesn't exist either - # because of incorrect naming syntax or because the env is still - # being built - - new_launcher_popup = self.page.locator( - ".jp-KernelSelector-Dialog .jp-NewLauncher-table table" - ).nth(0) - if new_launcher_popup.is_visible(): - # for when the jupyterlab-new-launcher extension is installed - new_launcher_popup.locator("td").nth(0).click() - else: - # for when only the native launcher is available - self.page.get_by_role("combobox").nth(1).select_option(kernel) - # click Select to close popup (deal with the two formats of this dialog) - try: - self.page.get_by_role("button", name="Select Kernel").click() - except Exception: - self.page.locator("div").filter( - has_text="No KernelSelect" - ).get_by_role("button", name="Select Kernel").click() - - def set_environment(self, kernel): - """Set environment of a jupyter notebook. - - IMPORTANT: The focus MUST be on the notebook on which you want to set - the environment. - - Conda environments may still be being built shortly after deployment. - - Parameters - ---------- - kernel: str - Name of kernel to set. - - Returns - ------- - None - """ - - popup = self._check_for_kernel_popup() - # if there is not a kernel popup, make it appear - if not popup: - self.page.get_by_role("menuitem", name="Kernel", exact=True).click() - self.page.get_by_role("menuitem", name="Change Kernel…").get_by_text( - "Change Kernel…" - ).click() - - self._set_environment_via_popup(kernel) - - # wait for the jupyter UI to catch up before moving forward - # see if the jupyter notebook label for the conda env is visible - kernel_label_loc = self.page.get_by_role("button", name=kernel) - if not kernel_label_loc.is_visible(): - kernel_label_loc.wait_for(state="attached") - - def open_terminal(self): - """Open Terminal in the Nebari Jupyter Lab""" - self.page.get_by_text("File", exact=True).click() - self.page.get_by_text("New", exact=True).click() - self.page.get_by_role("menuitem", name="Terminal").get_by_text( - "Terminal" - ).click() - - def run_terminal_command(self, command): - """Run a command on the terminal in the Nebari Jupyter Lab - - Parameters - ---------- - command: str - command to run in the terminal - """ - self.page.get_by_role("textbox", name="Terminal input").fill(command) - self.page.get_by_role("textbox", name="Terminal input").press("Enter") - - def write_file(self, filepath, content): - """Write a file to Nebari instance filesystem - - The terminal is a blackbox for the browser. We can't access any of the - displayed text, therefore we have no way of knowing if the commands - are done executing. For this reason, there is an unavoidable sleep - here that prevents playwright from moving on to ensure that the focus - remains on the Terminal until we are done issuing our commands. - - Parameters - ---------- - filepath: str - path to write the file on the nebari file system - content: str - text to write to that file. - """ - start = dt.datetime.now() - logger.debug(f"Writing notebook to {filepath}") - self.open_terminal() - self.run_terminal_command(f"cat <{filepath}") - self.run_terminal_command(content) - self.run_terminal_command("EOF") - self.run_terminal_command(f"ls {filepath}") - logger.debug(f"time to complete {dt.datetime.now() - start}") - time.sleep(2) - - def stop_server(self) -> None: - """Stops the JupyterHub server by navigating to the Hub Control Panel.""" - self.page.get_by_text("File", exact=True).click() - with self.context.expect_page() as page_info: self.page.get_by_role("menuitem", name="Home", exact=True).click() home_page = page_info.value home_page.wait_for_load_state() stop_button = home_page.get_by_role("button", name="Stop My Server") - if not stop_button.is_visible(): - stop_button.wait_for(state="visible") + stop_button.wait_for(state="visible") stop_button.click() stop_button.wait_for(state="hidden") + + +# Factory method for creating different navigators if needed +def navigator_factory(navigator_type, **kwargs): + navigators = { + "login": LoginNavigator, + "server": ServerManager, + } + return navigators[navigator_type](**kwargs) diff --git a/tests/common/playwright_fixtures.py b/tests/common/playwright_fixtures.py index 03e17a5065..35ea36baad 100644 --- a/tests/common/playwright_fixtures.py +++ b/tests/common/playwright_fixtures.py @@ -5,76 +5,102 @@ import dotenv import pytest -from tests.common.navigator import Navigator +from tests.common.navigator import navigator_factory logger = logging.getLogger() -@pytest.fixture(scope="session") -def _navigator_session(request, browser_name, pytestconfig): - """Set up a navigator instance, login with username/password, start - a server. Teardown when session is complete. - Do not use this for individual tests, use `navigator` fixture - for tests.""" +def load_env_vars(): + """Load environment variables using dotenv and return necessary parameters.""" dotenv.load_dotenv() - # try/except added here in attempt to reach teardown after error in - # order to close the browser context which will save the video so I debug - # the error. - try: - nav = Navigator( - nebari_url=request.param.get("nebari_url") or os.environ["NEBARI_FULL_URL"], - username=request.param.get("keycloak_username") - or os.environ["KEYCLOAK_USERNAME"], - password=request.param.get("keycloak_password") - or os.environ["KEYCLOAK_PASSWORD"], - headless=not pytestconfig.getoption("--headed"), - slow_mo=pytestconfig.getoption("--slowmo"), - browser=browser_name, - auth="password", - instance_name=request.param.get( - "instance_name" - ), # small-instance included by default - video_dir="videos/", - ) - except Exception as e: - logger.debug(e) - raise - - try: - nav.login_password() - nav.start_server() - yield nav - except Exception as e: - logger.debug(e) - raise - finally: + return { + "nebari_url": os.getenv("NEBARI_FULL_URL"), + "username": os.getenv("KEYCLOAK_USERNAME"), + "password": os.getenv("KEYCLOAK_PASSWORD"), + } + + +def build_params(request, pytestconfig, extra_params=None): + """Construct and return parameters for navigator instances.""" + env_vars = load_env_vars() + params = { + "nebari_url": request.param.get("nebari_url") or env_vars["nebari_url"], + "username": request.param.get("keycloak_username") or env_vars["username"], + "password": request.param.get("keycloak_password") or env_vars["password"], + "auth": "password", + "video_dir": "videos/", + "headless": pytestconfig.getoption("--headed"), + "slow_mo": pytestconfig.getoption("--slowmo"), + } + if extra_params: + params.update(extra_params) + return params + + +def create_navigator(navigator_type, params): + """Create and return a navigator instance.""" + return navigator_factory(navigator_type, **params) + + +def pytest_sessionstart(session): + """Called before the start of the session. Clean up the videos directory.""" + _videos_path = Path("./videos") + if _videos_path.exists(): + for filename in os.listdir("./videos"): + filepath = _videos_path / filename + filepath.unlink() + + +# scope="function" will make sure that the fixture is created and destroyed for each test function. +@pytest.fixture(scope="function") +def navigator_session(request, pytestconfig): + session_type = request.param.get("session_type") + extra_params = request.param.get("extra_params", {}) + + # Get the test function name for video naming + test_name = request.node.originalname + video_name_prefix = f"video_{test_name}" + extra_params["video_name_prefix"] = video_name_prefix + + params = build_params(request, pytestconfig, extra_params) + + with create_navigator(session_type, params) as nav: + # Setup the navigator instance (e.g., login or start server) try: - nav.stop_server() + if session_type == "login": + nav.login() + elif session_type == "server": + nav.start_server() + yield nav except Exception as e: logger.debug(e) - nav.teardown() + raise + + +def parameterized_fixture(session_type, **extra_params): + """Utility function to create parameterized pytest fixtures.""" + return pytest.mark.parametrize( + "navigator_session", + [{"session_type": session_type, "extra_params": extra_params}], + indirect=True, + ) + + +def server_parameterized(instance_name=None, **kwargs): + return parameterized_fixture("server", instance_name=instance_name, **kwargs) + + +def login_parameterized(**kwargs): + return parameterized_fixture("login", **kwargs) @pytest.fixture(scope="function") -def navigator(_navigator_session): - """High level navigator instance with a reset workspace.""" - _navigator_session.reset_workspace() - yield _navigator_session +def navigator(navigator_session): + """High-level navigator instance. Can be overridden based on the available + parameterized decorator.""" + yield navigator_session @pytest.fixture(scope="session") def test_data_root(): - here = Path(__file__).parent - return here / "notebooks" - - -def navigator_parameterized( - nebari_url=None, keycloak_username=None, keycloak_password=None, instance_name=None -): - param = { - "instance_name": instance_name, - "nebari_url": nebari_url, - "keycloak_username": keycloak_username, - "keycloak_password": keycloak_password, - } - return pytest.mark.parametrize("_navigator_session", [param], indirect=True) + return Path(__file__).parent / "notebooks" diff --git a/tests/common/run_notebook.py b/tests/common/run_notebook.py deleted file mode 100644 index 0c39f4b5ac..0000000000 --- a/tests/common/run_notebook.py +++ /dev/null @@ -1,284 +0,0 @@ -import logging -import re -import time -from pathlib import Path -from typing import List, Union - -from tests.common.navigator import Navigator - -logger = logging.getLogger() - - -class Notebook: - def __init__(self, navigator: Navigator): - self.nav = navigator - self.nav.initialize - - def run( - self, - path, - expected_outputs: List[str], - conda_env: str, - timeout: float = 1000, - completion_wait_time: float = 2, - retry: int = 2, - retry_wait_time: float = 5, - exact_match: bool = True, - ): - """Run jupyter notebook and check for expected output text anywhere on - the page. - - Note: This will look for and exact match of expected_output_text - _anywhere_ on the page so be sure that your text is unique. - - Conda environments may still be being built shortly after deployment. - - Parameters - ---------- - path: str - Path to notebook relative to the root of the jupyterlab instance. - expected_outputs: List[str] - Text to look for in the output of the notebook. This can be a - substring of the actual output if exact_match is False. - conda_env: str - Name of conda environment. Python conda environments have the - structure "conda-env-nebari-git-nebari-git-dashboard-py" where - the actual name of the environment is "dashboard". - timeout: float - Time in seconds to wait for the expected output text to appear. - default: 1000 - completion_wait_time: float - Time in seconds to wait between checking for expected output text. - default: 2 - retry: int - Number of times to retry running the notebook. - default: 2 - retry_wait_time: float - Time in seconds to wait between retries. - default: 5 - exact_match: bool - If True, the expected output must match exactly. If False, the - expected output must be a substring of the actual output. - default: True - """ - logger.debug(f">>> Running notebook: {path}") - filename = Path(path).name - - # navigate to specific notebook - self.open_notebook(path) - # make sure the focus is on the dashboard tab we want to run - self.nav.page.get_by_role("tab", name=filename).get_by_text(filename).click() - self.nav.set_environment(kernel=conda_env) - - # make sure that this notebook is one currently selected - self.nav.page.get_by_role("tab", name=filename).get_by_text(filename).click() - - for _ in range(retry): - self._restart_run_all() - # Wait for a couple of seconds to make sure it's re-started - time.sleep(retry_wait_time) - self._wait_for_commands_completion(timeout, completion_wait_time) - all_outputs = self._get_outputs() - assert_match_all_outputs(expected_outputs, all_outputs, exact_match) - - def create_notebook(self, conda_env=None): - file_locator = self.nav.page.get_by_text("File", exact=True) - file_locator.wait_for( - timeout=self.nav.wait_for_server_spinup, - state="attached", - ) - file_locator.click() - submenu = self.nav.page.locator('[data-type="submenu"]').all() - submenu[0].click() - self.nav.page.get_by_role("menuitem", name="Notebook").get_by_text( - "Notebook", exact=True - ).click() - self.nav.page.wait_for_load_state("networkidle") - # make sure the focus is on the dashboard tab we want to run - # self.nav.page.get_by_role("tab", name=filename).get_by_text(filename).click() - self.nav.set_environment(kernel=conda_env) - - def open_notebook(self, path): - file_locator = self.nav.page.get_by_text("File", exact=True) - file_locator.wait_for( - timeout=self.nav.wait_for_server_spinup, - state="attached", - ) - file_locator.click() - self.nav.page.get_by_role("menuitem", name="Open from Path…").get_by_text( - "Open from Path…" - ).click() - self.nav.page.get_by_placeholder("/path/relative/to/jlab/root").fill(path) - self.nav.page.get_by_role("button", name="Open", exact=True).click() - # give the page a second to open, otherwise the options in the kernel - # menu will be disabled. - self.nav.page.wait_for_load_state("networkidle") - - if self.nav.page.get_by_text( - "Could not find path:", - exact=False, - ).is_visible(): - logger.debug("Path to notebook is invalid") - raise RuntimeError("Path to notebook is invalid") - - def assert_code_output( - self, - code: str, - expected_output: str, - timeout: float = 1000, - completion_wait_time: float = 2, - exact_match: bool = True, - ): - """ - Run code in last cell and check for expected output text anywhere on - the page. - - - Parameters - ---------- - code: str - Code to run in last cell. - expected_outputs: List[Union[re.Pattern, str]] - Text to look for in the output of the notebook. - timeout: float - Time in seconds to wait for the expected output text to appear. - default: 1000 - completion_wait_time: float - Time in seconds to wait between checking for expected output text. - """ - self.run_in_last_cell(code) - self._wait_for_commands_completion(timeout, completion_wait_time) - outputs = self._get_outputs() - actual_output = outputs[-1] if outputs else "" - assert_match_output(expected_output, actual_output, exact_match) - - def run_in_last_cell(self, code): - self._create_new_cell() - cell = self._get_last_cell() - cell.click() - cell.type(code) - # Wait for it to be ready to be executed - time.sleep(1) - cell.press("Shift+Enter") - # Wait for execution to start - time.sleep(0.5) - - def _create_new_cell(self): - new_cell_button = self.nav.page.query_selector( - 'button[data-command="notebook:insert-cell-below"]' - ) - new_cell_button.click() - - def _get_last_cell(self): - cells = self.nav.page.locator(".CodeMirror-code").all() - for cell in reversed(cells): - if cell.is_visible(): - return cell - raise ValueError("Unable to get last cell") - - def _wait_for_commands_completion( - self, timeout: float, completion_wait_time: float - ): - """ - Wait for commands to finish running - - Parameters - ---------- - timeout: float - Time in seconds to wait for the expected output text to appear. - completion_wait_time: float - Time in seconds to wait between checking for expected output text. - """ - elapsed_time = 0.0 - still_visible = True - start_time = time.time() - while elapsed_time < timeout: - running = self.nav.page.get_by_text("[*]").all() - still_visible = any(list(map(lambda r: r.is_visible(), running))) - if not still_visible: - break - elapsed_time = time.time() - start_time - time.sleep(completion_wait_time) - if still_visible: - raise ValueError( - f"Timeout Waited for commands to finish, " - f"but couldn't finish in {timeout} sec" - ) - - def _get_outputs(self) -> List[str]: - output_elements = self.nav.page.query_selector_all(".jp-OutputArea-output") - text_content = [element.text_content().strip() for element in output_elements] - return text_content - - def _restart_run_all(self): - # restart run all cells - self.nav.page.get_by_role("menuitem", name="Kernel", exact=True).click() - self.nav.page.get_by_role( - "menuitem", name="Restart Kernel and Run All Cells…" - ).get_by_text("Restart Kernel and Run All Cells…").click() - - # Restart dialog appears most, but not all of the time (e.g. set - # No Kernel, then Restart Run All) - restart_dialog_button = self.nav.page.get_by_role( - "button", name="Confirm Kernel Restart" - ) - if restart_dialog_button.is_visible(): - restart_dialog_button.click() - - -def assert_match_output( - expected_output: str, actual_output: str, exact_match: bool -) -> None: - """Assert that the expected_output is found in the actual_output. - - ---------- - Parameters - - expected_output: str - The expected output text or regular expression to find in the - actual output. - actual_output: str - The actual output text to search for the expected output. - exact_match: bool - If True, then the expected_output must match the actual_output - exactly. Otherwise, the expected_output must be found somewhere in - the actual_output. - """ - regex = re.compile(rf"{expected_output}") - match = ( - regex.fullmatch(actual_output) if exact_match else regex.search(actual_output) - ) - assert ( - match is not None - ), f"Expected output: {expected_output} not found in actual output: {actual_output}" - - -def assert_match_all_outputs( - expected_outputs: List[str], - actual_outputs: List[str], - exact_matches: Union[bool, List[bool]], -) -> None: - """Assert that the expected_outputs are found in the actual_outputs. - The expected_outputs and actual_outputs must be the same length. - - ---------- - Parameters - - expected_outputs: List[str] - A list of expected output text or regular expression to find in - the actual output. - actual_outputs: List[str] - A list of actual output text to search for the expected output. - exact_matches: Union[bool, List[bool]] - If True, then the expected_output must match the actual_output - exactly. Otherwise, the expected_output must be found somewhere in - the actual_output. If a list is provided, then it must be the same - length as expected_outputs and actual_outputs. - """ - if isinstance(exact_matches, bool): - exact_matches = [exact_matches] * len(expected_outputs) - - for exact_output, actual_output, exact in zip( - expected_outputs, actual_outputs, exact_matches - ): - assert_match_output(exact_output, actual_output, exact) diff --git a/tests/common/tests/test_notebook.py b/tests/common/tests/test_notebook.py deleted file mode 100644 index ba8cbbbf84..0000000000 --- a/tests/common/tests/test_notebook.py +++ /dev/null @@ -1,35 +0,0 @@ -import pytest - -from tests.common.run_notebook import assert_match_output - - -@pytest.mark.parametrize( - "expected, actual, exact", - [ - ("success: 6", "success: 6", True), - ("success", "success: 6", False), - ("6", "6", True), - ("cde", "abcde", False), - ("12.*5", "12345", True), - (".*5", "12345", True), - ("ab.*ef", "123abcdef123", False), - ], -) -def test_output_match(expected, actual, exact): - assert_match_output(expected, actual, exact_match=exact) - - -@pytest.mark.parametrize( - "expected, actual, exact", - [ - ("True", "False", True), - ("success: 6", "success", True), - ("60", "6", True), - ("abcde", "cde", True), - ("ab.*ef", "123abcdef123", True), - ], -) -def test_output_not_match(expected, actual, exact): - msg = f"Expected output: {expected} not found in actual output: {actual}" - with pytest.raises(AssertionError, match=msg): - assert_match_output(expected, actual, exact_match=exact) diff --git a/tests/tests_e2e/playwright/README.md b/tests/tests_e2e/playwright/README.md index 99a285f3a6..c328681273 100644 --- a/tests/tests_e2e/playwright/README.md +++ b/tests/tests_e2e/playwright/README.md @@ -1,199 +1,190 @@ -# Nebari integration testing with Playwright +# Nebari Integration Testing with Playwright -## How does it work? +## How Does It Work? -Playwright manages interactions with any website. We are using it to interact -with a deployed Nebari instance and test the various integrations that are -included. +Playwright manages interactions with websites, and we use it to interact with a deployed Nebari instance and test various integrations. -For our test suite, we utilize Playwright's synchronous API. The first task -is to launch the web browser you'd like to test in. Options in our test suite -are `chromium`, `webkit`, and `firefox`. Playwright uses browser contexts to -achieve test isolation. The context can either be created by default or -manually (for the purposes of generating multiple contexts per test in the case -of admin vs user testing). Next the page on the browser is created. For all -tests this starts as a blank page, then during the test, we navigate to a given -url. This is all achieved in the `setup` method of the `Navigator` class. +We use Playwright's synchronous API for our test suite. The first task is to launch the web browser of your choice: `chromium`, `webkit`, or `firefox`. Playwright uses browser contexts for test isolation, which can be created by default or manually for scenarios like admin vs. user testing. Each test starts with a blank page, and we navigate to a given URL during the test. This setup is managed by the `setup` method in the `Navigator` class. + +## Directory Structure + +The project directory structure is as follows: + +``` +tests +├── common +│   ├── __init__.py +│   ├── navigator.py +│   ├── handlers.py +│   ├── playwright_fixtures.py +├── ... +├── tests_e2e +│   └── playwright +│   ├── README.md +│   └── test_playwright.py +``` + +- `test_data/`: Contains test files, such as sample notebooks. +- `test_playwright.py`: The main test script that uses Playwright for integration testing. +- `navigator.py`: Contains the `NavigatorMixin` class, which manages browser + interactions and context. As well as the `LoginNavigator` class, which manages user + authentication and `ServerManager` class, which manages the user instance spawning. +- `handlers.py`: Contains classes fore handling the different level of access to + services a User might encounter, such as Notebook, Conda-store and others. ## Setup -Install Nebari with the development requirements (which include Playwright) +1. **Install Nebari with Development Requirements** -`pip install -e ".[dev]"` + Install Nebari including development requirements (which include Playwright): -Then install playwright itself (required). + ```bash + pip install -e ".[dev]" + ``` -`playwright install` +2. **Install Playwright** -> If you see the warning `BEWARE: your OS is not officially supported by Playwright; downloading fallback build., it is not critical.` Playwright will likely still work microsoft/playwright#15124 + Install Playwright: -### Create environment file + ```bash + playwright install + ``` -Create a copy of the `.env` template file + *Note:* If you see the warning `BEWARE: your OS is not officially supported by Playwright; downloading fallback build`, it is not critical. Playwright should still work (see microsoft/playwright#15124). -```bash -cd tests_e2e/playwright -cp .env.tpl .env -``` +3. **Create Environment Vars** -Fill in the newly created `.env` file with the following values: + Fill in your execution space environment with the following values: -* KEYCLOAK_USERNAME: Nebari username for username/password login OR Google email address or Google sign in -* KEYCLOAK_PASSWORD: Password associated with USERNAME -* NEBARI_FULL_URL: full url path including scheme to Nebari instance, e.g. "https://nebari.quansight.dev/" + - `KEYCLOAK_USERNAME`: Nebari username for username/password login or Google email address/Google sign-in. + - `KEYCLOAK_PASSWORD`: Password associated with `KEYCLOAK_USERNAME`. + - `NEBARI_FULL_URL`: Full URL path including scheme to the Nebari instance (e.g., "https://nebari.quansight.dev/"). -This user can be created with the following command (or you can use an existing non-root user): + This user can be created with the following command (or use an existing non-root user): -``` -nebari keycloak adduser --user --config -``` + ```bash + nebari keycloak adduser --user --config + ``` -## Running the Playwright tests +## Running the Playwright Tests -The playwright tests are run inside of pytest using +Playwright tests are run inside of pytest using: -```python +```bash pytest tests_e2e/playwright/test_playwright.py ``` -Videos of the test playback will be available in `$PWD/videos/`. -To see what is happening while the test is run, pass the `--headed` option to `pytest`. -You can also add the `--slowmo=$MILLI_SECONDS` option to add a delay before each action -by Playwright and thus slowing down the process. +Videos of the test playback will be available in `$PWD/videos/`. To disabled the browser +runtime preview of what is happening while the test runs, pass the `--headed` option to `pytest`. You +can also add the `--slowmo=$MILLI_SECONDS` option to introduce a delay before each +action by Playwright, thereby slowing down the process. -Another option is to run playwright methods outside of pytest. Both -`navigator.py` and `run_notebook.py` can be run as scripts. For example, +Alternatively, you can run Playwright methods outside of pytest. Below an example of +how to run a test, where you can interface with the Notebook handler: ```python - import os - - import dotenv - # load environment variables from .env file - dotenv.load_dotenv() - # instantiate the navigator class - nav = Navigator( - nebari_url="https://nebari.quansight.dev/", - username=os.environ["KEYCLOAK_USERNAME"], - password=os.environ["KEYCLOAK_PASSWORD"], - auth="password", - instance_name="small-instance", - headless=False, - slow_mo=100, - ) - # go through login sequence (defined by `auth` method in Navigator class) - nav.login() - # start the nebari server (defined by `instance_type` in Navigator class) - nav.start_server() - # reset the jupyterlab workspace to ensure we're starting with only the - # Launcher screen open, and we're in the root directory. - nav.reset_workspace() - # instantiate our test application - test_app = RunNotebook(navigator=nav) - # Write the sample notebook on the nebari instance - notebook_filepath_in_repo = ( - "tests_e2e/playwright/test_data/test_notebook_output.ipynb" - ) - notebook_filepath_on_nebari = "test_notebook_output.ipynb" - with open(notebook_filepath_in_repo, "r") as notebook: - test_app.nav.write_file( - filepath=notebook_filepath_on_nebari, content=notebook.read() - ) - # run a sample notebook - test_app.run_notebook( - path="nebari/tests_e2e/playwright/test_data/test_notebook_output.ipynb", - expected_output_text="success: 6", - conda_env="conda-env-default-py", - ) - # close out playwright and its associated browser handles - nav.teardown() +import os +import dotenv +from pathlib import Path + +from tests.common.navigator import ServerManager +from tests.common.handlers import Notebook + + +# Instantiate the Navigator class +nav = ServerManage( + nebari_url="https://nebari.quansight.dev/", + username=os.environ["KEYCLOAK_USERNAME"], + password=os.environ["KEYCLOAK_PASSWORD"], + auth="password", + instance_name="small-instance", + headless=False, + slow_mo=100, +) + + +notebook_manager = Notebook(navigator=navigator) + +# Reset the JupyterLab workspace to ensure we're starting with only the Launcher screen open and in the root directory. +notebook_manager.reset_workspace() + +notebook_name = "test_notebook_output.ipynb" +notebook_path = Path("tests_e2e/playwright/test_data") / notebook_name + +assert notebook_path.exists() + +# Write the sample notebook on the Nebari instance +with open(notebook_path, "r") as notebook: + notebook_manager.write_file(filepath=notebook_name, content=notebook.read()) + +# Run a sample notebook (and collect the outputs) +outputs = notebook_manager.run_notebook( + notebook_name=notebook_name, kernel="default" +) + +# Close out Playwright and its associated browser handles +nav.teardown() ``` -## Writing Playwright tests +## Writing Playwright Tests -In general most of the testing happens through `locators` which is Playwright's -way of connecting a python object to the HTML element on the page. -The Playwright API has several mechanisms for getting a locator for an item on -the page (`get_by_role`, `get_by_text`, `get_by_label`, `get_by_placeholder`, -etc). +Most testing is done through `locators`, which connect Python objects to HTML elements on the page. Playwright offers several mechanisms for getting a locator for an item on the page, such as `get_by_role`, `get_by_text`, `get_by_label`, and `get_by_placeholder`. ```python button = self.page.get_by_role("button", name="Sign in with Keycloak") ``` -Once you have a handle on a locator, you can interact with it in different ways, -depending on the type of object. For example, clicking -a button: +Once you have a handle on a locator, you can interact with it in various ways, depending on the type of object. For example, clicking a button: ```python button.click() ``` -Occasionally you'll need to wait for things to load on the screen. We can -either wait for the page to finish loading: +Sometimes you'll need to wait for elements to load on the screen. You can wait for the page to finish loading: ```python self.page.wait_for_load_state("networkidle") ``` -or we can wait for something specific to happen with the locator itself: +Or wait for something specific to happen with the locator itself: ```python button.wait_for(timeout=3000, state="attached") ``` -Note that waiting for the page to finish loading may be deceptive inside of -Jupyterlab since things may need to load _inside_ the page, not necessarily -causing network traffic - or causing several bursts network traffic, which -would incorrectly pass the `wait_for_load_state` after the first burst. +Note that waiting for the page to finish loading may be misleading inside of JupyterLab since elements may need to load _inside_ the page or cause several bursts of network traffic. + +Playwright has a built-in auto-wait feature that waits for a timeout period for actionable items. See [Playwright Actionability](https://playwright.dev/docs/actionability). + +## Parameterized Decorators + +### Usage -Playwright has a built-in auto-wait feature which waits for a timeout period -for some actionable items. See https://playwright.dev/docs/actionability . +Parameterized decorators in your test setup allow you to run tests with different configurations or contexts. They are particularly useful for testing different scenarios, such as varying user roles or application states. -### Workflow for creating new tests +To easy the control over the initial setup of spawning the user instance and login, we +already provider two base decorators that can be used in your test: +- `server_parameterized`: Allows to login and spin a new instance of the server, based + on the provided instance type. Allows for the nav.page to be run within the JupyterLab environment. +- ` login_parameterized`: Allow login to Nebari and sets you test workspace to the main + hub, allow your tests to attest things like the launcher screen or the navbar components. -An example of running a new run notebook test might look like this: +For example, using parameterized decorators to test different user roles might look like this: ```python - import os - - import dotenv - # load environment variables from .env file - dotenv.load_dotenv() - # instantiate the navigator class - nav = Navigator( - nebari_url="https://nebari.quansight.dev/", - username=os.environ["KEYCLOAK_USERNAME"], - password=os.environ["KEYCLOAK_PASSWORD"], - auth="password", - instance_name="small-instance", - headless=False, - slow_mo=100, - ) - # go through login sequence (defined by `auth` method in Navigator class) - nav.login() - # start the nebari server (defined by `instance_type` in Navigator class) - nav.start_server() - # reset the jupyterlab workspace to ensure we're starting with only the - # Launcher screen open, and we're in the root directory. - nav.reset_workspace() - # instantiate our test application - test_app = RunNotebook(navigator=nav) - # Write the sample notebook on the nebari instance - notebook_filepath_in_repo = ( - "tests_e2e/playwright/test_data/test_notebook_output.ipynb" - ) - notebook_filepath_on_nebari = "test_notebook_output.ipynb" - with open(notebook_filepath_in_repo, "r") as notebook: - test_app.nav.write_file( - filepath=notebook_filepath_on_nebari, content=notebook.read() - ) - # run a sample notebook - test_app.run_notebook( - path="nebari/tests_e2e/playwright/test_data/test_notebook_output.ipynb", - expected_output_text="success: 6", - conda_env="conda-env-default-py", - ) - # close out playwright and its associated browser handles - nav.teardown() +@pytest.mark.parametrize("is_admin", [False]) +@login_parameterized() +def test_role_button(navigator, is_admin): + _ = navigator.page.get_by_role("button", name="Admin Button").is_visible() + assert _ == is_admin + # Perform tests specific to the user role... ``` +In the example above, we used the `login_parameterized` decorator which will log in as an user +(based on the KEYCLOAK_USERNAME and KEYCLOAK_PASSWORD) and and let you wander under the logged workspace, +we attest for the presence of the "Admin Button" in the page (which does not exist). + +If your test suit presents a need for a more complex sequence of actions or special +parsing around the contents present in each page, you can create +your own handler to execute the auxiliary actions while the test is running. Check the +`handlers.py` over some examples of how that's being done. diff --git a/tests/tests_e2e/playwright/test_playwright.py b/tests/tests_e2e/playwright/test_playwright.py index 903af3f0dd..9d04a4e027 100644 --- a/tests/tests_e2e/playwright/test_playwright.py +++ b/tests/tests_e2e/playwright/test_playwright.py @@ -1,18 +1,93 @@ -from tests.common.playwright_fixtures import navigator_parameterized -from tests.common.run_notebook import Notebook +import pytest +from playwright.sync_api import expect +from tests.common.handlers import CondaStore, Notebook +from tests.common.playwright_fixtures import login_parameterized, server_parameterized + + +@login_parameterized() +def test_login_logout(navigator): + expect(navigator.page.get_by_text(navigator.username)).to_be_visible() + + navigator.logout() + expect(navigator.page.get_by_text("Sign in with Keycloak")).to_be_visible() + + +@pytest.mark.parametrize( + "services", + [ + ( + [ + "Home", + "Token", + "User Management", + "Argo Workflows", + "Environment Management", + "Monitoring", + ] + ), + ], +) +@login_parameterized() +def test_navbar_services(navigator, services): + navigator.page.goto(navigator.nebari_url + "hub/home") + navigator.page.wait_for_load_state("networkidle") + navbar_items = navigator.page.locator("#thenavbar").get_by_role("link") + navbar_items_names = [item.text_content() for item in navbar_items.all()] + assert len(navbar_items_names) == len(services) + assert navbar_items_names == services + + +@pytest.mark.parametrize( + "expected_outputs", + [ + (["success: 6"]), + ], +) +@server_parameterized(instance_name="small-instance") +def test_notebook(navigator, test_data_root, expected_outputs): + notebook_manager = Notebook(navigator=navigator) + + notebook_manager.reset_workspace() -@navigator_parameterized(instance_name="small-instance") -def test_notebook(navigator, test_data_root): - test_app = Notebook(navigator=navigator) notebook_name = "test_notebook_output.ipynb" notebook_path = test_data_root / notebook_name + assert notebook_path.exists() + with open(notebook_path, "r") as notebook: - test_app.nav.write_file(filepath=notebook_name, content=notebook.read()) - test_app.run( - path=notebook_name, - expected_outputs=["success: 6"], - conda_env="default *", - timeout=500, + notebook_manager.write_file(filepath=notebook_name, content=notebook.read()) + + outputs = notebook_manager.run_notebook( + notebook_name=notebook_name, kernel="default" ) + + assert outputs == expected_outputs + + # Clean up + notebook_manager.reset_workspace() + + +@pytest.mark.parametrize( + "namespaces", + [ + (["analyst", "developer", "global", "nebari-git", "users"]), + ], +) +@server_parameterized(instance_name="small-instance") +def test_conda_store_ui(navigator, namespaces): + conda_store = CondaStore(navigator=navigator) + + conda_store.reset_workspace() + + conda_store.conda_store_ui() + + shown_namespaces = conda_store._get_shown_namespaces() + shown_namespaces.sort() + + namespaces.append(navigator.username) + namespaces.sort() + + assert shown_namespaces == namespaces + # Clean up + conda_store.reset_workspace()