Merge pull request #761 from aws-solutions/develop

Update to version v6.1.0

.nightswatch/functional/conftest.py

@@ -23,6 +23,7 @@
 from helpers.kendra_client import KendraClient
 from helpers.lex_client import LexClient
 from helpers.iam_client import IamClient
+from helpers.s3_client import S3Client
 from helpers.translate_client import TranslateClient
 from helpers.cloud_watch_client import CloudWatchClient
 from helpers.website_model.dom_operator import DomOperator
@@ -103,6 +104,10 @@ def translate_client(region: str) -> TranslateClient:
 def iam_client(region: str) -> IamClient:
     return IamClient(region)
 
+@pytest.fixture
+def s3_client(region: str) -> None:
+    return S3Client(region)
+
 @pytest.fixture
 def app_version(param_fetcher: ParameterFetcher) -> str:
     app_version = param_fetcher.get_deployment_version()
@@ -118,8 +123,9 @@ def skip_if_version_less_than(request, app_version):
 
 @pytest.fixture
 def cw_client(region: str, param_fetcher: ParameterFetcher) -> CloudWatchClient:
-    fulfillment_lambda_name = param_fetcher.get_fulfillment_lambda_name()
-    return CloudWatchClient(region, fulfillment_lambda_name)
+    stack_id = param_fetcher.get_stack_id()
+    stack_name = param_fetcher.stack_name
+    return CloudWatchClient(region, stack_id, stack_name)
 
 @pytest.fixture(autouse=True)
 def dom_operator():
@@ -210,3 +216,11 @@ def skip_embeddings(request, embeddings_is_enabled):
         if not embeddings_is_enabled:
             pytest.skip('Embeddings is not configured for this environment. Skipping...')
 
+
+@pytest.fixture
+def knowledge_base_model(param_fetcher: ParameterFetcher):
+    return param_fetcher.get_bedrock_knowledge_base_model()
+
+@pytest.fixture
+def content_designer_output_bucket_name(param_fetcher: ParameterFetcher):
+    return param_fetcher.get_content_designer_output_bucket_name()

.nightswatch/functional/helpers/cfn_parameter_fetcher.py

@@ -275,3 +275,40 @@ def get_lambda_hook_example_arn(self) -> str:
         examples_stack_param_fetcher = ParameterFetcher(self.region, examples_stack_name) 
         return examples_stack_param_fetcher.__get_stack_outputs('EXTCustomJSHook')
 
+
+    def get_stack_id(self) -> Optional[str]:
+        """
+        Retrieves the stack id.
+
+        Returns:
+        -------
+            The stack id.
+        """
+        response = self.cloudformation_client.describe_stacks(
+            StackName=self.stack_name
+        )
+        stack_id = response['Stacks'][0]['StackId'].split('/')[2]
+        return stack_id
+
+    def get_bedrock_knowledge_base_model(self) -> Optional[str]:
+        """
+        Retrieves the model of the Bedrock Knowledge Base from the stack parameters.
+
+        Returns:
+        -------
+            The Knowledge Base Model if found, otherwise None.
+        """
+
+        knowledge_base_model = self.__get_cfn_param('BedrockKnowledgeBaseModel')
+        return knowledge_base_model
+
+    def get_content_designer_output_bucket_name(self) -> Optional[str]:
+        """
+        Retrieves the name of the test all output bucket from the stack parameters.
+
+        Returns:
+        -------
+            The name of the test all bucket if found, otherwise None.
+        """
+
+        return self.__get_stack_outputs('ContentDesignerOutputBucket')

.nightswatch/functional/helpers/cloud_watch_client.py

@@ -21,15 +21,15 @@ class CloudWatchClient:
     Interacts with CloudWatch using Boto3.
     This class provides methods for pulling logs from log groups based on matches.
     """
-    def __init__(self, region: str, fulfillment_lambda_name: str):
+    def __init__(self, region: str, stack_id: str, stack_name: str):
         """
         Initializes the CloudWatchClient.
         :param region: The AWS region to connect to.
         :type region: st
         """
         self.client = boto3.client('logs', region_name=region)
         self.region = region
-        self.fulfillment_lambda_log_group = f'/aws/lambda/{fulfillment_lambda_name}'
+        self.fulfillment_lambda_log_group = f'/aws/lambda/{stack_name}-FulfillmentLambda-{stack_id}'
         self.start_time = int(time.time() * 1000)
 
     def __get_logs(self, log_group_name: str, start_time: int, filter_pattern: str) -> dict:

source/templates/examples/extensions/py_lambda_hooks/CanvasLMSHook/__tests__/conftest.py → .nightswatch/functional/helpers/s3_client.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python
 ######################################################################################################################
 #  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.                                                #
 #                                                                                                                    #
@@ -12,25 +11,46 @@
 #  and limitations under the License.                                                                                #
 ######################################################################################################################
 
-import os
-import pytest
 import boto3
-from moto import mock_secretsmanager
+import json
 
-@pytest.fixture(autouse=True)
-def aws_environment_variables():
-    """Mocked AWS evivronment variables such as AWS credentials and region"""
-    os.environ["AWS_ACCESS_KEY_ID"] = "mocked-aws-access-key-id"
-    os.environ["AWS_SECRET_ACCESS_KEY"] = "mocked-aws-secret-access-key"
-    os.environ["AWS_SESSION_TOKEN"] = "mocked-aws-session-token"
-    os.environ["AWS_REGION"] = "us-east-1"
-    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
-    os.environ["AWS_SDK_USER_AGENT"] = '{ "user_agent_extra": "solution/fakeID/fakeVersion" }'
-    os.environ["LOCALES"] = "en_US,es_US,fr_CA"
-    os.environ["SOLUTION_ID"] = "SO0189"
-    os.environ["SOLUTION_VERSION"] = "mock_version"
+class S3Client:
+    """
+    A Python class to interact with Amazon S3 using Boto3.
+    This class provides various methods to perform operations on S3.
+    """
 
-@pytest.fixture(scope="function")
-def mock_sm():
-    with mock_secretsmanager():
-        yield boto3.client("secretsmanager", region_name="us-east-1")
+    def __init__(self, region: str) -> None:
+        """
+        Initializes the S3Client class.
+
+        Args:
+            region (str): The AWS region to connect to.
+        Returns:
+            None.
+        Raises:
+            None.
+        """
+
+        self.s3_client = boto3.client('s3', region_name=region)
+
+    def get_file_versions_count(self, bucket_name, file_prefix):
+        """
+        Returns the number of versions for a given file in an S3 bucket.
+        
+        Args:
+        bucket_name (str) name of the bucket. 
+        file_key (str) name of the file in the bucket.
+            
+        Returns:
+            int: The number of versions for the specified file.
+        """
+
+        # Get the list of object versions for the specified file
+        versions = self.s3_client.list_object_versions(Bucket=bucket_name, Prefix=file_prefix)
+        # Count the number of versions
+        version_count = 0
+        if 'Versions' in versions:
+            version_count = len(versions['Versions'])
+
+        return version_count

.nightswatch/functional/helpers/website_model/edit_page.py

@@ -14,9 +14,12 @@
 
 import time
 import logging
+import random
+import string
 
 from helpers.website_model.dom_operator import DomOperator
 from helpers.utils.textbox import Textbox
+from selenium.webdriver.remote.webelement import WebElement
 
 MODAL_XPATH = '//div[@id="add-question-form"]'
 EDIT_MODAL_XPATH = '//div[@class="dialog dialog--active"]'
@@ -812,10 +815,16 @@ def execute_test_query(self, query: str) -> None:
         query_textbox.set_value(query)
         self.operator.select_id(TEST_TAB_QUERY_BUTTON_ID, click=True)
 
-    def generate_test_report(self) -> str:
+    def generate_test_report(self) -> WebElement:
         """
         Generates a test report and returns the text content of the job
         """
+        filename_textbox = Textbox(self.operator.select_id("filename"))
+        random_file_name = 'TestAll_' + ''.join(random.choices(string.ascii_letters + string.digits, k=4))
+        filename_textbox.set_value(random_file_name)
         self.operator.select_id(TEST_ALL_BUTTON_ID, click=True)
-        self.operator.wait_for_element_by_id_text(TEST_ALL_JOBS_ID, 'Completed', delay=300)
-        return self.operator.select_id(TEST_ALL_JOBS_ID).text
+        self.operator.wait_for_element_by_xpath(f"//div[starts-with(@id, 'test-job-{random_file_name}')]")
+
+        last_test_execution_element = self.operator.select_xpath(f"//div[starts-with(@id, 'test-job-{random_file_name}')]")
+        self.operator.wait_for_element_by_id_text(last_test_execution_element.get_property("id"), 'Completed', delay=300)
+        return self.operator.select_id(last_test_execution_element.get_property("id"))

.nightswatch/functional/helpers/website_model/menu_nav.py

@@ -33,6 +33,7 @@
 KENDRA_ID = 'page-link-kendraIndexing'
 CUSTOM_TERM_ID = 'page-link-customTranslate'
 CHAT_ID = 'page-link-client'
+TEST_ALL_ID = 'testAll-tab'
 
 class MenuNav:
     """Class representing a Menu Navigation Bar.
@@ -124,4 +125,10 @@ def open_chat_page(self) -> ChatPage:
         self.operator.click_element_by_id(CHAT_ID, wait=10)
         time.sleep(5)
         self.operator.switch_windows()
-        return ChatPage(self.operator)
+        return ChatPage(self.operator)
+
+    def open_testall_page(self) -> None:
+        """Opens the TestAllPage through navigation bar."""
+
+        self.operator.click_element_by_id(TEST_ALL_ID, wait=10)
+        time.sleep(5)

.nightswatch/functional/helpers/website_model/settings_page.py

@@ -12,7 +12,7 @@
 ######################################################################################################################
 
 import time
-
+import os
 import selenium
 
 from helpers.utils.textbox import Textbox
@@ -64,6 +64,15 @@
 TEXT_GENERATION_GENERAL_SUBGROUP_ID = 'text_generation_general_subgroup'
 AMAZON_BEDROCK_KNOWLEDGE_BASES_SUBGROUP_ID = 'amazon_bedrock_knowledge_bases_subgroup'
 
+KNOWLEDGE_BASE_SEARCH_TYPE_ID = 'KNOWLEDGE_BASE_SEARCH_TYPE'
+KNOWLEDGE_BASE_MAX_NUMBER_OF_RETRIEVED_RESULTS_ID = 'KNOWLEDGE_BASE_MAX_NUMBER_OF_RETRIEVED_RESULTS'
+KNOWLEDGE_BASE_MODEL_PARAMS_ID = 'KNOWLEDGE_BASE_MODEL_PARAMS'
+KNOWLEDGE_BASE_PROMPT_TEMPLATE_ID = 'KNOWLEDGE_BASE_PROMPT_TEMPLATE'
+
+BEDROCK_GUARDRAIL_IDENTIFIER_ID = 'BEDROCK_GUARDRAIL_IDENTIFIER'
+BEDROCK_GUARDRAIL_VERSION_ID = 'BEDROCK_GUARDRAIL_VERSION'
+BEDROCK_GUARDRAIL_SUBGROUP_ID = 'text_generation_guardrail_subgroup'
+
 class SettingsPage:
     """
     Class representing a Settings Page.
@@ -89,7 +98,7 @@ def save_settings(self) -> str:
 
         self.operator.select_xpath(SAVE_XPATH, click=True)
         self.operator.wait_for_element_by_xpath(SAVE_MODAL_CLOSE_XPATH)
-        time.sleep(1)
+        time.sleep(2)
 
         status = self.operator.select_css(SAVE_STATUS_CSS).text
         self.operator.select_xpath(SAVE_MODAL_CLOSE_XPATH, click=True)
@@ -154,6 +163,19 @@ def customize_empty_message(self, message) -> str:
         customize_empty_message = self.operator.select_id(EMPTY_MESSAGE_ID)
         self.__set_element_value(customize_empty_message, message)
         return self.save_settings()
+
+    def enable_debug_response(self) -> str:
+        """
+        Enables debug responses during the chat conversation and saves the changes.
+
+        Returns:
+            The status of the save operation.
+        """
+
+        enable_debug = self.operator.select_id(ENABLE_DEBUG_RESPONSES_ID)
+        self.__set_element_value(enable_debug, 'true')
+        return self.save_settings()
+
 
     def enable_multi_language_support(self) -> str:
         """
@@ -317,6 +339,39 @@ def disable_llm_disambiguation(self):
         self.__set_element_value(enable_generative_query, 'false')
 
         return self.save_settings()
+
+    def enable_bedrock_guardrail(self, region, guardrail_identifier, guardrail_version):
+        """
+        Enables the Bedrock guardrail for functional tests based on the nightswatch or local environment. 
+
+        Args:
+            region (str): The region for the guardrail.
+
+        Returns:
+            The status of the save operation.
+        """
+
+        mappings = {
+            'us-east-1': ('6wptcgn6mi7x', 2),
+            'us-west-2': ('nnbn5202wy5g', 2),
+            'eu-west-2': ('jsj81qgv3ky5', 2),
+            'ap-northeast-1': ('672yn8u1u3v5', 1)
+        }
+
+        if os.getenv('NIGHTSWATCH_TEST_DIR'):
+            guardrail_identifier = mappings[region][0]
+            guardrail_version = mappings[region][1]
+
+        if not guardrail_identifier or not guardrail_version:
+            return self.save_settings()
+
+        get_guardrail_identifier = self.operator.select_id(BEDROCK_GUARDRAIL_IDENTIFIER_ID)
+        self.__set_element_value(get_guardrail_identifier, guardrail_identifier)
+
+        get_guardrail_version = self.operator.select_id(BEDROCK_GUARDRAIL_VERSION_ID)
+        self.__set_element_value(get_guardrail_version, guardrail_version)
+
+        return self.save_settings()
 
     def enable_custom_terminology(self) -> str:
         """
@@ -410,6 +465,40 @@ def set_post_processing_lambda(self, l: str) -> str:
         post_processing_lambda = self.operator.select_id(POST_PROCESSING_LAMBDA_ID)
         self.__set_element_value(post_processing_lambda, l)
         return self.save_settings()
+
+    def disable_kb_prompt(self) -> str:
+        """
+        Disables prompt for knowledge base which is enabled by default
+
+        Returns:
+            The status of the save operation.
+        """
+        kb_prompt = self.operator.select_id(KNOWLEDGE_BASE_PROMPT_TEMPLATE_ID)
+        self.__set_element_value(kb_prompt, '')
+
+        return self.save_settings()
+    def enable_kb_advanced(self, knowledge_base_model) -> str:
+        """
+        Enables advanced settings for the knowledge base
+
+        Returns:
+            The status of the save operation.
+        """
+        kb_search_type = self.operator.select_id(KNOWLEDGE_BASE_SEARCH_TYPE_ID)
+        kb_max_results = self.operator.select_id(KNOWLEDGE_BASE_MAX_NUMBER_OF_RETRIEVED_RESULTS_ID)
+        kb_model_params = self.operator.select_id(KNOWLEDGE_BASE_MODEL_PARAMS_ID)
+
+        if knowledge_base_model.startswith('anthropic'):
+            self.__set_element_value(kb_search_type, 'HYBRID')
+            self.__set_element_value(kb_max_results, 3)
+            self.__set_element_value(kb_model_params, '{"temperature": 0.3, "maxTokens": 245, "topP": 0.9, "top_k": 240 }')
+        else:
+            self.__set_element_value(kb_search_type, 'HYBRID')
+            self.__set_element_value(kb_max_results, 5)
+            self.__set_element_value(kb_model_params, '{"temperature": 0.1, "maxTokens": 264, "topP": 0.9 }')
+
+
+        return self.save_settings()
 
     def expand_all_subgroups(self) -> None:
         """
@@ -477,6 +566,11 @@ def expand_all_subgroups(self) -> None:
                 amazon_bedrock_knowledge_bases_subgroup.click()
                 self.operator.wait_for_element_attribute(AMAZON_BEDROCK_KNOWLEDGE_BASES_SUBGROUP_ID, 'aria-expanded', 'true')
 
+            bedrock_guardrail_general_subgroup = self.operator.select_id(BEDROCK_GUARDRAIL_SUBGROUP_ID)
+            if bedrock_guardrail_general_subgroup.get_attribute('aria-expanded') == 'false':
+                bedrock_guardrail_general_subgroup.click()
+                self.operator.wait_for_element_attribute(BEDROCK_GUARDRAIL_SUBGROUP_ID, 'aria-expanded', 'true')
+
         except selenium.common.exceptions.ElementClickInterceptedException:
             # The exception above happens when a window obscures the settings page,
             # In this case it is safe to ignore that error and continue on with the test.

.nightswatch/functional/test_1_login.py

@@ -82,3 +82,13 @@ def test_invalid_client_login(self, invalid_client_login):
         assert title[0] == 'Signin'
         assert title[1] == 'Incorrect username or password.'
 
+    def test_test_all_before_import(self, designer_login, dom_operator: DomOperator):
+        """
+        Tests the test all functionality before importing questions.
+        """
+        menu = MenuNav(dom_operator)
+        edit_page = menu.open_edit_page()
+        edit_page.select_test_all_tab()
+        report_status = edit_page.generate_test_report()
+        assert 'Completed' in report_status.text
+