How to use action_options method in avocado

Best Python code snippet using avocado_python

post_actions.py

Source:post_actions.py

1from datetime import datetime, timedelta2from math import floor3from itertools import cycle4from typing import List, Any, Union5from pydantic import BaseModel, Field, root_validator6from astropy.coordinates import SkyCoord7from astropy import units as u8from try_pipelining.data_models import (9    ScienceAlert,10    SchedulingBlock,11    ObservationBlock,12    Coords,13    WobbleOptions,14)15from try_pipelining.observation_windows import ObservationWindow16available_post_actions = {}17available_post_action_options = {}18def register_post_action(cls):19    available_post_actions.update({cls.__name__: cls})20    return cls21def register_post_action_option(cls):22    available_post_action_options.update({cls.__name__.replace("Options", ""): cls})23    return cls24class Wobble(BaseModel):25    offsets: List[float]26    angles: List[float]27    @root_validator28    def validate_same_length(cls, values):29        if len(values.get("offsets")) != len(values.get("angles")):30            raise AttributeError("offsets and angles lists should be of same length.")31        return values32class Proposal(BaseModel):33    proposal_id: int34    proposal_class: str = Field(..., max_length=1)35    proposal_rank: float = Field(..., ge=0)36@register_post_action_option37class CreateWobbleSchedulingBlockOptions(BaseModel):38    wobble: Wobble39    proposal: Proposal40@register_post_action_option41class CreateObservationBlocksOptions(BaseModel):42    min_block_duration_minutes: float = Field(..., ge=0)43    max_block_duration_minutes: float = Field(..., ge=0)44class PostAction:45    def __init__(self, science_alert, action_type, action_options):46        self.science_alert: ScienceAlert = science_alert47        self.action_type: str = action_type48        self.action_options: Union[49            CreateWobbleSchedulingBlockOptions,50            CreateObservationBlocksOptions,51        ] = action_options52        self.validate()53    def validate(self):54        assert isinstance(55            self.action_options, available_post_action_options[self.action_type]56        )57    def run(self, task_result: Any):58        raise NotImplementedError(59            "A Raw Post Action class instance should not be used.\60            Use Chiled classes of PostAction only."61        )62@register_post_action63class CreateWobbleSchedulingBlock(PostAction):64    def run(self, task_result: ObservationWindow) -> SchedulingBlock:65        action_options: CreateWobbleSchedulingBlockOptions = self.action_options66        sb_dict = {67            "coords": self.science_alert.coords,68            "time_constraints": {69                "start_time": task_result.start_time,70                "end_time": task_result.end_time,71            },72            "wobble_options": action_options.wobble,73            "proposal": action_options.proposal,74        }75        return SchedulingBlock(**sb_dict)76@register_post_action77class CreateObservationBlocks(PostAction):78    def run(self, task_result: SchedulingBlock) -> List[ObservationBlock]:79        base_target_coords: Coords = task_result.coords80        sb_start: datetime = task_result.time_constraints.start_time81        sb_end: datetime = task_result.time_constraints.end_time82        sb_duration_sec = (sb_end - sb_start).total_seconds()83        wobble_opts: WobbleOptions = task_result.wobble_options84        action_options = self.action_options85        assert isinstance(action_options, CreateObservationBlocksOptions)86        n_blocks_max = floor(87            (sb_duration_sec / 60.0) / action_options.min_block_duration_minutes88        )89        #  TODO: get the optimal number of observation blocks (i.e. symetrical and not too short)90        #  going with max number of blocks for now...91        cycle_wobble_offset = cycle(wobble_opts.offsets)92        cycle_wobble_angles = cycle(wobble_opts.angles)93        base_target = SkyCoord(94            ra=base_target_coords.raInDeg * u.deg,95            dec=base_target_coords.decInDeg * u.deg,96            frame="fk5",97        )98        obs = []99        for i in range(n_blocks_max):100            coord_with_offset = base_target.directional_offset_by(101                position_angle=next(cycle_wobble_angles) * u.deg,102                separation=next(cycle_wobble_offset) * u.deg,103            )104            ra = coord_with_offset.ra / u.deg105            dec = coord_with_offset.dec / u.deg106            ra_in_deg = ra if ra > 0 else ra + 180107            dec_in_deg = dec if dec > 0 else dec + 180108            obs.append(109                ObservationBlock(110                    start_time=sb_start111                    + timedelta(minutes=i * action_options.min_block_duration_minutes),112                    end_time=sb_start113                    + timedelta(114                        minutes=(i + 1) * action_options.min_block_duration_minutes115                    ),116                    ra_target_deg=ra_in_deg,117                    dec_target_deg=dec_in_deg,118                )119            )...

namecheap.py

Source:namecheap.py

1import logging2import lib.utilities as utilities3from .main import DNS4class Namecheap(DNS):5	NC_ACTION_TEMPLATE = {6		'provider_name': 'namecheap',7		'namecheap': {8			'auth_token': utilities.parse_config('namecheap', 'auth_token'),9			'auth_username': utilities.parse_config('namecheap', 'auth_username'),10			'auth_client_ip': utilities.parse_config('namecheap', 'auth_client_ip'),11			'auth_sandbox': False12		}13	}14	def run(self):15		self.provider = 'namecheap'16		self.init()17	def check_exists(self, record, subdomain, show=False):18		action_options = self.NC_ACTION_TEMPLATE.copy()19		action_options['action'] = 'list'20		action_options['type'] = record21		action_options['domain'] = self.base_domain22		action_options['name'] = subdomain23		response = self.send_request(action_options)24		if response is None or len(response) <= 0:25			return False26		for r in response:27			# If the record type matches28			if r['type'] == action_options['type']:29				if record == 'TXT' and r['name'] == self.challenge_domain and r['content'] == self.content:30					return True31				elif record == 'A' and r['name'] == "{0}.{1}".format(action_options['name'], action_options['domain']):32					return True33		return False34	def do_create(self, record, subdomain, content):35		# Check if record exists36		exists = self.check_exists(record, subdomain)37		if exists is True and record != 'TXT':38			logging.info("{0} record already exists for {1}, not creating".format(record, self.full_domain))39			return True40		# If TXT record, delete before creating41		elif exists is True and record == 'TXT':42			logging.info("{0} record already exists for {1}, removing".format(record, self.full_domain))43			txt_delete = self.do_delete(record, self.challenge_domain)44			if txt_delete is False:45				logging.error("Failed to delete old TXT record for {0}".format(self.challenge_domain))46				return False47			logging.info("Deleted old TXT record for {0}".format(self.challenge_domain))48		# Record does not exist, create it49		else:50			logging.info("{0} record does not exist for {1}, creating".format(record, subdomain))51		action_options = self.NC_ACTION_TEMPLATE.copy()52		action_options['action'] = 'create'53		action_options['type'] = record54		action_options['ttl'] = 30055		action_options['domain'] = self.base_domain56		action_options['name'] = subdomain57		action_options['content'] = content58		logging.info("Creating {0} record with {1} content".format(record, content))59		response = self.send_request(action_options)60		if response is not True:61			return False62		else:63			return True64	def do_delete(self, record, subdomain):65		# A record is restricted, DO NOT REMOVE66		if utilities.is_restricted(self.full_domain) is True and record == 'A':67			return None68		# Check if record exists69		exists = self.check_exists(record, subdomain)70		if exists is False:71			return None72		# A record exists, remove it73		logging.info("{0} record exists for {1}.{2}, removing".format(record, subdomain, self.base_domain))74		action_options = self.NC_ACTION_TEMPLATE.copy()75		action_options['action'] = 'delete'76		action_options['type'] = record77		action_options['domain'] = self.base_domain78		action_options['name'] = subdomain79		response = self.send_request(action_options)80		if response is not True:81			return False82		else:...

godaddy.py

Source:godaddy.py

1import logging2import lib.utilities as utilities3from .main import DNS4class GoDaddy(DNS):5	GD_ACTION_TEMPLATE = {6		'provider_name': 'godaddy',7		'godaddy': {8			'auth_key': utilities.parse_config('godaddy', 'auth_key'),9			'auth_secret': utilities.parse_config('godaddy', 'auth_secret')10		}11	}12	def run(self):13		self.provider = 'godaddy'14		self.init()15	def check_exists(self, record, subdomain, show=False):16		action_options = self.GD_ACTION_TEMPLATE.copy()17		action_options['action'] = 'list'18		action_options['type'] = record19		action_options['domain'] = self.base_domain20		action_options['name'] = subdomain21		response = self.send_request(action_options)22		if response is None or len(response) <= 0:23			return False24		if show is True:25			logging.info(response)26		print(response[0]['content'])27		if response[0]['content'] != self.content:28			return False29		return True30	def do_create(self, record, subdomain, content):31		# Check if record exists32		exists = self.check_exists(record, subdomain)33		if exists is True and record != 'TXT':34			return None35		# If TXT record, delete before creating36		elif exists is True and record == 'TXT':37			logging.info("{0} record already exists for {1}, removing".format(record, self.full_domain))38			txt_delete = self.do_delete(record, self.challenge_domain)39			if txt_delete is False:40				logging.error("Failed to delete old TXT record for {0}".format(self.challenge_domain))41				return False42			logging.info("Deleted old TXT record for {0}".format(self.challenge_domain))43		# Record does not exist, create it44		else:45			logging.info("{0} record does not exist for {1}, creating".format(record, subdomain))46		action_options = self.GD_ACTION_TEMPLATE.copy()47		action_options['action'] = 'create'48		action_options['type'] = record49		action_options['ttl'] = 60050		action_options['domain'] = self.base_domain51		action_options['name'] = subdomain52		action_options['content'] = content53		logging.info("Creating {0} record with {1} content".format(record, content))54		response = self.send_request(action_options)55		if response is not True:56			return False57		else:58			return True59	def do_delete(self, record, subdomain):60		# A record is restricted, DO NOT REMOVE61		if utilities.is_restricted(self.full_domain) is True and record == 'A':62			return None63		# Check if record exists64		exists = self.check_exists(record, subdomain)65		if exists is False:66			return None67		# A record exists, remove it68		logging.info("{0} record exists for {1}.{2}, removing".format(record, subdomain, self.base_domain))69		action_options = self.GD_ACTION_TEMPLATE.copy()70		action_options['action'] = 'delete'71		action_options['type'] = record72		action_options['domain'] = self.base_domain73		action_options['name'] = subdomain74		response = self.send_request(action_options)75		if response is not True:76			return False77		else:...

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.