feat: POC automation test

JIRA: QA-23477
risk: nonprod

gooddata-sdk/integration_tests/env.py

@@ -0,0 +1,9 @@
+# (C) 2024 GoodData Corporation
+# env.py
+import os
+
+# Define environment variables
+HOST = os.getenv("GOODDATA_HOST", "xxx")
+TOKEN = os.getenv("GOODDATA_TOKEN", "xxx")
+DATASOURCE_ID = os.getenv("DATASOURCE_ID", "xxx")
+WORKSPACE_ID = "xxx"

gooddata-sdk/integration_tests/scripts/aiChat.py

@@ -0,0 +1,99 @@
+# (C) 2024 GoodData Corporation
+
+import os
+import sys
+from pprint import pprint
+
+import gooddata_api_client
+import pytest
+from gooddata_api_client.api import smart_functions_api
+from gooddata_api_client.model.chat_history_request import ChatHistoryRequest
+from gooddata_api_client.model.chat_history_result import ChatHistoryResult
+from gooddata_api_client.model.chat_request import ChatRequest
+from gooddata_api_client.model.chat_result import ChatResult
+
+# Add the root directory to sys.path
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from env import HOST, TOKEN, WORKSPACE_ID
+
+
+@pytest.fixture(scope="module")
+def api_client():
+    configuration = gooddata_api_client.Configuration(host=HOST)
+    configuration.access_token = TOKEN
+    with gooddata_api_client.ApiClient(configuration) as api_client:
+        yield api_client
+
+
+class GoodDataAiChatApp:
+    def __init__(self, api_client, workspace_id):
+        self.api_instance = smart_functions_api.SmartFunctionsApi(api_client)
+        self.workspace_id = workspace_id
+
+    async def ask_question(self, question: str):
+        chat_request = ChatRequest(question=question)
+        return self.api_instance.ai_chat(self.workspace_id, chat_request)
+
+    async def chat_history(self, chat_history_interaction_id: int, user_feedback: str):
+        chat_history_request = ChatHistoryRequest(
+            chat_history_interaction_id=chat_history_interaction_id,
+            user_feedback=user_feedback,
+        )
+        return self.api_instance.ai_chat_history(self.workspace_id, chat_history_request)
+
+
+def set_authorization_header(api_client, token):
+    api_client.default_headers["Authorization"] = f"Bearer {token}"
+
+
+def handle_api_response(api_response):
+    # Print the raw response
+    pprint(api_response.to_dict())
+    # Assert that the response is not empty
+    assert api_response.routing is not None, "Routing should not be None"
+    assert api_response.created_visualizations is not None, "createdVisualizations should not be None"
+    assert api_response.found_objects is not None, "foundObjects should not be None"
+    assert isinstance(api_response, ChatResult), "Response is not of type ChatResult"
+
+
+@pytest.fixture(scope="module")
+def app(api_client):
+    # Initialize the GoodDataAiChatApp class
+    app = GoodDataAiChatApp(api_client, WORKSPACE_ID)
+    # Set the Authorization header
+    set_authorization_header(api_client, TOKEN)
+    return app
+
+
+@pytest.mark.asyncio
+async def test_ai_chat(app):
+    question = "generate HEADLINE showing Sum of Amount"
+
+    try:
+        api_response = await app.ask_question(question)
+        handle_api_response(api_response)
+    except gooddata_api_client.ApiException as e:
+        print(f"Exception when calling SmartFunctionsApi->ai_chat: {e}")
+        pytest.fail(f"Exception when calling SmartFunctionsApi->ai_chat: {e}\n")
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")
+        pytest.fail(f"An unexpected error occurred: {e}\n")
+
+
+@pytest.mark.asyncio
+async def test_ai_chat_history(app):
+    try:
+        api_response = await app.chat_history(1260, "POSITIVE")
+        pprint(api_response.to_dict())
+        assert isinstance(api_response, ChatHistoryResult), "Response is not of type ChatHistoryResult"
+    except gooddata_api_client.ApiException as e:
+        print(f"Exception when calling SmartFunctionsApi->ai_chat_history: {e}")
+        pytest.fail(f"Exception when calling SmartFunctionsApi->ai_chat_history: {e}\n")
+    except Exception as e:
+        print(f"An unexpected error occurred: {e}")
+        pytest.fail(f"An unexpected error occurred: {e}\n")
+
+
+if __name__ == "__main__":
+    pytest.main()

gooddata-sdk/integration_tests/scripts/ai_chat.py

@@ -0,0 +1,36 @@
+# (C) 2024 GoodData Corporation
+import os
+import sys
+
+import pytest
+
+# Add the root directory to sys.path
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from env import HOST, TOKEN, WORKSPACE_ID
+from gooddata_sdk import GoodDataSdk
+
+
+@pytest.fixture
+def test_config():
+    return {"host": HOST, "token": TOKEN, "workspace_id": WORKSPACE_ID}
+
+
+questions = [
+    "What is the number of Accounts?",
+    "What is the total of Amount?",
+]
+
+
+@pytest.mark.parametrize("question", questions)
+def test_ask_ai(test_config, question):
+    sdk = GoodDataSdk.create(host_=test_config["host"], token_=test_config["token"])
+    workspace_id = test_config["workspace_id"]
+    chat_ai_res = sdk.compute.ai_chat(workspace_id, question=question)
+
+    print(f"Chat AI response: {chat_ai_res}")
+    assert chat_ai_res is not None, "Response should not be None"
+
+
+if __name__ == "__main__":
+    pytest.main()

gooddata-sdk/integration_tests/scripts/create_ref_workspace.py

@@ -0,0 +1,16 @@
+# (C) 2024 GoodData Corporation
+from env import DATASOURCE_ID, HOST, TOKEN, WORKSPACE_ID
+from workspace_manager import createWorkspace, getDataSource, update_env_file
+
+if __name__ == "__main__":
+    test_config = {"host": HOST, "token": TOKEN}
+
+    if WORKSPACE_ID:
+        print(f"Workspace ID '{WORKSPACE_ID}' already exists. Skipping workspace creation.")
+    else:
+        workspace_id = createWorkspace(test_config)
+        dataSource = getDataSource(DATASOURCE_ID, test_config)
+        if workspace_id:
+            update_env_file(workspace_id)
+        else:
+            print("Failed to create workspace.")

gooddata-sdk/integration_tests/scripts/delete_ref_workspace.py

@@ -0,0 +1,8 @@
+# (C) 2024 GoodData Corporation
+from env import HOST, TOKEN
+from workspace_manager import deleteWorkspace
+
+if __name__ == "__main__":
+    test_config = {"host": HOST, "token": TOKEN}
+
+    deleteWorkspace(test_config)

gooddata-sdk/integration_tests/scripts/workspace_manager.py

@@ -0,0 +1,73 @@
+# (C) 2024 GoodData Corporation
+import os
+import sys
+import time
+import uuid
+
+from gooddata_sdk import CatalogWorkspace, GoodDataSdk
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+try:
+    from env import WORKSPACE_ID
+except ImportError:
+    WORKSPACE_ID = None
+
+
+def createWorkspace(test_config):
+    sdk = GoodDataSdk.create(host_=test_config["host"], token_=test_config["token"])
+
+    workspace_id = uuid.uuid4().hex
+    timestamp = int(time.time())
+    workspace_name = f"pysdk_test_{timestamp}"
+
+    workspace = CatalogWorkspace(workspace_id, workspace_name)
+    try:
+        sdk.catalog_workspace.create_or_update(workspace)
+        workspace_o = sdk.catalog_workspace.get_workspace(workspace_id)
+        assert workspace_o == workspace
+
+        print(f"Workspace '{workspace_name}' with ID '{workspace_id}' created successfully.")
+        return workspace_id
+    except Exception as e:
+        print(f"An error occurred while creating the workspace: {e}")
+        return None
+
+
+def deleteWorkspace(test_config):
+    sdk = GoodDataSdk.create(host_=test_config["host"], token_=test_config["token"])
+    try:
+        workspaces = sdk.catalog_workspace.list_workspaces()
+        for workspace in workspaces:
+            if workspace.name.startswith("pysdk_test_"):
+                sdk.catalog_workspace.delete_workspace(workspace.id)
+                print(f"Workspace '{workspace.name}' with ID '{workspace.id}' deleted successfully.")
+        remove_env_file()
+    except Exception as e:
+        print(f"An error occurred while deleting workspaces: {e}")
+
+
+def update_env_file(workspace_id):
+    with open("env.py", "a") as f:
+        f.write(f'\nWORKSPACE_ID = "{workspace_id}"\n')
+
+
+def remove_env_file():
+    try:
+        with open("env.py") as f:  # Default mode is 'r'
+            lines = f.readlines()
+        with open("env.py", "w") as f:
+            for line in lines:
+                if "WORKSPACE_ID" not in line:
+                    f.write(line)
+        print("Removed WORKSPACE_ID from env.py")
+    except Exception as e:
+        print(f"An error occurred while removing WORKSPACE_ID from env.py: {e}")
+
+
+def getDataSource(data_source_id, test_config):
+    sdk = GoodDataSdk.create(host_=test_config["host"], token_=test_config["token"])
+    data_source = sdk.catalog_data_source.get_data_source(data_source_id)
+    data_source_schema = data_source.schema
+    print(f"Data source schema: {data_source_schema}")
+    return data_source_schema

gooddata-sdk/integration_tests/scripts/workspace_manager_test.py

@@ -0,0 +1,27 @@
+# (C) 2024 GoodData Corporation
+# import pytest
+# from workspace_manager import create_workspace, delete_workspace, get_data_source, update_env_file
+# from env import DATASOURCE_ID, HOST, TOKEN
+# @pytest.fixture
+# def test_config():
+#     return {
+#         'host': HOST,
+#         'token': TOKEN
+#     }
+
+# def test_create_workspace(test_config):
+#     workspace_id = create_workspace(test_config)
+#     assert workspace_id is not None, "Workspace creation failed"
+#     update_env_file(workspace_id)
+
+# # def test_delete_workspace(test_config):
+# #     delete_workspace(test_config)
+# #     # Assuming the function prints the deletion message, we can check the output
+# #     # Here we assume that the function works correctly if no exception is raised
+
+# def test_get_data_source(test_config):
+#     schema = get_data_source(DATASOURCE_ID, test_config)
+#     assert schema is not None, "Failed to get data source schema"
+
+# if __name__ == "__main__":
+#     pytest.main()