Daksha Download & Read functionality.

CHANGELOG.md

@@ -1,6 +1,10 @@
 # `DAKSHA` Change Log
 
-*version*: `2.2.2`
+*version*: `2.3.0`
+
+## v2.3.0
+1. Added files download functionality to Daksha
+2. Text & Csv file read functionality on the downloaded file
 
 ## v2.2.2
 1. Fixed server initialisation issue in case of running daksha server in local with no database

README.md

@@ -69,6 +69,10 @@ You can configure the application in a lot of ways by setting the following envi
 * **POSTMARK_TOKEN**
   * This is the postmark token which will be used to send reports by email.
   * If not set, we'll skip sending the email.
+
+* **DOWNLOAD_PATH**
+  * This is the location of directory where files downloaded by the application are stored.
+  * Defaults to 'downloads' in current directory.
 
 * **ALERT_URL**
   * This is the alert webhook url which will be used to send alerts in gchat/slack.

daksha/settings.py

@@ -158,6 +158,10 @@
 ALERT_URL = os.environ.get('ALERT_URL', '')
 # Github Configurations (if you choose to store test ymls in github)
 
+#Downloads configurations
+downloads = os.path.join(BASE_DIR, "downloads")
+DOWNLOAD_PATH = downloads
+
 """
 If either GIT_USER OR GIT_PASSWORD OR both are empty then only Public Repository or Organisation can be accessed
 If both GIT_USER and GIT_PASSWORD are given then Public/Private Repository or Organisation can be accessed

daksha_know-how/CreateTest.md

@@ -71,6 +71,8 @@ task:
    * **scroll_to**:  You need to provide the locator of the webelement that you want to scroll down/up to.**Locator** can be *xpath, id, css, name, tagname, classname, linktext and partiallinktext*
    * **wait-for**: This has 2 fields- mode, value/locator.This has 3 mode : *visibility,invisibility,hardwait*.For *visibility/invisibility* please provide the locator of the webelement that you want to wait for. In case of mode *hardwait*, please provide the value as number of seconds you want to wait for.
    * **quit_browser**: You are recommended to add this step at the end of your test case steps to quit the remote or local browser.
+   * **download_file**: This has two subfield; *locator* and *save_in*. **Locator** is the file path that needs to be downloaded. It can be *xpath, id, css, name, tagname, classname, linktext and partiallinktext*. **Save_in** is optional in case user wants to read the downloaded file. <br/> **Note**: download_file only works if both the daksha directory and the test chrome browser are on the same system. i.e. both on the server or the local system.
+   * **read_file**: This is to read a downloaded file in the same session. Pass the *save_in* value used to save the downloaded file with *read_from* key.
 
 
 

engine/executor.py

@@ -50,6 +50,7 @@ def execute_test(test_executor: TestExecutor, email):
         execution_result, error_stack = True, None
         step = {}
         test_yml = test_executor.test_yml
+        test_uuid = test_executor.test_uuid
         config = test_yml["config"]
         task = test_yml["task"]
         name = test_yml["name"]  # TODO: Alert user if no/null config/task/name is provided
@@ -59,7 +60,7 @@ def execute_test(test_executor: TestExecutor, email):
             logger.info("Users has opted for alerts via " + alert_type)
         else:
             logger.info("User has not opted for alerts")
-        web_driver = browser_config(config)
+        web_driver = browser_config(config,test_uuid)
         test_executor.web_driver = web_driver
         for step in task:
             execution_result, error_stack = execute_step(test_executor, step)

engine/method_mapper.py

@@ -43,4 +43,6 @@
     "wait_for": wait_for,
     "capture_ui_element": capture_ui_element,
     "scroll_to": scroll_to,
+    "download_file": download_file,
+    "read_file":read_file,
 }

engine/selenium_helper.py

@@ -14,7 +14,10 @@
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 """
-
+import csv
+import glob
+import os
+import pathlib
 import traceback
 import time
 from selenium import webdriver
@@ -24,13 +27,14 @@
 from selenium.webdriver.common.by import By
 from selenium.webdriver.support.ui import WebDriverWait, Select
 from selenium.webdriver.support import expected_conditions as EC
+from daksha.settings import DOWNLOAD_PATH
 
 from .logs import logger
 from .models import TestExecutor
 from .utils.screenshot_utils import take_screenshot
 
 
-def browser_config(config) -> WebDriver:
+def browser_config(config,test_uuid) -> WebDriver:
     """
     Configures the browser in accordance with mentioned specifications(env,browser,driverAddress) in YAML
      :param config: Browser configuration fetched from YAML
@@ -44,18 +48,23 @@ def browser_config(config) -> WebDriver:
         env = config['env']
         brow = config['browser']
         path = config['driverAddress']
+        download_path = os.path.join(DOWNLOAD_PATH,  test_uuid)
+        prefs = {'download.default_directory': download_path}
     except KeyError:
         raise Exception(
             "Ill formatted arguments, 'env', 'browser' and 'driverAddress' must be present in the list of args")
     if brow.lower() == 'chrome' and env.lower() == 'local':
-        web_driver = webdriver.Chrome(executable_path=path)
+        options = webdriver.ChromeOptions()
+        options.add_experimental_option('prefs', prefs)
+        web_driver = webdriver.Chrome(executable_path=path, options=options)
     elif brow.lower() == 'chrome' and env.lower() == 'remote':
         options = webdriver.ChromeOptions()
         options.add_argument("no-sandbox")
         options.add_argument("--disable-gpu")
         options.add_argument("--window-size=800,600")
         options.add_argument("--disable-dev-shm-usage")
         options.add_argument("--start-maximized")
+        options.add_experimental_option('prefs', prefs)
         web_driver = webdriver.Remote(
             command_executor=path,
             desired_capabilities=DesiredCapabilities.CHROME,
@@ -520,3 +529,68 @@ def scroll_to(test_executor: TestExecutor, **kwargs):
             logger.error("Attempt " + str(i) + " to scroll to element failed")
     return False, error_stack
 
+def download_file(test_executor: TestExecutor, **kwargs):
+    locator, locator_value = get_locator_info(**kwargs)
+    test_uuid = test_executor.test_uuid
+    wait = True
+    download_path = os.path.join(DOWNLOAD_PATH,  test_uuid)
+    try:
+        logger.info("Going to download file")
+        element = WebDriverWait(test_executor.web_driver, 10).until(
+            EC.visibility_of_element_located((locator, locator_value))
+        )
+        element.click()
+        logger.info("Downloading File....")
+        time.sleep(2)
+        # waiting for download to complete
+        while wait:
+            files = glob.glob(download_path + "/*")
+            downloaded_file = max(files, key=os.path.getctime)
+            file_extension = pathlib.Path(downloaded_file).suffix
+            if file_extension.lower() == ".crdownload" or file_extension.lower() == ".part":
+                wait = True
+                time.sleep(1)
+            else:
+                wait = False
+        # Printing file name & download path
+        file_name = pathlib.Path(downloaded_file).stem
+        file_downloaded = file_name+file_extension
+        logger.info("File "+file_downloaded+" downloaded in Daksha/"+test_uuid+" downloads folder")
+        # saving downloaded file in save_in variable
+        if 'save_in' in kwargs.keys():
+            save_in = kwargs['save_in']
+            test_executor.variable_dictionary[save_in] = downloaded_file
+            logger.info("File saved in: "+save_in+". Can be used for reading file ")
+        else :
+            logger.info("No SAVE_IN key passed. Read file won't be supported for this download")
+    except Exception as e:
+        logger.error(traceback.format_exc())
+        return False, None
+    return True, None
+
+def read_file(test_executor: TestExecutor, **kwargs):
+    # Reading text & CSV files
+    try:
+        save_in = kwargs['read_from']
+    except KeyError:
+        return False, "Ill formatted arguments, 'save_in' must be present in the list of args"
+    file = test_executor.variable_dictionary[save_in]
+    logger.info("Read file: "+file)
+    file_extension = pathlib.Path(file).suffix
+    if file_extension == '.txt':
+        logger.info("Reading Text File: \n")
+        with open(file) as f:
+            contents = f.read()
+            logger.info(contents)
+        f.close()
+    elif file_extension == '.csv':
+        logger.info("Reading CSV File: \n")
+        with open(file) as f:
+            csvFile = csv.reader(f)
+            for lines in csvFile:
+                logger.info(lines)
+        f.close()
+    else:
+        logger.info("Daksha currently only support txt & csv files readability !")
+    return True, None
+

examples/download_read_example.yml

@@ -0,0 +1,18 @@
+config:
+  env: local
+  browser: chrome
+  driverAddress: /Users/wayne/Desktop/chromedriver
+name: downloadTest
+task:
+  - open_url:
+      url: https://wsform.com/knowledgebase/sample-csv-files/
+  - wait_for:
+      mode: hardwait
+      value: 1
+  - download_file:
+      xpath: //a[contains(text(),'month.csv')]
+      save_in: downloadF
+  - download_file:
+      xpath: //a[contains(text(),'time.csv')]
+  - read_file:
+      read_from: downloadF