Merge branch 'dev' of https://github.com/microsoft/document-generation-solution-accelerator into psl-aisearchtool-dev

Author: Ragini-Microsoft

.github/workflows/pylint.yml

@@ -29,5 +29,5 @@ jobs:
       - name: Pylint
         run: |
           echo "Running Pylint..."
-          python -m flake8 --config=.flake8 --verbose .
+          python -m flake8 --config=.flake8 --exclude=tests --verbose .
           

tests/e2e-test/base/base.py

@@ -45,4 +45,35 @@ def validate_response_status(self, question_api=""):
             "response code is " + str(response.status) + " " + str(response.json())
         )
 
-  
+
+    def validate_generate_response_status(self, question_api=""):
+        load_dotenv()  # Ensure environment variables are loaded
+        # URL of the API endpoint
+        url = f"{URL}/history/generate"
+
+        # Prepare headers
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "*/*",
+        }
+
+        # Payload (data) to be sent in the POST request
+        payload = {
+            "chat_type": "template",
+            "messages": [
+                {
+                    "role": "user",
+                    "content": question_api,  # Use the passed question
+                }
+            ],
+        }
+
+        # Make the POST request
+        response = self.page.request.post(
+            url, headers=headers, data=json.dumps(payload), timeout=120000
+        )
+        assert response.status == 200, (
+            "response code is " + str(response.status) + " " + str(response.json())
+        )
+
+

tests/e2e-test/config/constants.py

@@ -1,5 +1,5 @@
 import os
-
+import json
 from dotenv import load_dotenv
 
 load_dotenv()
@@ -11,14 +11,6 @@
 # Get the absolute path to the repository root
 repo_root = os.getenv("GITHUB_WORKSPACE", os.getcwd())
 
-# # Construct the absolute path to the JSON file
-# #note: may have to remove 'Doc_gen' from below when running locally
-# json_file_path = os.path.join(repo_root,'testData', 'section_title.json')
-
-# with open(json_file_path, 'r') as file:
-#     data = json.load(file)
-#     sectionTitle = data['sectionTitle']
-
 # browse input data
 browse_question1 = "What are typical sections in a promissory note?"
 browse_question2 = "List the details of two promissory notes governed by the laws of the state of California"
@@ -29,3 +21,4 @@
 
 # Response Text Data
 invalid_response = "I was unable to find content related to your query and could not generate a template. Please try again."
+invalid_response1 = "An error occurred. Answers can't be saved at this time. If the problem persists, please contact the site administrator."

tests/e2e-test/pages/draftPage.py

@@ -1,90 +1,174 @@
 from base.base import BasePage
 from pytest_check import check
+from config.constants import *
+import logging
+logger = logging.getLogger(__name__)
 import time
-from collections import defaultdict
 
 
 class DraftPage(BasePage):
-    # Principal_Amount_and_Date = "div:nth-child(3) div:nth-child(2) span:nth-child(1) textarea:nth-child(1)"
-    # Borrower_Information = "div:nth-child(3) div:nth-child(2) span:nth-child(1) textarea:nth-child(1)"
-    # Payee_Information = "//div[3]//div[2]//span[1]//textarea[1]"
     Draft_Sections = "//textarea"
     Draft_headings = "//span[@class='fui-Text ___nl2uoq0 fk6fouc f4ybsrx f1i3iumi f16wzh4i fpgzoln f1w7gpdv f6juhto f1gl81tg f2jf649 fepr9ql febqm8h']"
     invalid_response = "The requested information is not available in the retrieved data. Please try another query or topic."
     invalid_response1 = "There was an issue fetching your data. Please try again."
-    invalid_response2 = " "
+    
 
     def __init__(self, page):
         self.page = page
 
-    def check_draft_sections(self, timeout: float = 180.0, poll_interval: float = 0.5):
-        """
-        Waits for all <textarea> draft sections to load valid content using .input_value().
-        Scrolls into view if needed, retries until timeout.
-        Raises clear errors if validation fails.
-        """
-        from collections import defaultdict
-        import time
-        start_time = time.time()
+    def validate_draft_sections_loaded(self):
+        max_wait_time = 180  # seconds
+        poll_interval = 2
 
+        self.page.wait_for_timeout(25000)
 
-        while time.time() - start_time < timeout:
-            section_elements = self.page.locator(self.Draft_Sections)
-            heading_elements = self.page.locator(self.Draft_headings)
+        # All draft section containers
+        section_blocks = self.page.locator("//div[@class='ms-Stack ___mit7380 f4zyqsv f6m9rw3 fwbpcpn folxr9a f1s274it css-103']")
+        total_sections = section_blocks.count()
 
-            section_count = section_elements.count()
-            heading_count = heading_elements.count()
+        logger.info(f"🔍 Total sections found: {total_sections}")
 
-            if section_count < 13 or heading_count < 13:
-                print("[WAIT] Waiting for all sections to appear...")
-                time.sleep(poll_interval)
-                continue
+        for index in range(total_sections):
+            section = section_blocks.nth(index)
 
-            failed_sections = defaultdict(str)
+            try:
+                section.scroll_into_view_if_needed()
+                self.page.wait_for_timeout(500)
 
-            for i in range(section_count):
-                section = section_elements.nth(i)
+                title_element = section.locator("//span[@class='fui-Text ___nl2uoq0 fk6fouc f4ybsrx f1i3iumi f16wzh4i fpgzoln f1w7gpdv f6juhto f1gl81tg f2jf649 fepr9ql febqm8h']")
+                title_text = title_element.inner_text(timeout=5000).strip()
+            except Exception as e:
+                logger.error(f"❌ Could not read title for section #{index + 1}: {e}")
+                continue
 
+            logger.info(f"➡️ Validating section [{index + 1}/{total_sections}]: '{title_text}'")
+
+            content_locator = section.locator("//textarea")
+            generate_btn = section.locator("//span[@class='fui-Button__icon rywnvv2 ___963sj20 f1nizpg2']")
+            spinner_locator = section.locator("//div[@id='section-card-spinner']")
+
+            content_loaded = False
+
+            # 🚨 If spinner is visible inside this section, click generate immediately
+            try:
+                if spinner_locator.is_visible(timeout=1000):
+                    logger.warning(f"⏳ Spinner found in section '{title_text}'. Clicking Generate immediately.")
+                    generate_btn.click()
+                    self.page.wait_for_timeout(3000)
+                    confirm_btn = self.page.locator("//button[@class='fui-Button r1alrhcs ___zqkcn80 fd1o0ie fjxutwb fwiml72 fj8njcf fzcpov4 f1d2rq10 f1mk8lai ff3glw6']")
+                    if confirm_btn.is_visible(timeout=3000):
+                        confirm_btn.click()
+                        logger.info(f"🟢 Clicked Confirm button for section '{title_text}'")
+                    else:
+                        logger.warning(f"⚠️ Confirm button not visible for section '{title_text}'")
+            except Exception as e:
+                logger.error(f"❌ Error while clicking Confirm button for section '{title_text}': {e}")
+
+            # ⏳ Retry short wait (15s) for content to load
+            short_wait = 15
+            short_start = time.time()
+            while time.time() - short_start < short_wait:
                 try:
-                    # Scroll into view and wait a bit for rendering
-                    section.scroll_into_view_if_needed(timeout=2000)
-                    self.page.wait_for_timeout(200)
-
-                    # Extract content from <textarea>
-                    section_text = section.input_value().strip()
-
-                    if not section_text:
-                        failed_sections[i] = "Empty"
-                    elif section_text in (self.invalid_response, self.invalid_response1):
-                        failed_sections[i] = f"Invalid: {repr(section_text[:30])}"
-
-                except Exception as e:
-                    failed_sections[i] = f"Exception: {str(e)}"
-
-            if not failed_sections:
-                break  # ✅ All good
-            else:
-                print(f"[WAITING] Sections not ready yet: {failed_sections}")
+                    content = content_locator.text_content(timeout=2000).strip()
+                    if content:
+                        logger.info(f"✅ Section '{title_text}' loaded after Generate + Confirm.")
+                        content_loaded = True
+                        break
+                except:
+                    pass
+                time.sleep(1)
+
+            if not content_loaded:
+                logger.error(f"❌ Section '{title_text}' still empty after Generate + Confirm wait ({short_wait}s). Skipping.")
+
+            # Step 1: Wait for content to load normally
+            start = time.time()
+            while time.time() - start < max_wait_time:
+                try:
+                    content = content_locator.text_content(timeout=2000).strip()
+                    if content:
+                        logger.info(f"✅ Section '{title_text}' loaded successfully.")
+                        content_loaded = True
+                        break
+                except:
+                    pass
                 time.sleep(poll_interval)
 
-        else:
-            raise TimeoutError(f"❌ Timeout: These sections did not load valid content: {failed_sections}")
-
-        # ✅ Final validations after loading
-        for i in range(section_count):
-            section = section_elements.nth(i)
-            heading = heading_elements.nth(i)
-
-            section.scroll_into_view_if_needed(timeout=2000)
-            self.page.wait_for_timeout(200)
-
-            heading_text = heading.inner_text(timeout=3000).strip()
-            content = section.input_value().strip()
+            # Step 2: If still not loaded, click Generate and retry
+            if not content_loaded:
+                logger.warning(f"⚠️ Section '{title_text}' is empty. Attempting 'Generate'...")
 
-            print(f"[VALIDATING] Section {i}: '{heading_text}' → {repr(content[:60])}...")
-
-            with check:
-                check.is_not_none(content, f"❌ Section '{heading_text}' is None")
-                check.not_equal(content, "", f"❌ Section '{heading_text}' is empty")
-                check.not_equal(content, self.invalid_response, f"❌ '{heading_text}' has invalid response")
-                check.not_equal(content, self.invalid_response1, f"❌ '{heading_text}' has invalid response")
+                try:
+                    generate_btn.click()
+                    logger.info(f"🔄 Clicked 'Generate' for section '{title_text}'")
+                except Exception as e:
+                    logger.error(f"❌ Failed to click 'Generate' for section '{title_text}': {e}")
+                    continue
+
+                # Retry wait
+                start = time.time()
+                while time.time() - start < max_wait_time:
+                    try:
+                        content = content_locator.text_content(timeout=2000).strip()
+                        if content:
+                            logger.info(f"✅ Section '{title_text}' loaded after clicking Generate.")
+                            content_loaded = True
+                            break
+                    except:
+                        pass
+                    time.sleep(poll_interval)
+
+                if not content_loaded:
+                    logger.error(f"❌ Section '{title_text}' still empty after retrying.")
+
+                    # Optional: take screenshot
+                    screenshot_dir = "screenshots"
+                    os.makedirs(screenshot_dir, exist_ok=True)
+                    screenshot_path = os.path.join(screenshot_dir, f"section_{index + 1}_{title_text.replace(' ', '_')}.png")
+                    try:
+                        section.screenshot(path=screenshot_path)
+                        logger.error(f"📸 Screenshot saved: {screenshot_path}")
+                    except Exception as e:
+                        logger.error(f"❌ Generate click failed in section '{title_text}': {e}")
+                        continue
+
+            try:
+                    content = content_locator.text_content(timeout=2000).strip()
+                    with check:
+                        if content == self.invalid_response or content == self.invalid_response1:
+                            logger.warning(f"❌ Invalid response found in '{title_text}'. Retrying Generate + Confirm...")
+
+                            try:
+                                generate_btn.click()
+                                self.page.wait_for_timeout(3000)
+
+                                confirm_btn = self.page.locator("//button[@class='fui-Button r1alrhcs ___zqkcn80 fd1o0ie fjxutwb fwiml72 fj8njcf fzcpov4 f1d2rq10 f1mk8lai ff3glw6']")
+                                if confirm_btn.is_visible(timeout=3000):
+                                    confirm_btn.click()
+                                    logger.info(f"🟢 Retried Confirm for section '{title_text}'")
+                                else:
+                                    logger.warning(f"⚠️ Confirm button not visible during retry for '{title_text}'")
+                            except Exception as e:
+                                logger.error(f"❌ Retry Generate/Confirm failed: {e}")
+
+                            retry_start = time.time()
+                            while time.time() - retry_start < short_wait:
+                                try:
+                                    content = content_locator.text_content(timeout=2000).strip()
+                                    if content and content not in [self.invalid_response, self.invalid_response1]:
+                                        logger.info(f"✅ Section '{title_text}' fixed after retry.")
+                                        break
+                                except:
+                                    pass
+                                time.sleep(1)
+
+                            with check:
+                                check.not_equal(content, self.invalid_response, f"❌ '{title_text}' still has invalid response after retry")
+                                check.not_equal(content, self.invalid_response1, f"❌ '{title_text}' still has invalid response after retry")
+
+                        else:
+                            logger.info(f"🎯 Section '{title_text}' has valid content.")
+            except Exception as e:
+                    logger.error(f"❌ Could not validate content for '{title_text}': {e}")
+
+                    logger.info(f"✔️ Completed section: '{title_text}'\n")

tests/e2e-test/pages/generatePage.py

@@ -1,12 +1,15 @@
+from asyncio.log import logger
 from base.base import BasePage
 from playwright.sync_api import expect
-from asyncio.log import logger
+import logging
+logger = logging.getLogger(__name__)
+
 
 class GeneratePage(BasePage):
     GENERATE_DRAFT = "//button[@title='Generate Draft']"
     TYPE_QUESTION = "//textarea[@placeholder='Type a new question...']"
     SEND_BUTTON = "//div[@aria-label='Ask question button']"
-    SHOW_CHAT_HISTORY_BUTTON="//span[text()='Show template history']"
+    SHOW_CHAT_HISTORY_BUTTON = "//span[text()='Show template history']"
     HIDE_CHAT_HISTORY_BUTTON = "//span[text()='Hide Chat History']"
     CHAT_HISTORY_ITEM = "//body//div[@id='root']//div[@role='presentation']//div[@role='presentation']//div[1]//div[1]//div[1]//div[1]//div[1]//div[1]"
     SHOW_CHAT_HISTORY = "//span//i"
@@ -28,7 +31,25 @@ def enter_a_question(self, text):
     def click_send_button(self):
         # Type a question in the text area
         self.page.locator(self.SEND_BUTTON).click()
-        self.page.wait_for_timeout(10000)
+        locator = self.page.locator("//p[contains(text(),'Generating template...this may take up to 30 secon')]")
+        stop_button = self.page.locator("//div[@aria-label='Stop generating']")
+
+        try:
+            # Wait up to 60s for the element to become **hidden**
+            locator.wait_for(state="hidden", timeout=60000)
+        except TimeoutError:
+            msg = "❌ TIMED-OUT: Not recieved response within 60 sec."
+            logger.info(msg)  # ✅ log to console/log file
+            raise AssertionError(msg)
+        
+        finally:
+            if stop_button.is_visible():
+                stop_button.click()
+                logger.info("Clicked on 'Stop generating' button after timeout.")
+            else:
+                logger.info("'Stop generating' button not visible.")
+
+        self.page.wait_for_timeout(5000)
 
     def click_generate_draft_button(self):
         # Type a question in the text area
@@ -52,23 +73,29 @@ def close_chat_history(self):
             hide_button.click()
             self.page.wait_for_timeout(2000)
         else:
-            logger.info("Hide button not visible. Chat history might already be closed.")
+            logger.info(
+                "Hide button not visible. Chat history might already be closed."
+            )
 
     def delete_chat_history(self):
 
         self.page.locator(self.SHOW_CHAT_HISTORY_BUTTON).click()
+        self.page.wait_for_timeout(4000)
         chat_history = self.page.locator("//span[contains(text(),'No chat history.')]")
         if chat_history.is_visible():
             self.page.wait_for_load_state("networkidle")
-            self.page.locator("button[title='Hide']").wait_for(state="visible", timeout=5000)
+            self.page.locator("button[title='Hide']").wait_for(
+                state="visible", timeout=5000
+            )
             self.page.locator("button[title='Hide']").click()
 
         else:
             self.page.locator(self.CHAT_HISTORY_OPTIONS).click()
             self.page.locator(self.CHAT_HISTORY_DELETE).click()
+            self.page.wait_for_timeout(5000)
             self.page.get_by_role("button", name="Clear All").click()
             self.page.wait_for_timeout(5000)
+            expect(self.page.locator("//span[contains(text(),'No chat history.')]")).to_be_visible()
             self.page.locator(self.CHAT_HISTORY_CLOSE).click()
             self.page.wait_for_load_state("networkidle")
             self.page.wait_for_timeout(2000)
-