KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _options method in Selene

Best Python code snippet using selene_python

purge_reddit.py

Source:purge_reddit.py

1from multiprocessing import queues2class PurgeReddit:3    """4    A class that contains functions to purge reddit posts.5    =================== ======================================================6    Attribute           Description7    =================== ======================================================8    ``reddit``          An instance of :class:`praw.Reddit`, pre-initialized9                        with ``client_id``, ``client_secret``, ``user_agent``,10                        ``username``, and ``password`` attributes.11    ``options``         Dictionary of options with str as keys12    =================== ======================================================13    """14    def __init__(self, reddit, options: dict):15        """16        Initializes :class:`PurgeReddit` instance17        :param reddit: initialized :class:`praw.Reddit` instance18        :param options: dictionary with options from config19        """20        reddit.validate_on_submit = True21        self._reddit = reddit22        self._options = options23        if self._options['max_score'] is None:24            self._options['max_score'] = float('-inf')25    def get_comment_total(self):26        """27        Counts how many comments were made by 'username'.28        :return: number of comments29        """30        count = 031        for _ in self._reddit.user.me().comments.new(32                limit=self._options['limitation']):33            count += 134        if count >= 1000:35            print("Reached max number of retrievable comments",36                  "(>1000 submissions found)")37        return count38    def get_submission_total(self):39        """40        Counts how many submissions were made by 'username'.41        :return: number of submissions42        """43        count = 044        for _ in self._reddit.user.me().submissions.new(45                limit=self._options['limitation']):46            count += 147        if count >= 1000:48            print("Reached max number of retrievable submissions",49                  "(>1000 submissions found)")50        return count51    def purge_comments(self, count: int = None, return_queue: queues = None):52        """53        Purge comments made by 'username'. If a comment could not be purged,54        it will be added to the list of skipped comments.55        :param count: Number of comments to purge, if unspecified then all.56        :param return_queue: Multiprocessing Queue to put list of skipped57        comments, if not specified then unused.58        :return: List of skipped comments.59        """60        # redditor = self._reddit.redditor(self._username)61        # limitation = self._limitation62        if count is None:63            count = self.get_comment_total()64        print(f"Purging {count} comments...")65        skipped = list()66        comments = self._reddit.user.me().comments67        tries = 068        while count > 0 and len(skipped) < 1000:69            if self._options['controversial_first']:70                posts = comments.controversial(limit=self._options['limitation'])71            else:72                posts = comments.new(limit=self._options['limitation'])73            for comment in posts:74                try:75                    msg = f"Comment {comment.submission}/{comment} in {comment.subreddit}"76                    if self._options['show_comment']:77                        msg += f": \"{comment.body[0:80]}...\""78                    if f'{comment}' in self._options['comment_whitelist']:79                        # skip whitelisted80                        msg += " whitelisted"81                        skipped.append(comment)82                    elif comment.score > self._options['max_score']:83                        msg += " skipped (above max score)"84                        skipped.append(comment)85                    else:86                        msg += " redacted"87                        comment.edit(self._options['redact_msg'])88                        if not self._options['redact_only']:89                            comment.delete()90                            msg += " and deleted"91                    count -= 192                    msg += f". {count} to go."93                    print(msg)94                except Exception as exc:95                    print(f"Failed to purge comment {comment}:", exc)96                    count -= 197                    skipped.append(comment)98                if count <= 0:99                    break100            # retry skipped comments up to 3 times101            if len(skipped) > 0 and tries <= 3 and not self._options['redact_only']:102                print(f"Retrying {len(skipped)} skipped comments.")103                count += len(skipped)104                tries += 1105                skipped = list()106        print(f"Comments purge finished! {len(skipped)} could not be purged.", end='')107        if return_queue is not None:108            return_queue.put(skipped)109        print("OK")110        return skipped111    def purge_submissions(self, count: int = None,112                          return_queue: queues = None):113        """114        Purge submissions made by 'username'. If a submission could not be115        purged, it will be added to the list of skipped submissions.116        If running on redact-only mode, image/video/link submissions117        will be skipped.118        :param count: Number of submissions to purge, if unspecified then all.119        :param return_queue: Multiprocessing Queue to put list of skipped120        submissions, if unspecified then unused.121        :return: List of skipped submissions.122        """123        if count is None:124            count = self.get_submission_total125        print(f"Purging {count} submissions...")126        skipped = list()127        submissions = self._reddit.user.me().submissions128        tries = 0129        while count > 0 and len(skipped) < 1000:130            if self._options['controversial_first']:131                posts = submissions.new(limit=self._options['limitation'])132            else:133                posts = submissions.controversial(limit=self._options['limitation'])134            for submission in posts:135                try:136                    msg = f"Submission {submission} in {submission.subreddit}"137                    if self._options['show_title']:138                        msg += f":\"{submission.title[0:140]}...\""139                    if f'{submission}' in self._options['submissions_whitelist']:140                        # skip whitelisted141                        msg += " whitelisted"142                        skipped.append(submission)143                    elif submission.score > self._options['max_score']:144                        # max score threshold145                        msg += " skipped (above max score)"146                        skipped.append(submission)147                    else:148                        # selftext == '' if post is media/link149                        if submission.selftext != '':150                            submission.edit(self._options['redact_msg'])151                            msg += " redacted"152                            if not self._options['redact_only']:153                                msg += " and"154                        elif self._options['redact_only']:155                            # not redacted/deleted156                            skipped.append(submission)157                            msg += " skipped (Media/Link)"158                        if not self._options['redact_only']:159                            submission.delete()160                            msg += " deleted"161                    count -= 1162                    msg += f". {count} to go."163                    print(msg)164                except Exception as exc:165                    print(f"Failed to purge submission {submission}:", exc)166                    count -= 1167                    skipped.append(submission)168                if count <= 0:169                    break170            # retry skipped submissions up to 3 times171            if len(skipped) > 0 and tries <= 3 and not self._options['redact_only']:172                print(f"Retrying {len(skipped)} skipped submissions.")173                count += len(skipped)174                tries += 1175                skipped = list()176        print("Submissions purge finished! Compiling results...", end='')177        if return_queue is not None:178            return_queue.put(skipped)179        print("OK")...

ConfigWrapper.py

Source:ConfigWrapper.py

1## system imports2import os3import sys4## timer5from Timer import Timer6## configuration7import TopAnalysis.Configuration.processes as input8##-----------------------------------------------------------------------------------9## wrapper class for a cfg to python transition10## twiki: https://twiki.cern.ch/twiki/bin/view/CMS/ConfigRunner#ConfigWrapper11class ConfigWrapper:12    ##13    ## configurables14    ##15    __config       = ''   ## input  config16    __output       = ''   ## output config17    __process      = ''   ## physics process18    __pathcounter  = 1000 ## counts the number of paths to be changed/added in19                          ## the predefined config file; 1000 leaves room for 20                          ## user defined process paths in the predefined cfg21                          ## file22##-----------------------------------------------------------------------------------23##  Constructor; options are source, output, event, paths24##  these can be changed within a predefined config file25    def __init__(self, cfg, outputFile, proc):26        ## read input config from source file 27        self.__config = self.readFromFile(cfg)28        ## create output config filename            29        self.__output = outputFile30        ## physics process31        self.__process= proc32        ## define options33        self._options = {}34        self._options['source'] = ""  ## change source35        self._options['output'] = ""  ## change output36        self._options['subset'] = ""  ## change subset output        37        self._options['events'] = ""  ## change number of events38        self._options['skips' ] = ""  ## change number of skipped events39        self._options['paths' ] = ""  ## add paths if specified40        self._options['lumi'  ] = ""  ## configure weights for given lumi41##-----------------------------------------------------------------------------------        42##  * modifies an option43    def modifyOption(self, optionName, value):44        if optionName in self._options.keys():45            self._options[optionName] = value46        else:47            print 'config option' , optionName, 'not found'    48##-----------------------------------------------------------------------------------        49##  * reads a file into a string50    def readFromFile(self, filename):51        file = open(filename, 'r')52        str = file.read()53        file.close()54        return str55##-----------------------------------------------------------------------------------            56##  * writes a string to a file (overwriting existing files)57    def writeToFile(self, filename, str):58        file = open(filename, 'w')59        file.write(str)60        file.close()61##-----------------------------------------------------------------------------------62##  * adds a path to the cfg file; must be given as: module1 * module2 + ...aso63    def addPath(self, path):64        self._options['paths'] += "process.p" + self.__pathcounter.__str__()65        self._options['paths'] += " = cms.Path(" + path + ")\n"66        self.__pathcounter += 167##-----------------------------------------------------------------------------------68##  * replace 'search' by 'replace' in the input config file69    def _replaceInFile(self, search, replace):70        self.__config = self.__config.replace(search.__str__(), replace.__str__())71        72##-----------------------------------------------------------------------------------73##  * replace all entries which were mapped to the option keys in inpit config file74    def _replaceAll(self):75        self.__config += '\n'76        self.__config += '####################################################### \n'77        self.__config += '## \n'78        self.__config += '## automatic parameter replacement \n'79        self.__config += '## \n'80        self.__config += '####################################################### \n'81        ## the loop over the option keys is repeated to get the replacements in fixed order82        for a in self._options.keys():83            if (a=='source'):84            ## modify input files                85                self.__config += 'process.source.fileNames  = ['86                self.__config += self._options[a] + ']'87                self.__config += '\n'88        for a in self._options.keys():                89            if (a=='events'):90            ## modify event numbers91                self.__config += 'process.maxEvents.input   = '92                self.__config += self._options[a].__str__()93                self.__config += '\n'94        for a in self._options.keys():                                95            if (a=='skips' ):96            ## modify skip numbers97                self.__config += 'process.source.skipEvents = cms.untracked.uint32('98                self.__config += self._options[a].__str__() + ')'99                self.__config += '\n'100        for a in self._options.keys():                                101            if (a=='output'):102            ## modify output file103                if(not self._options[a].lower() == 'none'):104                    self.__config += 'process.TFileService.fileName = \''105                    self.__config += self._options[a] + '\''106                    self.__config += '\n'107        for a in self._options.keys():                                    108            if (a=='subset'):109            ## modify subset file110                if(not self._options[a].lower() == ''):111                    self.__config += 'process.out.fileName = \''112                    self.__config += self._options[a] + '\''113                    self.__config += '\n'114        for a in self._options.keys():                                    115            if (a=='paths' ):116            ## modify process paths117                self.__config += self._options[a]118                self.__config += '\n'119        for a in self._options.keys():                                120            if (a=='lumi'  ):121            ## check according to which module the122            ## event weights should be re-configured123                if(not self._options[a] == ''):124                    self._configureEventWeightPlain(self._options[a])125##-----------------------------------------------------------------------------------126##  * returns the name of the temporary config file127    def _configureEventWeightPlain(self, lumi):128        if(self.__config.find("eventWeight")):129            self.__config += '\n'130            self.__config += '####################################################### \n'131            self.__config += '## \n'132            self.__config += '## configuration of evt wght (module: EventWeightPlain) \n'133            self.__config += '## \n'134            self.__config += '####################################################### \n'135            ## number of events136            self.__config += 'process.eventWeight.nevts = '137            self.__config += input.evts[self.__process].__str__()138            self.__config += '\n'139            ## recommended production cross section 140            self.__config += 'process.eventWeight.xsec  = '141            self.__config += input.xsec[self.__process].__str__()142            self.__config += '\n'143            ## proposed target lumi144            self.__config += 'process.eventWeight.lumi  = '145            self.__config += lumi146            self.__config += '\n'147            ## potential generator efficiency148            self.__config += 'process.eventWeight.eff   = '149            self.__config += input.eff [self.__process].__str__()150            self.__config += '\n'                        151##-----------------------------------------------------------------------------------152##  * returns the name of the temporary config file153    def returnTempCfg(self, path='.'):154        ## replace all specified parameters in155        ## the input config156        self._replaceAll()157        ## create job directory if it doesn't158        ## exist yet159        if (not os.path.exists(path)):160            os.system('mkdir ' + path)161        ## define output path162        outpath = path + '/' + self.__output163        self.writeToFile(outpath, self.__config)164        return outpath165##-----------------------------------------------------------------------------------    166    def endJob(self, proc):167        print 'nothing defined for: ', proc168        ...

nagios_nrdp_return.py

Source:nagios_nrdp_return.py

...41    """42    Return virtualname43    """44    return "nagios.list_plugins" in __salt__45def _get_options(ret=None):46    """47    Get the requests options from salt.48    """49    attrs = {50        "url": "url",51        "token": "token",52        "service": "service",53        "checktype": "checktype",54    }55    _options = salt.returners.get_returner_options(56        __virtualname__, ret, attrs, __salt__=__salt__, __opts__=__opts__57    )58    log.debug("attrs %s", attrs)59    if "checktype" not in _options or _options["checktype"] == "":60        # default to passive check type61        _options["checktype"] = "1"62    if _options["checktype"] == "active":63        _options["checktype"] = "0"64    if _options["checktype"] == "passive":65        _options["checktype"] = "1"66    # checktype should be a string67    _options["checktype"] = str(_options["checktype"])68    return _options69def _prepare_xml(options=None, state=None):70    """71    Get the requests options from salt.72    """73    if state:74        _state = "0"75    else:76        _state = "2"77    xml = "<?xml version='1.0'?>\n<checkresults>\n"78    # No service defined then we set the status of the hostname79    if options.get("service"):80        xml += (81            "<checkresult type='service' checktype='" + str(options["checktype"]) + "'>"82        )83        xml += "<hostname>" + html.escape(options["hostname"]) + "</hostname>"84        xml += "<servicename>" + html.escape(options["service"]) + "</servicename>"85    else:86        xml += "<checkresult type='host' checktype='" + str(options["checktype"]) + "'>"87        xml += "<hostname>" + html.escape(options["hostname"]) + "</hostname>"88    xml += "<state>" + _state + "</state>"89    if options.get("output"):90        xml += "<output>" + html.escape(options["output"]) + "</output>"91    xml += "</checkresult>"92    xml += "\n</checkresults>"93    return xml94def _getText(nodelist):95    """96    Simple function to return value from XML97    """98    rc = []99    for node in nodelist:100        if node.nodeType == node.TEXT_NODE:101            rc.append(node.data)102    return "".join(rc)103def _post_data(options=None, xml=None):104    """105    Post data to Nagios NRDP106    """107    params = {"token": options["token"].strip(), "cmd": "submitcheck", "XMLDATA": xml}108    res = salt.utils.http.query(109        url=options["url"],110        method="POST",111        params=params,112        data="",113        decode=True,114        status=True,115        header_dict={},116        opts=__opts__,117    )118    if res.get("status", None) == http.client.OK:119        if res.get("dict", None) and isinstance(res["dict"], list):120            _content = res["dict"][0]121            if _content.get("status", None):122                return True123            else:124                return False125        else:126            log.error("No content returned from Nagios NRDP.")127            return False128    else:129        log.error("Error returned from Nagios NRDP. Status code: %s.", res.status_code)130        return False131def returner(ret):132    """133    Send a message to Nagios with the data134    """135    _options = _get_options(ret)136    log.debug("_options %s", _options)137    _options["hostname"] = ret.get("id")138    if "url" not in _options or _options["url"] == "":139        log.error("nagios_nrdp.url not defined in salt config")140        return141    if "token" not in _options or _options["token"] == "":142        log.error("nagios_nrdp.token not defined in salt config")143        return144    xml = _prepare_xml(options=_options, state=ret["return"])145    res = _post_data(options=_options, xml=xml)...

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.