diff --git a/sdks/python/src/opik/config.py b/sdks/python/src/opik/config.py index dc82c1f108..011ef89399 100644 --- a/sdks/python/src/opik/config.py +++ b/sdks/python/src/opik/config.py @@ -147,6 +147,9 @@ def config_file_fullpath(self) -> pathlib.Path: def save_to_file(self) -> None: """ Save configuration to a file + + Raises: + OSError: If there is an issue writing to the file. """ config_file_content = configparser.ConfigParser() @@ -158,12 +161,15 @@ def save_to_file(self) -> None: if self.api_key is not None: config_file_content["opik"]["api_key"] = self.api_key - with open( - self.config_file_fullpath, mode="w+", encoding="utf-8" - ) as config_file: - config_file_content.write(config_file) - - LOGGER.info(f"Saved configuration to a file: {self.config_file_fullpath}") + try: + with open( + self.config_file_fullpath, mode="w+", encoding="utf-8" + ) as config_file: + config_file_content.write(config_file) + LOGGER.info(f"Configuration saved to file: {self.config_file_fullpath}") + except OSError as e: + LOGGER.error(f"Failed to save configuration: {e}") + raise def update_session_config(key: str, value: Any) -> None: diff --git a/sdks/python/src/opik/opik_configure.py b/sdks/python/src/opik/opik_configure.py index 0920922d03..e2bc6016c0 100644 --- a/sdks/python/src/opik/opik_configure.py +++ b/sdks/python/src/opik/opik_configure.py @@ -1,15 +1,13 @@ -import logging -from typing import cast import getpass -from typing import Final, Optional +import logging +from typing import Final, List, Optional, Tuple, cast import httpx import opik.config -from opik import httpx_client from opik.config import ( - OPIK_BASE_URL_LOCAL, OPIK_BASE_URL_CLOUD, + OPIK_BASE_URL_LOCAL, OPIK_WORKSPACE_DEFAULT_NAME, ) from opik.exceptions import ConfigurationError @@ -19,6 +17,9 @@ HEALTH_CHECK_URL_POSTFIX: Final[str] = "/is-alive/ping" HEALTH_CHECK_TIMEOUT: Final[float] = 1.0 +URL_ACCOUNT_DETAILS: Final[str] = "https://www.comet.com/api/rest/v2/account-details" +URL_WORKSPACE_GET_LIST: Final[str] = "https://www.comet.com/api/rest/v2/workspaces" + def is_interactive() -> bool: """ @@ -30,116 +31,121 @@ def is_interactive() -> bool: def is_instance_active(url: str) -> bool: """ - Returns True if given Opik URL responds to an HTTP GET request. - """ - http_client = httpx_client.get( - workspace=OPIK_WORKSPACE_DEFAULT_NAME, - api_key=None, - ) + Returns True if the given Opik URL responds to an HTTP GET request. - try: - http_client.timeout = HEALTH_CHECK_TIMEOUT - response = http_client.get(url=url + HEALTH_CHECK_URL_POSTFIX) + Args: + url (str): The base URL of the instance to check. - if response.status_code == 200: - return True + Returns: + bool: True if the instance responds with HTTP status 200, otherwise False. + """ + try: + with httpx.Client(timeout=HEALTH_CHECK_TIMEOUT) as http_client: + response = http_client.get(url=url + HEALTH_CHECK_URL_POSTFIX) + return response.status_code == 200 + except httpx.ConnectTimeout: + return False except Exception: return False - return False - def is_workspace_name_correct(api_key: str, workspace: str) -> bool: """ - Returns True if given cloud Opik workspace are correct. - - Raises: - ConnectionError: + Verifies whether the provided workspace name exists in the user's cloud Opik account. - """ + Args: + api_key (str): The API key used for authentication with the Opik service. + workspace (str): The name of the workspace to check. - url = "https://www.comet.com/api/rest/v2/workspaces" + Returns: + bool: True if the workspace is found, False otherwise. - client = httpx.Client() - client.headers.update( - { - "Authorization": f"{api_key}", - } - ) + Raises: + ConnectionError: Raised if there's an issue with connecting to the Opik service, or the response is not successful. + """ try: - response = client.get(url=url) + with httpx.Client() as client: + client.headers.update({"Authorization": f"{api_key}"}) + response = client.get(url=URL_WORKSPACE_GET_LIST) + except httpx.RequestError as e: + # Raised for network-related errors such as timeouts + raise ConnectionError(f"Network error: {str(e)}") except Exception as e: - raise ConnectionError(f"Error while checking workspace status: {str(e)}") + raise ConnectionError(f"Unexpected error occurred: {str(e)}") if response.status_code != 200: - raise ConnectionError(f"Error while checking workspace status: {response.text}") + raise ConnectionError(f"HTTP error: {response.status_code} - {response.text}") - workspaces = response.json()["workspaceNames"] + workspaces: List[str] = response.json().get("workspaceNames", []) - if workspace in workspaces: - return True - else: - return False + return workspace in workspaces def is_api_key_correct(api_key: str) -> bool: """ - Returns True if given cloud Opik API is correct. + Validates if the provided Opik API key is correct by sending a request to the cloud API. - Raises: - ConnectionError: - """ - url = "https://www.comet.com/api/rest/v2/account-details" + Args: + api_key (str): The API key used for authentication. - client = httpx.Client() - client.headers.update( - { - "Authorization": f"{api_key}", - } - ) + Returns: + bool: True if the API key is valid (status 200), False if the key is invalid (status 401 or 403). + Raises: + ConnectionError: If a network-related error occurs or the response status is neither 200, 401, nor 403. + """ try: - response = client.get(url=url) + with httpx.Client() as client: + client.headers.update({"Authorization": f"{api_key}"}) + response = client.get(url=URL_ACCOUNT_DETAILS) if response.status_code == 200: return True elif response.status_code in [401, 403]: return False + else: + raise ConnectionError(f"Error while checking API key: {response.text}") - raise ConnectionError(f"Error while checking API key: {response.text}") - + except httpx.RequestError as e: + raise ConnectionError(f"Network error occurred: {str(e)}") except Exception as e: - raise ConnectionError(f"Error while checking API key: {str(e)}") + raise ConnectionError(f"Unexpected error occurred: {str(e)}") def get_default_workspace(api_key: str) -> str: """ - Returns default Opik workspace name. + Retrieves the default Opik workspace name associated with the given API key. + + Args: + api_key (str): The API key used for authentication. + + Returns: + str: The default workspace name. Raises: - ConnectionError: + ConnectionError: If there's an error while fetching the default workspace. """ - url = "https://www.comet.com/api/rest/v2/account-details" + try: + with httpx.Client() as client: + client.headers.update({"Authorization": f"{api_key}"}) + response = client.get(url=URL_ACCOUNT_DETAILS) - client = httpx.Client() - client.headers.update( - { - "Authorization": f"{api_key}", - } - ) + if response.status_code != 200: + raise ConnectionError( + f"Error while getting default workspace name: {response.text}" + ) - try: - response = client.get(url=url) - except Exception as e: - raise ConnectionError(f"Error while getting default workspace name: {str(e)}") + default_workspace_name = response.json().get("defaultWorkspaceName") + if not default_workspace_name: + raise ConnectionError("defaultWorkspaceName not found in the response.") - if response.status_code != 200: - raise ConnectionError( - f"Error while getting default workspace name: {response.text}" - ) + return default_workspace_name - return response.json()["defaultWorkspaceName"] + except httpx.RequestError as e: + raise ConnectionError(f"Network error occurred: {str(e)}") + except Exception as e: + raise ConnectionError(f"Unexpected error occurred: {str(e)}") def _update_config( @@ -148,14 +154,15 @@ def _update_config( workspace: str, ) -> None: """ - Save changes to config file and update current session config + Save changes to the config file and update the current session configuration. Args: - api_key - url - workspace + api_key (Optional[str]): The API key for the Opik Cloud service. Can be None if not using Opik Cloud. + url (str): The base URL of the Opik instance (local or cloud). + workspace (str): The name of the workspace to be saved. + Raises: - ConfigurationError + ConfigurationError: Raised if there is an issue saving the configuration or updating the session. """ try: new_config = opik.config.OpikConfig( @@ -165,45 +172,55 @@ def _update_config( ) new_config.save_to_file() - # update session config + # Update current session configuration opik.config.update_session_config("api_key", api_key) opik.config.update_session_config("url_override", url) opik.config.update_session_config("workspace", workspace) - return - except Exception as e: - raise ConfigurationError(str(e)) + LOGGER.error(f"Failed to update config: {str(e)}") + raise ConfigurationError("Failed to update configuration.") def _ask_for_url() -> str: """ - Ask user for Opik instance URL and check if it is accessible. + Prompt the user for an Opik instance URL and check if it is accessible. + The function retries up to 3 times if the URL is not accessible. + + Returns: + str: A valid Opik instance URL. + + Raises: + ConfigurationError: Raised if the URL provided by the user is not accessible after 3 attempts. """ - retries = 2 + retries = 3 while retries > 0: user_input_opik_url = input("Please enter your Opik instance URL:") - # Validate it is accessible using health if is_instance_active(user_input_opik_url): - # If yes → Save return user_input_opik_url else: - # If no → Retry up to 2 times - ? Add message to docs ? LOGGER.error( f"Opik is not accessible at {user_input_opik_url}. Please try again, the URL should follow a format similar to {OPIK_BASE_URL_LOCAL}" ) retries -= 1 raise ConfigurationError( - "Can't use URL provided by user. Opik instance is not active or not found." + "Cannot use the URL provided by the user. Opik instance is not active or not found." ) def _ask_for_api_key() -> str: """ - Ask user for cloud Opik instance API key and check if is it correct. + Prompt the user for an Opik cloud API key and verify its validity. + The function retries up to 3 times if the API key is invalid. + + Returns: + str: A valid Opik API key. + + Raises: + ConfigurationError: Raised if the API key provided by the user is invalid after 3 attempts. """ retries = 3 LOGGER.info( @@ -226,7 +243,18 @@ def _ask_for_api_key() -> str: def _ask_for_workspace(api_key: str) -> str: """ - Ask user for cloud Opik instance workspace name. + Prompt the user for an Opik instance workspace name and verify its validity. + + The function retries up to 3 times if the workspace name is invalid. + + Args: + api_key (str): The API key used to verify the workspace name. + + Returns: + str: A valid workspace name. + + Raises: + ConfigurationError: Raised if the workspace name is invalid after 3 attempts. """ retries = 3 @@ -247,9 +275,20 @@ def _ask_for_workspace(api_key: str) -> str: def ask_user_for_approval(message: str) -> bool: + """ + Prompt the user with a message for approval (Y/Yes/N/No). + + Args: + message (str): The message to display to the user. + + Returns: + bool: True if the user approves (Y/Yes/empty input), False if the user disapproves (N/No). + + Logs: + Error when the user input is not recognized. + """ while True: - users_choice = input(message) - users_choice = users_choice.upper() + users_choice = input(message).strip().upper() if users_choice in ("Y", "YES", ""): return True @@ -260,6 +299,98 @@ def ask_user_for_approval(message: str) -> bool: LOGGER.error("Wrong choice. Please try again.") +def _get_api_key( + api_key: Optional[str], + current_config: opik.config.OpikConfig, + force: bool, +) -> Tuple[str, bool]: + """ + Determines the correct API key based on the current configuration, force flag, and user input. + + Args: + api_key (Optional[str]): The user-provided API key. + current_config (OpikConfig): The current configuration object. + force (bool): Whether to force reconfiguration. + + Returns: + Tuple[str, bool]: A tuple containing the validated API key and a boolean indicating + if the configuration file needs updating. + """ + config_file_needs_updating = False + + if force and api_key is None: + api_key = _ask_for_api_key() + config_file_needs_updating = True + elif api_key is None and current_config.api_key is None: + api_key = _ask_for_api_key() + config_file_needs_updating = True + elif api_key is None and current_config.api_key is not None: + api_key = current_config.api_key + # fixme if force is True -> need to save anyway? + + # todo add force and api_key is NOT None -> need to save? + # todo force is False, api_key is not None -> need to save? + + # fixme is this check for mypy? + # Ensure the API key is not None + api_key = cast(str, api_key) + + return api_key, config_file_needs_updating + + +def _get_workspace( + workspace: Optional[str], + api_key: str, + current_config: opik.config.OpikConfig, + force: bool, +) -> Tuple[str, bool]: + """ + Determines the correct workspace based on current configuration, force flag, and user input. + + Args: + workspace (Optional[str]): The user-provided workspace name. + api_key (str): The validated API key. + current_config (OpikConfig): The current configuration object. + force (bool): Whether to force reconfiguration. + + Returns: + Tuple[str, bool]: The validated or selected workspace name and a boolean + indicating whether the configuration file needs updating. + + Raises: + ConfigurationError: If the provided workspace is invalid. + """ + + # Case 1: Workspace was provided by the user and is valid + if workspace is not None: + if not is_workspace_name_correct(api_key, workspace): + raise ConfigurationError( + "Workspace `%s` is incorrect for the given API key.", workspace + ) + return workspace, True + + # Case 2: Use workspace from current configuration if not forced to change + if ( + "workspace" in current_config.model_fields_set + and current_config.workspace != OPIK_WORKSPACE_DEFAULT_NAME + and not force + ): + return current_config.workspace, False + + # Case 3: No workspace provided, prompt the user + default_workspace = get_default_workspace(api_key) + use_default_workspace = ask_user_for_approval( + f'Do you want to use "{default_workspace}" workspace? (Y/n)' + ) + + if use_default_workspace: + workspace = default_workspace + else: + workspace = _ask_for_workspace(api_key=api_key) + + return workspace, True + + def configure( api_key: Optional[str] = None, workspace: Optional[str] = None, @@ -268,11 +399,12 @@ def configure( force: bool = False, ) -> None: """ - Create a local configuration file for the Python SDK. If a configuration file already exists, it will not be overwritten unless the `force` parameter is set to True. + Create a local configuration file for the Python SDK. If a configuration file already exists, + it will not be overwritten unless the `force` parameter is set to True. Args: - api_key: The API key if using a Opik Cloud. - workspace: The workspace name if using a Opik Cloud. + api_key: The API key if using an Opik Cloud. + workspace: The workspace name if using an Opik Cloud. url: The URL of the Opik instance if you are using a local deployment. use_local: Whether to use a local deployment. force: If true, the configuration file will be recreated and existing settings will be overwritten. @@ -301,15 +433,14 @@ def _configure_cloud( force: bool = False, ) -> None: """ - Login to cloud Opik instance + Configure the cloud Opik instance by handling API key and workspace settings. Args: - api_key: The API key if using a Opik Cloud. - workspace: The workspace name if using a Opik Cloud. - force: If true, the configuration file will be recreated and existing settings will be overwritten. + api_key (Optional[str]): The API key for the Opik Cloud. + workspace (Optional[str]): The workspace name for the Opik Cloud. + force (bool): If True, forces reconfiguration by overwriting the existing settings. """ current_config = opik.config.OpikConfig() - config_file_needs_updating = False # TODO: Update the is_interactive() check, today always returns True so commented the code below # # first check parameters. @@ -323,51 +454,23 @@ def _configure_cloud( # ): # raise ConfigurationError("No workspace name provided for cloud Opik instance.") - # Ask for API key - if force and api_key is None: - api_key = _ask_for_api_key() - config_file_needs_updating = True - elif api_key is None and current_config.api_key is None: - api_key = _ask_for_api_key() - config_file_needs_updating = True - elif api_key is None and current_config.api_key is not None: - api_key = current_config.api_key - - api_key = cast(str, api_key) # by that moment we must be sure it's not None. - - # Check passed workspace (if it was passed) - if workspace is not None: - if is_workspace_name_correct(api_key, workspace): - config_file_needs_updating = True - else: - raise ConfigurationError( - "Workspace `%s` is incorrect for the given API key.", workspace - ) - else: - # Workspace was not passed, we check if there is already configured value - # if workspace already configured - will use this value - if ( - "workspace" in current_config.model_fields_set - and current_config.workspace != OPIK_WORKSPACE_DEFAULT_NAME - and not force - ): - workspace = current_config.workspace - - # Check what their default workspace is, and we ask them if they want to use the default workspace - if workspace is None: - default_workspace = get_default_workspace(api_key) - use_default_workspace = ask_user_for_approval( - f'Do you want to use "{default_workspace}" workspace? (Y/n)' - ) - - if use_default_workspace: - workspace = default_workspace - else: - workspace = _ask_for_workspace(api_key=api_key) + # Handle API key: get or prompt for one if needed + api_key, update_config_with_api_key = _get_api_key( + api_key=api_key, + current_config=current_config, + force=force, + ) - config_file_needs_updating = True + # Handle workspace: get or prompt for one if needed + workspace, update_config_with_workspace = _get_workspace( + workspace=workspace, + api_key=api_key, + current_config=current_config, + force=force, + ) - if config_file_needs_updating: + # Update configuration if either API key or workspace has changed + if update_config_with_api_key or update_config_with_workspace: _update_config( api_key=api_key, url=OPIK_BASE_URL_CLOUD, @@ -375,21 +478,21 @@ def _configure_cloud( ) else: LOGGER.info( - "Opik is already configured, you can check the settings by viewing the config file at %s", - opik.config.OpikConfig().config_file_fullpath, + "Opik is already configured. You can check the settings by viewing the config file at %s", + current_config.config_file_fullpath, ) def _configure_local(url: Optional[str], force: bool = False) -> None: """ - Login to local Opik deployment + Configure the local Opik instance by setting the local URL and workspace. Args: - url: The URL of the local Opik instance. - force: Whether to force the configuration even if local settings exist. + url (Optional[str]): The URL of the local Opik instance. + force (bool): Whether to force the configuration even if local settings exist. Raises: - ConfigurationError + ConfigurationError: Raised if the Opik instance is not active or not found. """ # TODO: this needs to be refactored - _login_local might only need url from the outside. # But we still have to init api_key and workspace because they are required in order to update config @@ -397,6 +500,7 @@ def _configure_local(url: Optional[str], force: bool = False) -> None: workspace = OPIK_WORKSPACE_DEFAULT_NAME current_config = opik.config.OpikConfig() + # Step 1: If the URL is provided and active, update the configuration if url is not None and is_instance_active(url): _update_config( api_key=api_key, @@ -405,15 +509,15 @@ def _configure_local(url: Optional[str], force: bool = False) -> None: ) return + # Step 2: Check if the default local instance is active if is_instance_active(OPIK_BASE_URL_LOCAL): if not force and current_config.url_override == OPIK_BASE_URL_LOCAL: - # Local Opik url is configured and local - # instance is running, everything is ready. LOGGER.info( - f"Opik is already configured to local to the running instance at {OPIK_BASE_URL_LOCAL}." + f"Opik is already configured to local instance at {OPIK_BASE_URL_LOCAL}." ) return + # Step 4: Ask user if they want to use the found local instance use_url = ask_user_for_approval( f"Found local Opik instance on: {OPIK_BASE_URL_LOCAL}, do you want to use it? (Y/n)" ) @@ -426,6 +530,7 @@ def _configure_local(url: Optional[str], force: bool = False) -> None: ) return + # Step 5: Ask user for URL if no valid local instance is found or approved user_input_url = _ask_for_url() _update_config( api_key=api_key, diff --git a/sdks/python/tests/unit/test_config.py b/sdks/python/tests/unit/test_config.py new file mode 100644 index 0000000000..48007599c3 --- /dev/null +++ b/sdks/python/tests/unit/test_config.py @@ -0,0 +1,77 @@ +import configparser +from pathlib import Path +from unittest.mock import mock_open, patch + +import pytest + +from opik.config import OpikConfig + + +@pytest.fixture(autouse=True) +def mock_env_and_file(monkeypatch): + monkeypatch.delenv("OPIK_API_KEY", raising=False) + monkeypatch.delenv("OPIK_WORKSPACE", raising=False) + monkeypatch.delenv("OPIK_URL_OVERRIDE", raising=False) + + with patch("builtins.open", side_effect=FileNotFoundError): + yield + + +@patch("builtins.open", new_callable=mock_open) +@patch("pathlib.Path.expanduser", return_value=Path("/fake/path/config.ini")) +def test_save_to_file_content(mock_expanduser, mock_open_file): + config = OpikConfig( + api_key="test_api_key", + url_override="http://test-url", + workspace="test_workspace", + ) + + config.save_to_file() + + # Assert the file was opened with the correct path and mode + mock_open_file.assert_called_once() + assert Path(mock_open_file.call_args_list[0].args[0]) == Path( + "/fake/path/config.ini" + ) + assert mock_open_file.call_args_list[0].kwargs == { + "encoding": "utf-8", + "mode": "w+", + } + + # Get the file handle to check what was written + handle = mock_open_file() + + # Collect all the written content + written_content = "".join(call.args[0] for call in handle.write.call_args_list) + + # Create a config parser to parse the written content + parsed_config = configparser.ConfigParser() + parsed_config.read_string(written_content) + + # Assert the correct content was written to the file + assert parsed_config["opik"]["url_override"] == "http://test-url" + assert parsed_config["opik"]["workspace"] == "test_workspace" + assert parsed_config["opik"]["api_key"] == "test_api_key" + + +@patch("builtins.open", new_callable=mock_open) +@patch("pathlib.Path.expanduser", return_value=Path("/fake/path/config.ini")) +def test_save_to_file_without_api_key(mock_expanduser, mock_open_file): + config = OpikConfig(url_override="http://test-url", workspace="test_workspace") + + config.save_to_file() + + # Get the file handle to check what was written + handle = mock_open_file() + + # Collect all the written content + written_content = "".join(call.args[0] for call in handle.write.call_args_list) + + # Create a config parser to parse the written content + parsed_config = configparser.ConfigParser() + parsed_config.read_string(written_content) + + # Assert the correct content was written to the file, without the API key + assert parsed_config["opik"]["url_override"] == "http://test-url" + assert parsed_config["opik"]["workspace"] == "test_workspace" + assert "api_key" not in parsed_config["opik"] diff --git a/sdks/python/tests/unit/test_opik_configure.py b/sdks/python/tests/unit/test_opik_configure.py index 71c5998f1f..32bc14dd82 100644 --- a/sdks/python/tests/unit/test_opik_configure.py +++ b/sdks/python/tests/unit/test_opik_configure.py @@ -1,97 +1,1089 @@ -from opik import configure +from pathlib import Path +from unittest.mock import MagicMock, Mock, patch + +import httpx import pytest +from opik.config import ( + OPIK_BASE_URL_CLOUD, + OPIK_BASE_URL_LOCAL, + OPIK_WORKSPACE_DEFAULT_NAME, + OpikConfig, +) from opik.exceptions import ConfigurationError +from opik.opik_configure import ( + _ask_for_api_key, + _ask_for_url, + _ask_for_workspace, + _configure_cloud, + _configure_local, + _get_api_key, + _get_workspace, + _update_config, + ask_user_for_approval, + get_default_workspace, + is_api_key_correct, + is_instance_active, + is_workspace_name_correct, +) -@pytest.mark.skip -@pytest.mark.parametrize( - "api_key, url, workspace, local, should_raise", - [ - ( - None, - "http://example.com", - "workspace1", - True, - False, - ), # Missing api_key, local=True - ( - None, - "http://example.com", - "workspace1", - False, - True, - ), # Missing api_key, local=False - ("apikey123", None, "workspace1", True, True), # Missing url, local=True - ("apikey123", None, "workspace1", False, True), # Missing url, local=False - ( - "apikey123", - "http://example.com", - None, - True, - True, - ), # Missing workspace, local=True - ( - "apikey123", - "http://example.com", - None, - False, - True, - ), # Missing workspace, local=False - (None, None, "workspace1", True, True), # Missing api_key and url, local=True - (None, None, "workspace1", False, True), # Missing api_key and url, local=False - ( - None, - "http://example.com", - None, - True, - True, - ), # Missing api_key and workspace, local=True - ( - None, - "http://example.com", - None, - False, - True, - ), # Missing api_key and workspace, local=False - ("apikey123", None, None, True, True), # Missing url and workspace, local=True - ( - "apikey123", - None, +@pytest.fixture(autouse=True) +def mock_env_and_file(monkeypatch): + monkeypatch.delenv("OPIK_API_KEY", raising=False) + monkeypatch.delenv("OPIK_WORKSPACE", raising=False) + monkeypatch.delenv("OPIK_URL_OVERRIDE", raising=False) + + with patch("builtins.open", side_effect=FileNotFoundError): + yield + + +class TestIsInstanceActive: + @pytest.mark.parametrize( + "status_code, expected_result", + [ + (200, True), + (404, False), + (500, False), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_is_instance_active(self, mock_httpx_client, status_code, expected_result): + """ + Test various HTTP status code responses to check if the instance is active. + """ + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = status_code + + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + url = "http://example.com" + result = is_instance_active(url) + + assert result == expected_result + + @patch("opik.opik_configure.httpx.Client") + def test_is_instance_active_timeout(self, mock_httpx_client): + """ + Test that a connection timeout results in False being returned. + """ + mock_client_instance = MagicMock() + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.side_effect = httpx.ConnectTimeout("timeout") + + mock_httpx_client.return_value = mock_client_instance + + url = "http://example.com" + result = is_instance_active(url) + + assert result is False + + @patch("opik.opik_configure.httpx.Client") + def test_is_instance_active_general_exception(self, mock_httpx_client): + """ + Test that any general exception results in False being returned. + """ + mock_client_instance = MagicMock() + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.side_effect = Exception("Unexpected error") + + mock_httpx_client.return_value = mock_client_instance + + url = "http://example.com" + result = is_instance_active(url) + + assert result is False + + +class TestIsWorkspaceNameCorrect: + @pytest.mark.parametrize( + "api_key, workspace, workspace_names, expected_result", + [ + ("valid_api_key", "correct_workspace", ["correct_workspace"], True), + ("valid_api_key", "incorrect_workspace", ["other_workspace"], False), + ("valid_api_key", "empty_workspace", [], False), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_workspace_valid_api_key( + self, mock_httpx_client, api_key, workspace, workspace_names, expected_result + ): + """ + Test cases with valid API keys and workspace verification. + These tests simulate different workspace existence conditions. + """ + # Mock the HTTP response for valid API key cases + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"workspaceNames": workspace_names} + + # Mock the context manager behavior + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + result = is_workspace_name_correct(api_key, workspace) + assert result == expected_result + + @pytest.mark.parametrize( + "status_code, response_text", + [(500, "Internal Server Error"), (404, "Not Found"), (403, "Forbidden")], + ) + @patch("opik.opik_configure.httpx.Client") + def test_workspace_non_200_response( + self, mock_httpx_client, status_code, response_text + ): + """ + Test cases where the API responds with a non-200 status code. + These responses should raise a ConnectionError. + """ + # Mock the HTTP response for non-200 status code cases + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = status_code + mock_response.text = response_text + + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + api_key = "valid_api_key" + workspace = "any_workspace" + + with pytest.raises(ConnectionError): + is_workspace_name_correct(api_key, workspace) + + @pytest.mark.parametrize( + "exception", + [ + (httpx.RequestError("Timeout", request=MagicMock())), + (Exception("Unexpected error")), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_workspace_request_exceptions(self, mock_httpx_client, exception): + """ + Test cases where an exception is raised during the HTTP request. + These cases should raise a ConnectionError with the appropriate message. + """ + # Mock the HTTP request to raise an exception + mock_client_instance = MagicMock() + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.side_effect = exception + + mock_httpx_client.return_value = mock_client_instance + + api_key = "valid_api_key" + workspace = "any_workspace" + + # Check that the appropriate ConnectionError is raised + with pytest.raises(ConnectionError): + is_workspace_name_correct(api_key, workspace) + + +class TestIsApiKeyCorrect: + @pytest.mark.parametrize( + "status_code, expected_result", + [ + (200, True), + (401, False), + (403, False), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_is_api_key_correct(self, mock_httpx_client, status_code, expected_result): + """ + Test valid, invalid, and forbidden API key scenarios by simulating HTTP status codes. + """ + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = status_code + + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + api_key = "dummy_api_key" + result = is_api_key_correct(api_key) + + assert result == expected_result + + @pytest.mark.parametrize( + "status_code, response_text", + [(500, "Internal Server Error")], + ) + @patch("opik.opik_configure.httpx.Client") + def test_is_api_key_correct_non_200_response( + self, mock_httpx_client, status_code, response_text + ): + """ + Test that a non-200, 401, or 403 response raises a ConnectionError. + """ + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = status_code + mock_response.text = response_text + + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + api_key = "dummy_api_key" + + with pytest.raises(ConnectionError): + is_api_key_correct(api_key) + + @pytest.mark.parametrize( + "exception", + [ + (httpx.RequestError("Timeout", request=MagicMock())), + (Exception("Unexpected error")), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_is_api_key_correct_exceptions(self, mock_httpx_client, exception): + """ + Test that RequestError and general exceptions are properly raised as ConnectionError. + """ + mock_client_instance = MagicMock() + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.__exit__.return_value = False + mock_client_instance.get.side_effect = exception + + mock_httpx_client.return_value = mock_client_instance + + api_key = "dummy_api_key" + + with pytest.raises(ConnectionError): + is_api_key_correct(api_key) + + +class TestGetDefaultWorkspace: + @pytest.mark.parametrize( + "status_code, response_json, expected_result", + [ + (200, {"defaultWorkspaceName": "workspace1"}, "workspace1"), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_get_default_workspace_success( + self, mock_httpx_client, status_code, response_json, expected_result + ): + """ + Test successful retrieval of the default workspace name. + """ + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = status_code + mock_response.json.return_value = response_json + + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + api_key = "valid_api_key" + result = get_default_workspace(api_key) + assert result == expected_result + + @pytest.mark.parametrize( + "status_code, response_text", + [ + (500, "Internal Server Error"), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_get_default_workspace_non_200_status( + self, mock_httpx_client, status_code, response_text + ): + """ + Test that non-200 status codes raise a ConnectionError. + """ + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = status_code + mock_response.text = response_text + + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + api_key = "valid_api_key" + with pytest.raises(ConnectionError): + get_default_workspace(api_key) + + @pytest.mark.parametrize( + "response_json", + [ + {}, + {"otherKey": "value"}, None, - False, - True, - ), # Missing url and workspace, local=False - (None, None, None, True, True), # All missing, local=True - (None, None, None, False, True), # All missing, local=False - ( - "apikey123", - "http://example.com", - "workspace1", - True, - False, - ), # All present, local=True - ( - "apikey123", - "http://example.com", - "workspace1", - False, - False, - ), # All present, local=False - ], -) -def test_login__force_new_settings__fail(api_key, url, workspace, local, should_raise): - if should_raise: + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_get_default_workspace_missing_key(self, mock_httpx_client, response_json): + """ + Test that missing 'defaultWorkspaceName' in the response raises a ConnectionError. + """ + mock_client_instance = MagicMock() + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = response_json + + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.get.return_value = mock_response + mock_httpx_client.return_value = mock_client_instance + + api_key = "valid_api_key" + with pytest.raises(ConnectionError): + get_default_workspace(api_key) + + @pytest.mark.parametrize( + "exception", + [ + httpx.RequestError("Timeout", request=MagicMock()), + Exception("Unexpected error"), + ], + ) + @patch("opik.opik_configure.httpx.Client") + def test_get_default_workspace_exceptions(self, mock_httpx_client, exception): + """ + Test that network and unexpected exceptions are raised as ConnectionError. + """ + mock_client_instance = MagicMock() + mock_client_instance.__enter__.return_value = mock_client_instance + mock_client_instance.get.side_effect = exception + mock_httpx_client.return_value = mock_client_instance + + api_key = "valid_api_key" + + with pytest.raises(ConnectionError): + get_default_workspace(api_key) + + +class TestUpdateConfig: + @patch("opik.opik_configure.opik.config.OpikConfig") + @patch("opik.opik_configure.opik.config.update_session_config") + def test_update_config_success(self, mock_update_session_config, mock_opik_config): + """ + Test successful update of the config and session. + """ + mock_config_instance = MagicMock() + mock_opik_config.return_value = mock_config_instance + + api_key = "dummy_api_key" + url = "http://example.com" + workspace = "workspace1" + + _update_config(api_key, url, workspace) + + # Ensure config object is created and saved + mock_opik_config.assert_called_once_with( + api_key=api_key, + url_override=url, + workspace=workspace, + ) + mock_config_instance.save_to_file.assert_called_once() + + # Ensure session config is updated + mock_update_session_config.assert_any_call("api_key", api_key) + mock_update_session_config.assert_any_call("url_override", url) + mock_update_session_config.assert_any_call("workspace", workspace) + + @patch("opik.opik_configure.opik.config.OpikConfig") + @patch("opik.opik_configure.opik.config.update_session_config") + def test_update_config_raises_exception( + self, mock_update_session_config, mock_opik_config + ): + """ + Test that ConfigurationError is raised when an exception occurs. + """ + mock_opik_config.side_effect = Exception("Unexpected error") + + api_key = "dummy_api_key" + url = "http://example.com" + workspace = "workspace1" + + with pytest.raises(ConfigurationError, match="Failed to update configuration."): + _update_config(api_key, url, workspace) + + # Ensure save_to_file is not called due to the exception + mock_update_session_config.assert_not_called() + + @patch("opik.opik_configure.opik.config.OpikConfig") + @patch("opik.opik_configure.opik.config.update_session_config") + def test_update_config_session_update_failure( + self, mock_update_session_config, mock_opik_config + ): + """ + Test that ConfigurationError is raised if updating the session configuration fails. + """ + mock_config_instance = MagicMock() + mock_opik_config.return_value = mock_config_instance + mock_update_session_config.side_effect = Exception("Session update failed") + + api_key = "dummy_api_key" + url = "http://example.com" + workspace = "workspace1" + + with pytest.raises(ConfigurationError, match="Failed to update configuration."): + _update_config(api_key, url, workspace) + + # Ensure config object is created and saved + mock_opik_config.assert_called_once_with( + api_key=api_key, + url_override=url, + workspace=workspace, + ) + mock_config_instance.save_to_file.assert_called_once() + + +class TestAskForUrl: + @patch("builtins.input", side_effect=["http://valid-url.com"]) + @patch("opik.opik_configure.is_instance_active", return_value=True) + def test_ask_for_url_success(self, mock_is_instance_active, mock_input): + """ + Test successful input of a valid Opik URL. + """ + result = _ask_for_url() + assert result == "http://valid-url.com" + mock_is_instance_active.assert_called_once_with("http://valid-url.com") + + @patch("builtins.input", side_effect=["http://invalid-url.com"] * 3) + @patch("opik.opik_configure.is_instance_active", return_value=False) + def test_ask_for_url_all_retries_fail(self, mock_is_instance_active, mock_input): + """ + Test that after 3 failed attempts, a ConfigurationError is raised. + """ + with pytest.raises(ConfigurationError, match="Cannot use the URL provided"): + _ask_for_url() + + assert mock_is_instance_active.call_count == 3 + + @patch( + "builtins.input", side_effect=["http://invalid-url.com", "http://valid-url.com"] + ) + @patch("opik.opik_configure.is_instance_active", side_effect=[False, True]) + def test_ask_for_url_success_on_second_try( + self, mock_is_instance_active, mock_input + ): + """ + Test that the URL is successfully returned on the second attempt after the first failure. + """ + result = _ask_for_url() + assert result == "http://valid-url.com" + assert mock_is_instance_active.call_count == 2 + + @patch( + "builtins.input", + side_effect=[ + "http://invalid-url.com", + "http://invalid-url-2.com", + "http://valid-url.com", + ], + ) + @patch("opik.opik_configure.is_instance_active", side_effect=[False, False, True]) + def test_ask_for_url_success_on_third_try( + self, mock_is_instance_active, mock_input + ): + """ + Test that the URL is successfully returned on the third attempt after two failures. + """ + result = _ask_for_url() + assert result == "http://valid-url.com" + assert mock_is_instance_active.call_count == 3 + + @patch("builtins.input", side_effect=["http://invalid-url.com"] * 3) + @patch("opik.opik_configure.is_instance_active", return_value=False) + @patch("opik.opik_configure.LOGGER.error") + def test_ask_for_url_logging( + self, mock_logger_error, mock_is_instance_active, mock_input + ): + """ + Test that errors are logged when the URL is not accessible. + """ + with pytest.raises(ConfigurationError): + _ask_for_url() + + assert mock_logger_error.call_count == 3 + mock_logger_error.assert_called_with( + f"Opik is not accessible at http://invalid-url.com. Please try again, the URL should follow a format similar to {OPIK_BASE_URL_LOCAL}" + ) + + +class TestAskForApiKey: + @patch("opik.opik_configure.getpass.getpass", return_value="valid_api_key") + @patch("opik.opik_configure.is_api_key_correct", return_value=True) + def test_ask_for_api_key_success(self, mock_is_api_key_correct, mock_getpass): + """ + Test successful entry of a valid API key. + """ + result = _ask_for_api_key() + assert result == "valid_api_key" + mock_is_api_key_correct.assert_called_once_with("valid_api_key") + + @patch("opik.opik_configure.getpass.getpass", return_value="invalid_api_key") + @patch("opik.opik_configure.is_api_key_correct", return_value=False) + def test_ask_for_api_key_all_retries_fail( + self, mock_is_api_key_correct, mock_getpass + ): + """ + Test that after 3 invalid API key attempts, a ConfigurationError is raised. + """ + with pytest.raises(ConfigurationError, match="API key is incorrect."): + _ask_for_api_key() + + assert mock_is_api_key_correct.call_count == 3 + + @patch( + "opik.opik_configure.getpass.getpass", side_effect=["invalid_key", "valid_key"] + ) + @patch("opik.opik_configure.is_api_key_correct", side_effect=[False, True]) + def test_ask_for_api_key_success_on_second_try( + self, mock_is_api_key_correct, mock_getpass + ): + """ + Test that the correct API key is entered on the second attempt after the first one is invalid. + """ + result = _ask_for_api_key() + assert result == "valid_key" + assert mock_is_api_key_correct.call_count == 2 + + @patch( + "opik.opik_configure.getpass.getpass", + side_effect=["invalid_key1", "invalid_key2", "valid_key"], + ) + @patch("opik.opik_configure.is_api_key_correct", side_effect=[False, False, True]) + def test_ask_for_api_key_success_on_third_try( + self, mock_is_api_key_correct, mock_getpass + ): + """ + Test that the correct API key is entered on the third attempt after two invalid attempts. + """ + result = _ask_for_api_key() + assert result == "valid_key" + assert mock_is_api_key_correct.call_count == 3 + + +class TestAskForWorkspace: + @patch("builtins.input", return_value="valid_workspace") + @patch("opik.opik_configure.is_workspace_name_correct", return_value=True) + def test_ask_for_workspace_success( + self, mock_is_workspace_name_correct, mock_input + ): + """ + Test successful entry of a valid workspace name. + """ + api_key = "valid_api_key" + result = _ask_for_workspace(api_key) + assert result == "valid_workspace" + mock_is_workspace_name_correct.assert_called_once_with( + api_key, "valid_workspace" + ) + + @patch("builtins.input", return_value="invalid_workspace") + @patch("opik.opik_configure.is_workspace_name_correct", return_value=False) + def test_ask_for_workspace_all_retries_fail( + self, mock_is_workspace_name_correct, mock_input + ): + """ + Test that after 3 invalid workspace name attempts, a ConfigurationError is raised. + """ + api_key = "valid_api_key" + + with pytest.raises( + ConfigurationError, + match="User does not have access to the workspaces provided.", + ): + _ask_for_workspace(api_key) + + assert mock_is_workspace_name_correct.call_count == 3 + + @patch("builtins.input", side_effect=["invalid_workspace", "valid_workspace"]) + @patch("opik.opik_configure.is_workspace_name_correct", side_effect=[False, True]) + def test_ask_for_workspace_success_on_second_try( + self, mock_is_workspace_name_correct, mock_input + ): + """ + Test that the workspace name is successfully entered on the second attempt after the first one is invalid. + """ + api_key = "valid_api_key" + result = _ask_for_workspace(api_key) + assert result == "valid_workspace" + assert mock_is_workspace_name_correct.call_count == 2 + + @patch( + "builtins.input", + side_effect=["invalid_workspace1", "invalid_workspace2", "valid_workspace"], + ) + @patch( + "opik.opik_configure.is_workspace_name_correct", + side_effect=[False, False, True], + ) + def test_ask_for_workspace_success_on_third_try( + self, mock_is_workspace_name_correct, mock_input + ): + """ + Test that the workspace name is successfully entered on the third attempt after two invalid attempts. + """ + api_key = "valid_api_key" + result = _ask_for_workspace(api_key) + assert result == "valid_workspace" + assert mock_is_workspace_name_correct.call_count == 3 + + +class TestAskUserForApproval: + @patch("builtins.input", return_value="Y") + def test_user_approves_with_y(self, mock_input): + """ + Test that 'Y' returns True for approval. + """ + result = ask_user_for_approval("Do you approve?") + assert result is True + + @patch("builtins.input", return_value="YES") + def test_user_approves_with_yes(self, mock_input): + """ + Test that 'YES' returns True for approval. + """ + result = ask_user_for_approval("Do you approve?") + assert result is True + + @patch("builtins.input", return_value="") + def test_user_approves_with_empty_input(self, mock_input): + """ + Test that empty input returns True for approval. + """ + result = ask_user_for_approval("Do you approve?") + assert result is True + + @patch("builtins.input", return_value="N") + def test_user_disapproves_with_n(self, mock_input): + """ + Test that 'N' returns False for disapproval. + """ + result = ask_user_for_approval("Do you disapprove?") + assert result is False + + @patch("builtins.input", return_value="NO") + def test_user_disapproves_with_no(self, mock_input): + """ + Test that 'NO' returns False for disapproval. + """ + result = ask_user_for_approval("Do you disapprove?") + assert result is False + + @patch("builtins.input", side_effect=["INVALID", "Y"]) + @patch("opik.opik_configure.LOGGER.error") + def test_user_enters_invalid_choice_then_approves( + self, mock_logger_error, mock_input + ): + """ + Test that invalid input triggers error logging and prompts again until valid input is entered. + """ + result = ask_user_for_approval("Do you approve?") + assert result is True + mock_logger_error.assert_called_once_with("Wrong choice. Please try again.") + + @patch("builtins.input", side_effect=["INVALID", "NO"]) + @patch("opik.opik_configure.LOGGER.error") + def test_user_enters_invalid_choice_then_disapproves( + self, mock_logger_error, mock_input + ): + """ + Test that invalid input triggers error logging and prompts again until valid input is entered. + """ + result = ask_user_for_approval("Do you disapprove?") + assert result is False + mock_logger_error.assert_called_once_with("Wrong choice. Please try again.") + + +class TestGetApiKey: + @patch("opik.opik_configure._ask_for_api_key", return_value="new_api_key") + def test_get_api_key_force_ask(self, mock_ask_for_api_key): + """ + Test that when force=True and no API key is provided, the user is asked for an API key. + """ + current_config = OpikConfig(api_key=None) + api_key, needs_update = _get_api_key( + api_key=None, current_config=current_config, force=True + ) + + assert api_key == "new_api_key" + assert needs_update is True + mock_ask_for_api_key.assert_called_once() + + @patch("opik.opik_configure._ask_for_api_key", return_value="new_api_key") + def test_get_api_key_ask_for_missing_key(self, mock_ask_for_api_key): + """ + Test that when no API key is provided and none is present in the config, the user is asked for an API key. + """ + current_config = OpikConfig(api_key=None) + api_key, needs_update = _get_api_key( + api_key=None, current_config=current_config, force=False + ) + + assert api_key == "new_api_key" + assert needs_update is True + mock_ask_for_api_key.assert_called_once() + + def test_get_api_key_use_config_key(self): + """ + Test that the API key is taken from the current config when provided and force=False. + """ + current_config = OpikConfig(api_key="config_api_key") + api_key, needs_update = _get_api_key( + api_key=None, current_config=current_config, force=False + ) + + assert api_key == "config_api_key" + assert needs_update is False + + def test_get_api_key_provided_key(self): + """ + Test that the user-provided API key is used directly if it's passed in. + """ + current_config = OpikConfig(api_key="config_api_key") + api_key, needs_update = _get_api_key( + api_key="user_provided_api_key", current_config=current_config, force=False + ) + + assert api_key == "user_provided_api_key" + assert needs_update is False + + +class TestGetWorkspace: + @patch("opik.opik_configure.is_workspace_name_correct", return_value=True) + def test_get_workspace_user_provided_valid(self, mock_is_workspace_name_correct): + """ + Test that the workspace provided by the user is valid and used. + """ + current_config = OpikConfig(workspace="existing_workspace") + workspace, needs_update = _get_workspace( + workspace="new_workspace", + api_key="valid_api_key", + current_config=current_config, + force=False, + ) + + assert workspace == "new_workspace" + assert needs_update is True + mock_is_workspace_name_correct.assert_called_once_with( + "valid_api_key", "new_workspace" + ) + + @patch("opik.opik_configure.is_workspace_name_correct", return_value=False) + def test_get_workspace_user_provided_invalid(self, mock_is_workspace_name_correct): + """ + Test that a ConfigurationError is raised if the user-provided workspace is invalid. + """ + current_config = OpikConfig(workspace="existing_workspace") + with pytest.raises(ConfigurationError): - configure( - api_key=api_key, - url=url, - workspace=workspace, - force=True, - use_local=local, + _get_workspace( + workspace="invalid_workspace", + api_key="valid_api_key", + current_config=current_config, + force=False, ) - else: - # No exception should be raised - configure( - api_key=api_key, url=url, workspace=workspace, force=True, use_local=local + + mock_is_workspace_name_correct.assert_called_once_with( + "valid_api_key", "invalid_workspace" + ) + + def test_get_workspace_use_config(self): + """ + Test that the workspace from the current config is used when no workspace is provided and not forced. + """ + current_config = OpikConfig(workspace="configured_workspace") + workspace, needs_update = _get_workspace( + workspace=None, + api_key="valid_api_key", + current_config=current_config, + force=False, + ) + + assert workspace == "configured_workspace" + assert needs_update is False + + @patch( + "opik.opik_configure.get_default_workspace", return_value="default_workspace" + ) + @patch("opik.opik_configure.ask_user_for_approval", return_value=True) + def test_get_workspace_accept_default( + self, mock_ask_user_for_approval, mock_get_default_workspace + ): + """ + Test that the user accepts the default workspace. + """ + current_config = OpikConfig(workspace=OPIK_WORKSPACE_DEFAULT_NAME) + workspace, needs_update = _get_workspace( + workspace=None, + api_key="valid_api_key", + current_config=current_config, + force=False, + ) + + assert workspace == "default_workspace" + assert needs_update is True + mock_get_default_workspace.assert_called_once_with("valid_api_key") + mock_ask_user_for_approval.assert_called_once_with( + 'Do you want to use "default_workspace" workspace? (Y/n)' + ) + + @patch( + "opik.opik_configure.get_default_workspace", return_value="default_workspace" + ) + @patch("opik.opik_configure.ask_user_for_approval", return_value=False) + @patch("opik.opik_configure._ask_for_workspace", return_value="new_workspace") + def test_get_workspace_choose_different( + self, + mock_ask_for_workspace, + mock_ask_user_for_approval, + mock_get_default_workspace, + ): + """ + Test that the user declines the default workspace and chooses a new one. + """ + current_config = OpikConfig(workspace=OPIK_WORKSPACE_DEFAULT_NAME) + workspace, needs_update = _get_workspace( + workspace=None, + api_key="valid_api_key", + current_config=current_config, + force=False, + ) + + assert workspace == "new_workspace" + assert needs_update is True + mock_get_default_workspace.assert_called_once_with("valid_api_key") + mock_ask_user_for_approval.assert_called_once_with( + 'Do you want to use "default_workspace" workspace? (Y/n)' + ) + mock_ask_for_workspace.assert_called_once_with(api_key="valid_api_key") + + +class TestConfigureCloud: + @patch("opik.opik_configure._get_api_key", return_value=("valid_api_key", True)) + @patch("opik.opik_configure._get_workspace", return_value=("valid_workspace", True)) + @patch("opik.opik_configure._update_config") + def test_configure_cloud_with_update( + self, mock_update_config, mock_get_workspace, mock_get_api_key + ): + """ + Test that the configuration is updated when both API key and workspace require updates. + """ + _configure_cloud(api_key=None, workspace=None, force=False) + + mock_get_api_key.assert_called_once() + mock_get_workspace.assert_called_once() + mock_update_config.assert_called_once_with( + api_key="valid_api_key", + url=OPIK_BASE_URL_CLOUD, + workspace="valid_workspace", + ) + + @patch("opik.opik_configure._get_api_key", return_value=("valid_api_key", False)) + @patch( + "opik.opik_configure._get_workspace", return_value=("valid_workspace", False) + ) + @patch("opik.opik_configure.LOGGER.info") + @patch("opik.opik_configure.opik.config.OpikConfig") + @patch("opik.opik_configure._update_config") + def test_configure_cloud_no_update_needed( + self, + mock_update_config, + mock_opik_config, + mock_logger_info, + mock_get_workspace, + mock_get_api_key, + ): + """ + Test that no configuration update happens when both API key and workspace are already set. + """ + # Mock the config file path to return a specific path + mock_config_instance = MagicMock() + mock_config_instance.config_file_fullpath = Path("/some/path/.opik.config") + mock_opik_config.return_value = mock_config_instance + + # Call the function + _configure_cloud( + api_key="valid_api_key", workspace="valid_workspace", force=False + ) + + # Ensure API key and workspace were checked + mock_get_api_key.assert_called_once() + mock_get_workspace.assert_called_once() + + # Check config file wasn't overwritten + mock_update_config.assert_not_called() + + # Check the logging message + mock_logger_info.assert_called_with( + "Opik is already configured. You can check the settings by viewing the config file at %s", + Path("/some/path/.opik.config"), + ) + + @patch("opik.opik_configure._get_api_key", return_value=("new_api_key", True)) + @patch( + "opik.opik_configure._get_workspace", + return_value=("configured_workspace", False), + ) + @patch("opik.opik_configure._update_config") + def test_configure_cloud_api_key_updated( + self, mock_update_config, mock_get_workspace, mock_get_api_key + ): + """ + Test that the configuration is updated when only the API key changes. + """ + _configure_cloud(api_key=None, workspace="configured_workspace", force=False) + + mock_get_api_key.assert_called_once() + mock_get_workspace.assert_called_once() + mock_update_config.assert_called_once_with( + api_key="new_api_key", + url=OPIK_BASE_URL_CLOUD, + workspace="configured_workspace", + ) + + @patch("opik.opik_configure._get_api_key", return_value=("valid_api_key", False)) + @patch("opik.opik_configure._get_workspace", return_value=("new_workspace", True)) + @patch("opik.opik_configure._update_config") + def test_configure_cloud_workspace_updated( + self, mock_update_config, mock_get_workspace, mock_get_api_key + ): + """ + Test that the configuration is updated when only the workspace changes. + """ + _configure_cloud(api_key="valid_api_key", workspace=None, force=False) + + mock_get_api_key.assert_called_once() + mock_get_workspace.assert_called_once() + mock_update_config.assert_called_once_with( + api_key="valid_api_key", url=OPIK_BASE_URL_CLOUD, workspace="new_workspace" + ) + + +class TestConfigureLocal: + @patch( + "opik.opik_configure._ask_for_url", return_value="http://user-provided-url.com" + ) + @patch("opik.opik_configure.is_instance_active", return_value=False) + @patch("opik.opik_configure._update_config") + def test_configure_local_asks_for_url( + self, mock_update_config, mock_is_instance_active, mock_ask_for_url + ): + """ + Test that the function asks for a URL if no local instance is active and no URL is provided. + """ + _configure_local(url=None, force=False) + + mock_ask_for_url.assert_called_once() + mock_update_config.assert_called_once_with( + api_key=None, + url="http://user-provided-url.com", + workspace=OPIK_WORKSPACE_DEFAULT_NAME, + ) + + @patch("opik.opik_configure._ask_for_url") + @patch("opik.opik_configure.is_instance_active", return_value=True) + @patch("opik.opik_configure._update_config") + def test_configure_local_with_provided_url( + self, mock_update_config, mock_is_instance_active, mock_ask_for_url + ): + """ + Test that the function configures the provided URL if it is active. + """ + _configure_local(url="http://custom-local-instance.com", force=False) + + mock_ask_for_url.assert_not_called() + mock_is_instance_active.assert_called_once_with( + "http://custom-local-instance.com" + ) + mock_update_config.assert_called_once_with( + api_key=None, + url="http://custom-local-instance.com", + workspace=OPIK_WORKSPACE_DEFAULT_NAME, + ) + + @patch("opik.opik_configure._ask_for_url") + @patch("opik.opik_configure.is_instance_active", return_value=True) + @patch("opik.opik_configure.opik.config.OpikConfig") + @patch("opik.opik_configure.LOGGER.info") + def test_configure_local_no_update_needed( + self, + mock_logger_info, + mock_opik_config, + mock_is_instance_active, + mock_ask_for_url, + ): + """ + Test that no update happens if the local instance is already configured and force=False. + """ + mock_config_instance = MagicMock() + mock_config_instance.url_override = OPIK_BASE_URL_LOCAL + mock_opik_config.return_value = mock_config_instance + + _configure_local(url=None, force=False) + + mock_ask_for_url.assert_not_called() + mock_is_instance_active.assert_called_once_with(OPIK_BASE_URL_LOCAL) + mock_logger_info.assert_called_once_with( + f"Opik is already configured to local instance at {OPIK_BASE_URL_LOCAL}." + ) + + @patch("opik.opik_configure.ask_user_for_approval", return_value=True) + @patch("opik.opik_configure.is_instance_active", return_value=True) + @patch("opik.opik_configure._update_config") + def test_configure_local_uses_local_instance( + self, mock_update_config, mock_is_instance_active, mock_ask_user_for_approval + ): + """ + Test that the function configures the local instance when found and user approves. + """ + _configure_local(url=None, force=False) + + mock_ask_user_for_approval.assert_called_once_with( + f"Found local Opik instance on: {OPIK_BASE_URL_LOCAL}, do you want to use it? (Y/n)" + ) + mock_update_config.assert_called_once_with( + api_key=None, url=OPIK_BASE_URL_LOCAL, workspace=OPIK_WORKSPACE_DEFAULT_NAME + ) + + @patch("opik.opik_configure.ask_user_for_approval", return_value=False) + @patch( + "opik.opik_configure._ask_for_url", return_value="http://user-provided-url.com" + ) + @patch("opik.opik_configure.is_instance_active", return_value=True) + @patch("opik.opik_configure._update_config") + def test_configure_local_user_declines_local_instance( + self, + mock_update_config, + mock_is_instance_active, + mock_ask_for_url, + mock_ask_user_for_approval, + ): + """ + Test that if the user declines using the local instance, they are prompted for a URL. + """ + _configure_local(url=None, force=False) + + mock_ask_user_for_approval.assert_called_once_with( + f"Found local Opik instance on: {OPIK_BASE_URL_LOCAL}, do you want to use it? (Y/n)" + ) + mock_ask_for_url.assert_called_once() + mock_update_config.assert_called_once_with( + api_key=None, + url="http://user-provided-url.com", + workspace=OPIK_WORKSPACE_DEFAULT_NAME, )