add more methods

Author: Álvaro Santos Laserna López

testui/elements/testui_element.py

@@ -42,9 +42,12 @@ def testui_error(driver, exception):
     root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
     Path(root_dir + "/report_screenshots").mkdir(parents=True, exist_ok=True)
     image_name = f'ERROR-{driver.device_name}-{current_time}.png'
-    driver.save_screenshot(image_name, directory='report_screenshots/')
-    exception = exception + f'{logger.bcolors.FAIL} \n Screenshot taken and saved as: ' \
-                            f'report_screenshots/testui-{image_name}{logger.bcolors.ENDC}\n'
+    try:
+        driver.save_screenshot(image_name, directory='report_screenshots/')
+        exception += f'{logger.bcolors.FAIL} \n Screenshot taken and saved as: ' \
+                     f'report_screenshots/testui-{image_name}{logger.bcolors.ENDC}\n'
+    except Exception as error:
+        exception += f'{logger.bcolors.FAIL} \nCould not take screenshot:{logger.bcolors.ENDC}\n {error}'
     full_exception = error_with_traceback(exception)
     if driver.soft_assert:
         logger.log_error(full_exception)
@@ -440,9 +443,10 @@ def screenshot(self, image_name='cropped_image.png'):
 
         return self
 
-    def find_image_match(self, image_name, threshold=0.9, image_match='', max_scale=2.0):
+    def find_image_match(self, image_name, threshold=0.9, image_match='', max_scale=2.0, min_scale=0.3):
         """
         Takes screenshot of the element and compares it with the one you provide as 'image_name'
+        :param min_scale:
         :param max_scale:
         :param image_match: returns the image with a rectangle showing the match
         :param image_name: relative path to image
@@ -453,7 +457,7 @@ def find_image_match(self, image_name, threshold=0.9, image_match='', max_scale=
         self.__is_not = False
         self.screenshot(self.device_name + '.png')
         found, p = ImageRecognition(self.device_name + '.png', image_name, threshold, self.device_name) \
-            .compare(image_match, max_scale)
+            .compare(image_match=image_match, max_scale=max_scale, min_scale=min_scale)
         if not found and not is_not:
             self.__show_error(f'{self.device_name}: The images compared did not match. threshold={threshold}. '
                               f'Matched = {p}')
@@ -466,6 +470,30 @@ def find_image_match(self, image_name, threshold=0.9, image_match='', max_scale=
 
         return self
 
+    def is_image_match(self, image_name, threshold=0.9, image_match='', max_scale=2.0, min_scale=0.3):
+        """
+        Takes screenshot of the element and compares it with the one you provide as 'image_name'
+        :param min_scale:
+        :param max_scale:
+        :param image_match: returns the image with a rectangle showing the match
+        :param image_name: relative path to image
+        :param threshold: limit to consider image as a match (0 to 1)
+        :return: Elements
+        """
+        is_not = self.__is_not
+        self.__is_not = False
+        self.screenshot(self.device_name + '.png')
+        found, p = ImageRecognition(self.device_name + '.png', image_name, threshold, self.device_name) \
+            .compare(image_match=image_match, max_scale=max_scale, min_scale=min_scale)
+        if not found and not is_not:
+            return False
+        if found and is_not:
+            return False
+        root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + '/'
+        os.remove(root_dir + self.device_name + '.png')
+
+        return True
+
     def swipe(self, start_x=None, start_y=None, end_x=None, end_y=None, el=None, duration=None):
         """
         It swipes from element to the el(second element) or to the coordinates you define with start_x, start_y,
@@ -686,7 +714,11 @@ def press_and_compare(self, image, milliseconds=1000, threshold=0.9, fps_reducti
         return self
 
     def collection_size(self):
+        self.__is_collection = False
+        index = self.index
+        self.index = 0
         self.wait_until_visible()
+        self.index = index
         self.__is_collection = True
         return len(self.__find_by_collection())
 
@@ -743,4 +775,8 @@ def class_name(cls, text: str):
 
     @classmethod
     def parent(cls, parent_method, child_method):
-        return f'fromParent({child_method}).{parent_method}'
+        return f'fromParent({parent_method}).{child_method}'
+
+    @classmethod
+    def child(cls, parent_method, child_method):
+        return f'childSelector({child_method}).{parent_method}'

testui/support/appium_driver.py

@@ -77,6 +77,7 @@ def set_extra_caps(self, caps=None):
             caps = {}
         for cap in caps:
             self.__desired_capabilities[cap] = caps[cap]
+        return self
 
     def set_app_path(self, path: str):
         root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

testui/support/logger.py

@@ -73,6 +73,16 @@ def log_warn(message, jump_line=False):
 
 
 def log_test_name(message, jump_line=False):
+    log_info(message, jump_line)
+    new_line = ''
+    if jump_line:
+        new_line = '\n'
+    file = open(__file_tests(), 'a+')
+    file.write(f'{message}\n{new_line}')
+    file.close()
+
+
+def log_info(message, jump_line=False):
     now = datetime.now()
     current_time = now.strftime("%Y-%m-%d %H:%M:%S")
     new_line = ''
@@ -93,3 +103,9 @@ def __file_log(log_file='stdout.log'):
     else:
         file_name = f'appium_logs/{log_file}'
     return root_dir + file_name
+
+
+def __file_tests(log_file='report_cases.txt'):
+    root_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + '/'
+    Path(root_dir + "appium_logs").mkdir(parents=True, exist_ok=True)
+    return root_dir + log_file

testui/support/parallel.py

@@ -1,7 +1,8 @@
-import os
-import multiprocessing as mp
 import argparse
+import multiprocessing as mp
+import os
 import time
+import subprocess
 
 from testui.support import logger
 
@@ -13,10 +14,21 @@ def parallel_testui():
         os.remove('report_fails.txt')
     except FileNotFoundError:
         pass
+    try:
+        os.remove('report_cases.txt')
+    except FileNotFoundError:
+        pass
+    test_run_id = None
+    if args.testrail_id is not None:
+        test_run_id = args.testrail_id
+        logger.log_info(f'Starting testrail run {test_run_id}')
+    elif args.testrail is not None:
+        test_run_id = __start_run_id(args, args.testrail)
+        logger.log_info(f'Created testrail run {test_run_id} with testrail name: {args.testrail}')
     number_of_cases = get_total_number_of_cases(args)
-    logger.log_test_name(f'------------ Total Number of Test Cases: {number_of_cases} ---------------')
+    logger.log_info(f'------------ Total Number of Test Cases: {number_of_cases} ---------------')
     start = time.time()
-    __start_processes(args.markers, args)
+    __start_processes(args.markers, args, test_run_id)
     end_time = time.time() - start
     f = open('fails.txt')
     fails = f.read()
@@ -88,8 +100,9 @@ def get_total_number_of_cases(args):
                     number = cases.split(' selected')[0]
                     number_of_cases += int(number)
     if args.single_thread_marker is not None:
-        output = subprocess.run(['pytest', '-m', f'{args.single_thread_marker} {args.general_markers}', f'--collect-only'],
-                                stdout=subprocess.PIPE)
+        output = subprocess.run(
+            ['pytest', '-m', f'{args.single_thread_marker} {args.general_markers}', f'--collect-only'],
+            stdout=subprocess.PIPE)
         response = output.stdout
         string_response = response.__str__()
         if string_response.__contains__(' / '):
@@ -122,10 +135,22 @@ def __arg_parser():
     parser.add_argument('--parallel', type=int,
                         help='number of parallel threads')
 
-    parser.add_argument('--s',  type=__str2bool, nargs='?',
+    parser.add_argument('--s', type=__str2bool, nargs='?',
                         const=True, default=False,
                         help='make more verbose logs (pytest -s)')
 
+    parser.add_argument('--testrail', type=str,
+                        help='Use test rail and specify name. '
+                             'You will have to add a file testrail.cfg in root project dir')
+
+    parser.add_argument('--testrail_id', type=str,
+                        help='Use test rail and specify testrail id. '
+                             'You will have to add a file testrail.cfg in root project dir')
+
+    parser.add_argument('--testrail_pwd', type=str,
+                        help='Use test rail and specify password. '
+                             'You will have to add a file testrail.cfg in root project dir')
+
     parser.add_argument('--general', type=str,
                         help='general flags to put in all threads')
 
@@ -159,13 +184,13 @@ def __arg_parser():
     else:
         logger.log(f'General flags: {args.general}')
     if args.single_thread_marker is not None:
-        logger.log_test_name(f'Single Thread marker: {args.single_thread_marker} {args.general_markers}')
+        logger.log_info(f'Single Thread marker: {args.single_thread_marker} {args.general_markers}')
     return args
 
 
 def __str2bool(v):
     if isinstance(v, bool):
-       return v
+        return v
     if v.lower() in ('yes', 'true', 't', 'y', '1'):
         return True
     elif v.lower() in ('no', 'false', 'f', 'n', '0'):
@@ -174,7 +199,7 @@ def __str2bool(v):
         raise argparse.ArgumentTypeError('Boolean value expected.')
 
 
-def __start_processes(markers: list, args):
+def __start_processes(markers: list, args, test_run_id=None):
     ps = list()
     amount = len(markers) // args.parallel
     amount_plus = 0
@@ -193,18 +218,62 @@ def __start_processes(markers: list, args):
                 if len(markers) > i:
                     tags.append(markers[i])
                     i += 1
-        logger.log_test_name(f'Thread {j} has markers: {tags}')
-        ps.append(mp.Process(target=__process, args=(tags, args, j)))
+        logger.log_info(f'Thread {j} has markers: {tags}')
+        ps.append(mp.Process(target=__process, args=(tags, args, j, test_run_id)))
     for p in ps:
         p.daemon = True
         p.start()
     for p in ps:
         p.join()
     if args.single_thread_marker is not None:
-        __process([args.single_thread_marker], args)
+        __process([args.single_thread_marker], args, 0, test_run_id)
 
 
-def __process(markers: list, args, thread=0):
+def __start_run_id(args, test_run_name):
+    end_marker = ''
+    for i, marker in enumerate(args.markers):
+        if i == len(args.markers):
+            end_marker += f'{marker} {args.general_markers}'
+        else:
+            end_marker += f'{marker} {args.general_markers} or '
+    if args.single_thread_marker is not None:
+        end_marker += f'{args.single_thread_marker} {args.general_markers}'
+    with open('testrail_id_file.txt', 'wb') as out:
+        cmd = ['pytest', '-m', end_marker, '--testrail', f'--tr-config=testrail.cfg',
+               f'--tr-testrun-name={test_run_name}']
+        if args.testrail_pwd is not None:
+            cmd.append(f'--tr-password={args.testrail_pwd}')
+        process = subprocess.Popen(cmd, stdout=out, stderr=out)
+    i = 0
+    while True:
+        time.sleep(0.1)
+        out = open('testrail_id_file.txt')
+        text = out.read()
+        if text.__contains__("New testrun created") or i > 50:
+            out.close()
+            process.terminate()
+            process.wait()
+            os.remove('testrail_id_file.txt')
+            if text.__contains__('ID='):
+                id_test = text.split("ID=")[1]
+                if text.split("ID=")[1].__contains__('\n'):
+                    id_test = text.split("ID=")[1].split("\n")[0]
+                logger.log_info(f'Test run: {id_test}')
+                return id_test
+            raise Exception('Failed to create Test Run')
+        elif text.__contains__("Failed to create testrun"):
+            out.close()
+            process.send_signal(signal=2)
+            process.terminate()
+            process.wait()
+            os.remove('testrail_id_file.txt')
+            logger.log_error(f'Failed to create Test Run: \n {text}')
+            raise Exception(f'Failed to create Test Run')
+        out.close()
+        i += 1
+
+
+def __process(markers: list, args, thread=0, test_run_id=None):
     for i, marker in enumerate(markers):
         try:
             os.remove(f'.my_cache_dir_{thread}/v/cache/lastfailed')
@@ -215,10 +284,15 @@ def __process(markers: list, args, thread=0):
         quiet = '-q'
         if args.s:
             quiet = '-s'
+        testrail = ''
+        if test_run_id is not None:
+            testrail = f'--testrail --tr-config=testrail.cfg --tr-run-id={test_run_id}'
+            if args.testrail_pwd is not None:
+                testrail += f' --tr-password={args.testrail_pwd}'
         cache = f'-o cache_dir=.my_cache_dir_{thread}'
         start_1 = time.time()
-        logger.log(f'Starting: pytest {quiet} -m "{marker} {args.general_markers}" {args.general} {cache}')
-        pr_1 = os.system(f'pytest {quiet} -m "{marker} {args.general_markers}" {args.general} {cache}')
+        logger.log(f'Starting: pytest {quiet} -m "{marker} {args.general_markers}" {testrail} {args.general} {cache}')
+        pr_1 = os.system(f'pytest {quiet} -m "{marker} {args.general_markers}" {testrail} {args.general} {cache}')
         file = open('fails.txt', 'a+')
         file.write(f'{pr_1}')
         file.close()