Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use schedule_entry method in autotest

Best Python code snippet using autotest_python

report.py

Source:report.py

1import re2import unicodedata3from datetime import date, datetime, timezone4from operator import attrgetter5from dateutil.relativedelta import relativedelta6from reportlab.lib import colors7from reportlab.lib.enums import TA_JUSTIFY8from reportlab.lib.pagesizes import letter9from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle10from reportlab.lib.units import inch11from reportlab.platypus import SimpleDocTemplate, Paragraph, Image, Spacer, Table, TableStyle, KeepTogether12def utc_to_local(utc_dt):13    return utc_dt.replace(tzinfo=timezone.utc).astimezone(tz=None)14def hours_to_days(hours):15    if hours == 0:16        return 'Full Day'17    elif hours <= 2:18        return 'Quarter Day'19    elif hours <= 4:20        return 'Half Day'21    elif hours < 16:22        return 'Full Day'23    elif hours >= 16:24        return str(round(hours / 8, 2)) + ' Days'25class Report:26    def __init__(self, name, filters, sort):27        self.name = name28        self.filters = filters29        self.sort = sort30    def __repr__(self):31        return "<Report {}>".format(self.name)32    @staticmethod33    def clean_str(value):34        value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore').decode('ascii')35        value = re.sub(r'[^\w\s-]', '', value).strip().lower()36        value = re.sub(r'[-\s]+', '-', value)37        return value38    def _generate_report_content(self):39        pass  # either modify this or create a subclass for each report type and override this method there40    def get_dateframe(self, on_or_after, before):41        """42        Dates can be either datetime objects or strings43        """44        if isinstance(on_or_after, str): on_or_after = datetime.strptime(on_or_after, '%Y-%m-%d')45        if isinstance(before, str): before = datetime.strptime(before, '%Y-%m-%d')46        if not before:47            before = datetime.today()48        if on_or_after.day == 1 and (on_or_after + relativedelta(months=1)).strftime('%Y%m%d') == before.strftime('%Y%m%d'):49            month = on_or_after.strftime('%B %Y')50        else:51            month = '{} to {}'.format(on_or_after.strftime('%b %-d, %Y'), (before - relativedelta(days=1)).strftime('%b %-d, %Y'))52        return month53    def format_month(self, on_or_after, before):54        on_or_after = datetime.strptime(on_or_after, '%Y-%m-%d')55        if on_or_after.day == 1 and (on_or_after + relativedelta(months=1)).strftime('%Y-%m-%d') == before:56            return on_or_after.strftime('%B %Y')57        return None58    def save_pdf(self):59        doc = SimpleDocTemplate(60            "output/{}/{} Report.pdf".format(self.output_subdir, self.name),61            pagesize=letter,62            rightMargin=36, leftMargin=36,63            topMargin=72, bottomMargin=40)64        Story = self._generate_report_content()65        doc.build(Story, onFirstPage=self._header_footer, onLaterPages=self._header_footer)66    def _header_footer(self, canvas, doc):67        # Save the state of our canvas so we can draw on it68        canvas.saveState()69        styles = getSampleStyleSheet()70        # Header71        header_text = '<font size=8>'72        header_text += '<strong>Date Run:</strong> {}'.format(datetime.today().strftime('%Y-%m-%d at %-I:%M%p'))73        if self.filters:74            header_text += '<br /><strong>Filters:</strong> {}'.format(' - '.join(['{}: {}'.format(k, v) for k, v in self.filters.items()]))75        if self.sort:76            header_text += '<br /><strong>Sort:</strong> By {}'.format(self.sort.title())77        header_text += '</font>'78        header = Paragraph(header_text, styles['Normal'])79        # header = Paragraph(80        #     '<font size=8><strong>Date Run:</strong> {}<br />'81        #     '<strong>Filters:</strong> {}<br />'82        #     '<strong>Sort:</strong> By {}</font>'.format(83        #         datetime.today().strftime('%Y-%m-%d at %-I:%M%p'),84        #         ' - '.join(['{}: {}'.format(k, v) for k, v in self.filters.items()]),85        #         self.sort86        #     ), styles['Normal'])87        w, h = header.wrap(doc.width, doc.topMargin)88        header.drawOn(canvas, doc.leftMargin, doc.height + doc.topMargin - h)89        # Footer90        footer = Paragraph('Â© {} Consortium for Education Change. All rights reserved.'.format(date.today().year),91                           styles['Normal'])92        w, h = footer.wrap(doc.width, doc.bottomMargin)93        footer.drawOn(canvas, doc.leftMargin, h + 16)94        # Release the canvas95        canvas.restoreState()96class UpcomingSchedule(Report):97    def __init__(self, schedule_entries, filters, sort='dateStart'):98        self.schedule_entries = schedule_entries99        self.filters = filters100        self.sort = sort101        self.name = 'Upcoming Schedule'102        self.output_subdir = 'ConnectWise'103    def save_pdf(self):104        doc = SimpleDocTemplate(105            "output/{}/{} By {} Report.pdf".format(self.output_subdir, self.name, self.sort),106            pagesize=letter,107            rightMargin=36, leftMargin=36,108            topMargin=72, bottomMargin=40)109        Story = self._generate_report_content()110        doc.build(Story, onFirstPage=self._header_footer, onLaterPages=self._header_footer)111    def _generate_report_content(self):112        Story = []113        styles = getSampleStyleSheet()114        styles.add(ParagraphStyle(name='Justify', alignment=TA_JUSTIFY))115        logo = "static/img/BeyondComplianceAdminAcademyStorylineLogoHighRes.gif"116        titletext = "{} By {} Report<br />{} Team".format(self.name, self.sort, self.filters['Team'])117        # description_text = "This is a schedule report"118        im = Image(logo, 2 * inch, 0.84 * inch)119        count_consultants = len(set([schedule_entry.member['identifier'] for schedule_entry in self.schedule_entries]))120        count_companies = len(set([schedule_entry.name.split(' / ')[0] for schedule_entry in self.schedule_entries]))121        # Story.append(im)122        Story.append(Spacer(1, 12))123        Story.append(Paragraph(titletext, styles["Title"]))124        Story.append(Spacer(1, 12))125        Story.append(Paragraph('Schedule Entries: {}'.format(len(self.schedule_entries)), styles["Normal"]))126        Story.append(Paragraph('Consultants Scheduled: {}'.format(count_consultants), styles["Normal"]))127        Story.append(Paragraph('Districts: {}'.format(count_companies), styles["Normal"]))128        # schedule_table = [[129        #     Paragraph('<strong>Date</strong>', styles["Normal"]),130        #     Paragraph('<strong>Consultant</strong>', styles["Normal"]),131        #     Paragraph('<strong>District/Project/Phase(s)/Ticket</strong>', styles["Normal"])132        # ]]133        # sorted_schedule_entries = sorted(self.schedule_entries, key=lambda x: x.member['identifier'][1:])134        sorted_schedule_entries = sorted(self.schedule_entries, key=attrgetter('dateStart'))135        if self.sort.lower() == 'consultant':136            # sorted_schedule_entries = sorted(sorted_schedule_entries, key=lambda x: x.member['identifier'][1:])137            member_identifiers = sorted(138                set([schedule_entry.member['identifier'] for schedule_entry in sorted_schedule_entries]))139            for member_identifier in member_identifiers:140                schedule_table = []141                for schedule_entry in sorted(self.schedule_entries, key=attrgetter('dateStart')):142                    if schedule_entry.member['identifier'] == member_identifier:143                        schedule_name_arr = schedule_entry.name.split(' / ')144                        company = schedule_name_arr.pop(0)145                        schedule_name = ' / '.join(schedule_name_arr)146                        schedule_table.append(147                            [Paragraph('{}<br />{}'.format(148                                utc_to_local(149                                    datetime.strptime(schedule_entry.dateStart, '%Y-%m-%dT%H:%M:%SZ')).strftime(150                                    '%b %-d, %-I:%M%p'),151                                hours_to_days(schedule_entry.hours)152                            ), styles["Normal"]),153                                Paragraph(company, styles["Normal"]),154                                Paragraph('<font size=9>%s</font>' % schedule_name, styles["Normal"])]155                        )156                t = Table(schedule_table, [1.3 * inch, 1.6 * inch, 4.3 * inch])157                t.setStyle(TableStyle([('VALIGN', (0, 0), (-1, -1), 'TOP'),158                                       ('INNERGRID', (0, 0), (-1, -1), 0.25, colors.black),159                                       ('BOX', (0, 0), (-1, -1), 0.25, colors.black),160                                       ]))161                number_of_entries = '{} Schedule Entries'.format(len(schedule_table)) if len(schedule_table) > 1 else '1 Schedule Entry'162                member_section = []163                member_section.append(Spacer(1, 12))164                member_section.append(Paragraph('<strong>{}</strong> - {}:'.format(member_identifier, number_of_entries), styles["Normal"]))165                member_section.append(Spacer(1, 4))166                member_section.append(t)167                Story.append(KeepTogether(member_section))168        elif self.sort.lower() == 'date' or self.sort.lower() == 'month':169            months = sorted(set(170                [utc_to_local(datetime.strptime(schedule_entry.dateStart, '%Y-%m-%dT%H:%M:%SZ')).strftime('%Y-%-m-1')171                 for schedule_entry in sorted_schedule_entries]))172            for month in months:173                month_table = []174                for schedule_entry in sorted(self.schedule_entries, key=attrgetter('dateStart')):175                    schedule_name = schedule_entry.name.split(' / ')176                    schedule_company_name = schedule_name.pop(0)177                    schedule_name = ' / '.join(schedule_name)178                    if month == utc_to_local(179                            datetime.strptime(schedule_entry.dateStart, '%Y-%m-%dT%H:%M:%SZ')).strftime('%Y-%-m-1'):180                        month_table.append(181                            [Paragraph('{}<br />{}'.format(182                                utc_to_local(183                                    datetime.strptime(schedule_entry.dateStart, '%Y-%m-%dT%H:%M:%SZ')).strftime(184                                    '%b %-d, %-I:%M%p'),185                                hours_to_days(schedule_entry.hours)186                            ), styles["Normal"]),187                                Paragraph(schedule_entry.member['identifier'], styles["Normal"]),188                                Paragraph('<font size=9>%s</font>' % schedule_name, styles["Normal"])]189                        )190                        t = Table(month_table, [1.3 * inch, 1.2 * inch, 4.7 * inch])191                        t.setStyle(TableStyle([('VALIGN', (0, 0), (-1, -1), 'TOP'),192                                               ('INNERGRID', (0, 0), (-1, -1), 0.25, colors.black),193                                               ('BOX', (0, 0), (-1, -1), 0.25, colors.black),194                                               ]))195                        # Story.append(Paragraph('<strong>{}</strong>'.format(company_name), styles["Normal"]))196                number_of_entries = '{} Schedule Entries'.format(len(month_table)) if len(197                    month_table) > 1 else '1 Schedule Entry'198                Story.append(Spacer(1, 12))199                print(month)200                themonth = date(2017, int(month.split('-')[1]), 1).strftime('%B %Y')201                print(themonth)202                Story.append(Paragraph('<strong>{}</strong> - {}:'.format(themonth, number_of_entries), styles["Normal"]))203                Story.append(Spacer(1, 4))204                Story.append(t)205                # Story.append(company_section)206                # Story.append(KeepTogether(company_section))207        elif self.sort.lower() == 'district' or self.sort.lower() == 'company' or self.sort.lower() == 'organization':208            company_names = sorted(209                set([schedule_entry.name.split(' / ')[0] for schedule_entry in sorted_schedule_entries]))210            for company_name in company_names:211                company_table = []212                for schedule_entry in sorted(self.schedule_entries, key=attrgetter('dateStart')):213                    schedule_name = schedule_entry.name.split(' / ')214                    schedule_company_name = schedule_name.pop(0)215                    schedule_name = ' / '.join(schedule_name)216                    if schedule_company_name == company_name:217                        company_table.append(218                            [Paragraph('{}<br />{}'.format(219                                utc_to_local(220                                    datetime.strptime(schedule_entry.dateStart, '%Y-%m-%dT%H:%M:%SZ')).strftime(221                                    '%b %-d, %-I:%M%p'),222                                hours_to_days(schedule_entry.hours)223                            ), styles["Normal"]),224                                Paragraph(schedule_entry.member['identifier'], styles["Normal"]),225                                Paragraph('<font size=9>%s</font>' % schedule_name, styles["Normal"])]226                        )227                        t = Table(company_table, [1.3 * inch, 1.2 * inch, 4.7 * inch])228                        t.setStyle(TableStyle([('VALIGN', (0, 0), (-1, -1), 'TOP'),229                                               ('INNERGRID', (0, 0), (-1, -1), 0.25, colors.black),230                                               ('BOX', (0, 0), (-1, -1), 0.25, colors.black),231                                               ]))232                # Story.append(Paragraph('<strong>{}</strong>'.format(company_name), styles["Normal"]))233                number_of_entries = '{} Schedule Entries'.format(len(company_table)) if len(234                    company_table) > 1 else '1 Schedule Entry'235                # company_section = []236                Story.append(Spacer(1, 12))237                Story.append(Paragraph('<strong>{}</strong> - {}:'.format(company_name, number_of_entries), styles["Normal"]))238                Story.append(Spacer(1, 4))239                Story.append(t)240                # Story.append(KeepTogether(company_section))241                # sorted_schedule_entries = sorted(sorted_schedule_entries, key=attrgetter('name'))242        else:243            sorted_schedule_entries = sorted(sorted_schedule_entries, key=attrgetter(self.sort))244        # schedule_table = []245        # for schedule_entry in sorted_schedule_entries:246        # [schedule_table.append(247        #     [248        #         Paragraph(249        #             '{}<br />{}'.format(250        #                 utc_to_local(datetime.strptime(schedule_entry.dateStart, '%Y-%m-%dT%H:%M:%SZ')).strftime('%b %-d, %-I:%M%p'),251        #                 hours_to_days(schedule_entry.hours)252        #             ), styles["Normal"]),253        #         Paragraph(schedule_entry.member['identifier'], styles["Normal"]),254        #         Paragraph('<font size=9>%s</font>' % schedule_entry.name, styles["Normal"])255        #     ]) for schedule_entry in sorted_schedule_entries256        # ]257        #258        # t = Table(schedule_table, [1.2 * inch, 1 * inch, 5 * inch])259        # t.setStyle(TableStyle([('VALIGN', (0, 0), (-1, -1), 'TOP'),260        #                        ('INNERGRID', (0, 0), (-1, -1), 0.25, colors.black),261        #                        ('BOX', (0, 0), (-1, -1), 0.25, colors.black),262        #                        ]))263        # for schedule_entry in self.schedule_entries:264        #     Story.append(265        #         Paragraph(schedule_entry.name, styles["Normal"])266        #     )267        # Story.append(Paragraph(description_text, styles["Normal"]))268        Story.append(Spacer(1, 12))269        # Story.append(t)...

run_document_async_conc.py

Source:run_document_async_conc.py

1import logging2import random3import time4import sys5import keyring6from datetime import datetime7from microstrategy_api.task_proc.executable_base import ExecutableBase8from microstrategy_api.task_proc.object_type import ObjectSubType9from microstrategy_api.task_proc.status import Status10from microstrategy_api.task_proc.task_proc import TaskProc11class ScheduleEntry(object):12    def __init__(self,13                 folder_object: TaskProc.FolderObject,14                 executable_object: ExecutableBase,15                 prompts=None,16                 ou_name=None,17                 ):18        self.folder_object = folder_object19        self.executable_object = executable_object20        self.prompts = prompts21        self.ou_name = ou_name22        self.message = None23        self.start_time = None24        self.end_time = None25        self.done = False26    def __str__(self):27        if self.ou_name:28            result = "{self.folder_object} for {self.ou_name}".format(self=self)29        else:30            if self.prompts:31                result = "{self.folder_object} with {self.prompts}".format(self=self)32            else:33                result = "{self.folder_object}".format(self=self)34        if self.message:35            result += " Status={}".format(self.message)36        if self.start_time:37            result += " start_time={}".format(self.start_time)38        if self.end_time:39            result += " end_time={}".format(self.end_time)40        if self.start_time and self.end_time:41            result += " duration={}".format(self.end_time - self.start_time)42        return result43class RunConcurrent(object):44    OU_GUID = '7039371C4B5CC07DC6682D9C0EC8F45C'45    OU_DICT = {46        # "Angola":                           "XOivy2uDpMF",47        # "Asia Regional Program":            "iD2i0aynOGm",48        # "Botswana":                         "l1KFEXKI4Dg",49        # "Burma":                            "wChmwjpXOw2",50        # "Burundi":                          "Qh4XMQJhbk8",51        # "Cambodia":                         "XWZK2nop7pM",52        "Cameroon":                         "bQQJe0cC1eD",53        # "Caribbean Region":                 "nBo9Y4yZubB",54        # "Central America Region":           "vSu0nPMbq7b",55        # "Central Asia Region":              "t25400wXrNB",56        "Cote dIvoire":                     "ds0ADyc9UCU",57        "Democratic Republic of the Congo": "ANN4YCOufcP",58        # "Dominican Republic":               "NzelIFhEv3C",59        "Ethiopia":                         "IH1kchw86uA",60        # "Ghana":                            "y3zhsvdXlhN",61        # "Guyana":                           "PeOHqAwdtez",62        # "Haiti":                            "JTypsdEUNPw",63        # "India":                            "skj3e4YSiJY",64        # "Indonesia":                        "W73PRZcjFIU",65        # "Kenya":                            "HfVjCurKxh2",66        # "Lesotho":                          "qllxzIjjurr",67        # "Malawi":                           "lZsCb6y0KDX",68        # "Mozambique":                       "h11OyvlPxpJ",69        "Namibia":                          "FFVkaV9Zk1S",70        "Nigeria":                          "PqlFzhuPcF1",71        # "Papua New Guinea":                 "cl7jVQOW3Ks",72        "Rwanda":                           "XtxUYCsDWrR",73        # "South Africa":                     "cDGPF739ZZr",74        # "South Sudan":                      "WLG0z5NxQs8",75        # "Swaziland":                        "V0qMZH29CtN",76        "Tanzania":                         "mdXu6iCbn2G",77        "Uganda":                           "FETQ6OmnsKB",78        # "Ukraine":                          "ligZVIYs2rL",79        # "Vietnam":                          "YM6xn5QxNpY",80        "Zambia":                           "f5RoebaDLMx",81        "Zimbabwe":                         "a71G4Gtcttv"82    }83    def __init__(self,84                 max_concurrent: int = 585                 ):86        logging.basicConfig()87        log = logging.getLogger(__name__)88        log.setLevel(logging.WARNING)89        log.root.setLevel(logging.WARNING)90        self.log = log91        self.max_concurrent = max_concurrent92        self.cmd = None93        self.project = None94        self.task_client = None95        self._schedule_entries = list()96    def add_to_schedule(self, schedule_entry):97        self._schedule_entries.append(schedule_entry)98    def _wait(self):99        jobs_are_running = True100        while jobs_are_running:101            time.sleep(15)102            jobs = self.cmd.find_running_jobs(project=self.project)103            if len(jobs) == 0:104                jobs_are_running = False105            else:106                self.log.debug("{} jobs are still active".format(len(jobs)))107    def execute_entry(self, schedule_entry: ScheduleEntry):108        self.log.debug("Executing {}".format(schedule_entry))109        schedule_entry.start_time = datetime.now()110        schedule_entry.message = schedule_entry.executable_object.execute_async(111            element_prompt_answers=schedule_entry.prompts,112            # refresh_cache=True,113            max_wait_secs=1,114        )115    def execute_schedule(self):116        self.log.debug("Starting cache updates")117        done = False118        running_cnt = 0119        error_jobs = []120        while not done:121            pending_cnt = 0122            #if folder_obj.object_subtype == ObjectSubType.ReportCube:123            for schedule_entry in self._schedule_entries:124                # Check if we need to start or check on running job125                if schedule_entry.message is None:126                    pending_cnt += 1127                    if running_cnt < self.max_concurrent:128                        running_cnt += 1129                        self.execute_entry(schedule_entry)130                elif schedule_entry.message.status not in {Status.Result, Status.Prompt}:131                    pending_cnt += 1132                    self.log.debug("Checking on {} which was {}".format(schedule_entry, schedule_entry.message))133                    schedule_entry.message.update_status(max_wait_ms=250)134                if not schedule_entry.done and schedule_entry.message is not None:135                    # Check for status on running jobs136                    if schedule_entry.message.status == Status.Result:137                        schedule_entry.done = True138                        schedule_entry.executable_object.execute(arguments={'messageID': schedule_entry.message.guid})139                        schedule_entry.end_time = datetime.now()140                        self.log.debug("Job {} completed".format(schedule_entry))141                        running_cnt -= 1142                        pending_cnt -= 1143                    elif schedule_entry.message.status == Status.Prompt:144                        schedule_entry.done = True145                        self.log.error("{} has un-resolved prompts".format(schedule_entry))146                        error_jobs.append(schedule_entry)147                        pending_cnt -= 1148                        running_cnt -= 1149                    elif schedule_entry.message.status == Status.ErrMsg:150                        schedule_entry.done = True151                        schedule_entry.end_time = datetime.now()152                        self.log.error("{} has an error".format(schedule_entry))153                        error_jobs.append(schedule_entry)154                        pending_cnt -= 1155                        running_cnt -= 1156            time.sleep(0.1)157            if pending_cnt == 0:158                done = True159            else:160                self.log.debug("{} pending jobs, {} running jobs".format(pending_cnt, running_cnt))161        # self.log.info("-" * 80)162        # self.log.info("Job Summary")163        # self.log.info("-" * 80)164        # for schedule_entry in self._schedule_entries:165        #     print("{s.ou_name},{s.start_time},{s.end_time},{duration}".format(s=schedule_entry, duration=schedule_entry.end_time-schedule_entry.start_time))166        #167        # self.log.info("-" * 80)168        # self.log.info("-" * 80)169        #170        # if error_jobs:171        #     self.log.error("-" * 80)172        #     self.log.error("-" * 80)173        #     self.log.error("Summary of all jobs with errors")174        #     for schedule_entry in error_jobs:175        #         self.log.error("Error in job {}".format(schedule_entry))176        #     self.log.error("-" * 80)177    def run(self, jobs_to_create=50):178        project = 'my_project'179        user_name = 'Administrator'180        password = keyring.get_password('Development', user_name)181        self.project = project182        self.task_client = TaskProc(base_url='https://my_hostname/MicroStrategy/asp/TaskProc.aspx?',183                                    server='my_server',184                                    project_name=project,185                                    username=user_name,186                                    password=password,187                                    concurrent_max=self.max_concurrent * 2,188                                    )189        folder_objs = contents = self.task_client.get_folder_contents(190            "\\Public Objects\\Reports\\",191            flatten_structure=True,192            recursive=False,193            type_restriction={ObjectSubType.ReportWritingDocument}194        )195        test_reports = [196            'A',197            'B',198            'C',199        ]200        for conc in range(15, 71, 5):201            self.max_concurrent = conc202            jobs_to_create = conc * 3203            for folder_obj in folder_objs:204                if folder_obj.object_subtype != ObjectSubType.Folder and folder_obj.name in test_reports:205                    executable_object = self.task_client.get_executable_object(folder_obj)206                    ou_prompt = None207                    prompts = executable_object.get_prompts()208                    for prompt in prompts:209                        if prompt.attribute.guid == self.OU_GUID:210                            ou_prompt = prompt211                    self._schedule_entries = list()212                    #self.log.info("Scheduling jobs")213                    for _ in range(jobs_to_create):214                        ou_name = random.choice(list(self.OU_DICT.keys()))215                        ou_uid = self.OU_DICT[ou_name]216                        self.add_to_schedule(217                            ScheduleEntry(folder_obj, executable_object, prompts={ou_prompt: ou_uid}, ou_name=ou_name)218                        )219                    #self.log.info("Running jobs")220                    test_start = datetime.now()221                    self.execute_schedule()222                    test_end = datetime.now()223                    total_time = 0224                    max_time = None225                    for schedule_entry in self._schedule_entries:226                        duration = schedule_entry.end_time - schedule_entry.start_time227                        if max_time is None:228                            max_time = duration229                        elif duration > max_time:230                            max_time = duration231                        total_time += (duration.seconds + duration.microseconds/1E6)232                    print("{name},{concurrent},{avg_time},{max_time},{test_duration}".format(233                        name=folder_obj.name,234                        concurrent=self.max_concurrent,235                        avg_time=1.0 * total_time / len(self._schedule_entries),236                        max_time=max_time,237                        test_duration=test_end - test_start,238                    ))239        #self.log.info("Done")240if __name__ == '__main__':241    print("name,concurrent,avg_time,max_time,test_duration")...

schedule.py

Source:schedule.py

1from decimal import Decimal2from pprint import pprint3from dateutil.relativedelta import relativedelta4from lib.connectwise_py.connectwise.activity import Activity5from lib.connectwise_py.connectwise.ticket import Ticket6from lib.connectwise_py.connectwise.connectwise import Connectwise7from datetime import date, timedelta, datetime8class ScheduleEntry:9    def __init__(self, id, **kwargs):10        self.id = id11        self.objectId = None12        self.where = {'name': None}13        self.member = {'identifier': None}14        self.dateStart = None15        for kwarg in kwargs:16            setattr(self, kwarg, kwargs[kwarg])17    def __repr__(self):18        return "<Schedule Entry {}>".format(self.id)19    def to_dict(self, include_self=False):20        """21        Get a representation of the ScheduleEntry object as a Python Dictionary,22        including calculated values from methods.23        :param include_self: when True, include the original object under the dictionary24        key 'self' so that you have it in case you need to reference it later25        """26        schedule_dict = {}27        schedule_dict['days'] = self.days()28        schedule_dict['formatted_days'] = self.formatted_days()29        schedule_dict['each_calendar_date'] = self.each_calendar_date()30        schedule_dict['calendar_days'] = self.calendar_days()31        schedule_dict['duration_per_calendar_day'] = self.duration_per_calendar_day()32        if include_self: schedule_dict['self'] = self33        return {**vars(self), **schedule_dict}34    @classmethod35    def fetch_all(cls):36        return [cls(**schedule_entry) for schedule_entry in Connectwise.submit_request('schedule/entries')]37    @classmethod38    def fetch_by_object_ids(cls, object_ids, on_or_after=None, before=None):39        # i = 140        # schedule_entries = []41        # print(ticket_ids)42        # while i < len(ticket_ids):43        #     print(ticket_ids[i-1:10])44        #     schedule_entries.extend(CWSchedule.fetch_by_object_ids(ticket_ids[i-1:10], on_or_after, before))45        #     i += 1046        # self.schedule_entries = schedule_entries47        if len(object_ids) > 10:48            raise IOError('Cannot lookup more than 10 objectIds at once')49        conditions = ['({})'.format(50            'objectId={}'.format(' or objectId='.join('{}'.format(_id) for _id in object_ids)))]51        if on_or_after: conditions.append('dateStart>=[{}]'.format(on_or_after))52        if before: conditions.append('dateStart<[{}]'.format(before))53        return [cls(**schedule_entry) for schedule_entry in54                Connectwise.submit_request('schedule/entries', conditions)]55    @classmethod56    def fetch_by_object_id(cls, object_id, on_or_after=None, before=None):57        conditions = ['objectId={}'.format(object_id)]58        if on_or_after: conditions.append('dateStart>=[{}]'.format(on_or_after))59        if before: conditions.append('dateStart<[{}]'.format(before))60        return [cls(**schedule_entry) for schedule_entry in Connectwise.submit_request('schedule/entries', conditions)]61    @classmethod62    def fetch_by_date_range(cls, on_or_after, before):63        conditions = 'dateStart>=[{}] and dateStart<[{}]'.format(on_or_after, before)64        return [cls(**schedule_entry) for schedule_entry in Connectwise.submit_request('schedule/entries', conditions)]65    @classmethod66    def fetch_this_fy(cls, member_identifier=None):67        on_or_after, before = Connectwise.current_fy()68        conditions = 'dateStart>=[{}] and dateStart<[{}]'.format(on_or_after, before)69        if member_identifier: conditions += ' and member/identifier="{}"'.format(member_identifier)70        return [cls(**schedule_entry) for schedule_entry in Connectwise.submit_request('schedule/entries', conditions)]71    @classmethod72    def fetch_remaining_fy(cls):73        on_or_after, before = Connectwise.current_fy()74        on_or_after = date.today().strftime('%Y-%m-%d')75        conditions = 'dateStart>=[{}] and dateStart<[{}]'.format(on_or_after, before)76        return [cls(**schedule_entry) for schedule_entry in Connectwise.submit_request('schedule/entries', conditions)]77    @classmethod78    def fetch_upcoming(cls, days=30):79        on_or_after = date.today().strftime('%Y-%m-%d')80        before = date.today() + timedelta(days=days)81        conditions = 'dateStart>=[{}] and dateStart<[{}]'.format(on_or_after, before)82        return [cls(**schedule_entry) for schedule_entry in Connectwise.submit_request('schedule/entries', conditions)]83    @classmethod84    def fetch_upcoming_by_board_names(cls, board_names=[], days=30):85        schedule_entries = cls.fetch_upcoming(days)86        ticket_ids = set([schedule_entry.objectId for schedule_entry in schedule_entries])87        tickets = Ticket.fetch_by_ids(ticket_ids)88        board_ticket_ids = [ticket.id for ticket in tickets if ticket.board['name'] in board_names]89        return [schedule_entry for schedule_entry in schedule_entries if schedule_entry.objectId in board_ticket_ids]90    @classmethod91    def fetch_upcoming_by_project_ids(cls, project_ids=[], days=30):92        schedule_entries = cls.fetch_upcoming(days)93        ticket_ids = set([schedule_entry.objectId for schedule_entry in schedule_entries])94        tickets = Ticket.fetch_by_ids(ticket_ids)95        project_ticket_ids = [ticket.id for ticket in tickets if ticket.project and ticket.project['id'] in project_ids]96        return [schedule_entry for schedule_entry in schedule_entries if schedule_entry.objectId in project_ticket_ids]97    @classmethod98    def fetch_by_company_id(cls, company_id, on_or_after=None, before=None):99        conditions = []100        if on_or_after:101            conditions.append('dateStart>=[{}]'.format(on_or_after))102        if before:103            conditions.append('dateStart<[{}]'.format(before))104        conditions = ' and '.join(conditions)105        company_tickets = Ticket.fetch_by_company_id(company_id)106        company_activities = Activity.fetch_by_company_id(company_id)107        schedule_entries = [cls(**schedule_entry) for schedule_entry in Connectwise.submit_request('schedule/entries', conditions)]108        return [s for s in schedule_entries109                if s.objectId == company_id or110                s.objectId in [ticket.id for ticket in company_tickets] or111                s.objectId in [activity.id for activity in company_activities]112                ]113    def days(self):114        return Decimal(round(self.hours / 8, 2))115    def formatted_days(self):116        days = self.days()117        if days == 1.0:118            return 'Full Day'119        elif days == 0.5:120            return 'Half Day'121        elif days == 0.25:122            return 'Quarter Day'123        else:124            return '{} Days'.format(days)125    def each_calendar_date(self):126        start_date = datetime.strptime(self.dateStart[:10], '%Y-%m-%d')127        dates = []128        while start_date.strftime('%Y-%m-%d') <= self.dateEnd[:10]:129            dates.append(start_date.strftime('%Y-%m-%d'))130            start_date += relativedelta(days=1)131        return dates132    def calendar_days(self):133        return (datetime.strptime(self.dateEnd[:10], '%Y-%m-%d') - datetime.strptime(134            self.dateStart[:10], '%Y-%m-%d')).days + 1135    def duration_per_calendar_day(self):136        return self.days() / self.calendar_days()137    def get_charge_to_info(self, tickets=[], activities=[], charge_codes=[], return_type='string', include_company=True,138                           include_project_name=True, bold_first_item=False):139        """Optionally include a list of pre-fetched Tickets and/or Activities to prevent it from re-querying CW.140                return_type must be 'list' or 'string'"""141        return Connectwise.get_charge_to_info(self, tickets, activities, charge_codes, return_type, include_company,...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.