KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
