Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _collect_data method in yandex-tank

Best Python code snippet using yandex-tank

test_commands.py

Source:test_commands.py

...26            command._write_message(success_message, MessageType.SUCCESS)27            command._write_message(error_message, MessageType.ERROR)28        self.assertIn(success_message, output.getvalue())29        self.assertIn(error_message, output.getvalue())30    def test_successful_collect_data(self):31        """32        Should be able to read the input file data into list33        """34        command = Command()35        data = command._collect_data(DATA_FILE)36        self.assertIsInstance(data, list)37        with open(DATA_FILE, "r") as json_file:38            self.assertEqual(data, json.load(json_file))39    def test_failing_collect_data(self):40        """41        Should properly write to stdout when failing to collect data from file42        """43        output = StringIO()44        command = Command()45        error_message = "Simulated error message"46        expected_message = (47            f"Error trying to read {DATA_FILE}. Got {error_message}"48        )49        with patch.object(command, "stdout", new=output):50            with patch("builtins.open", side_effect=IOError(error_message)):51                command._collect_data(DATA_FILE)52        self.assertIn(expected_message, output.getvalue())53    def test_successful_perform_insertion(self):54        """Should properly insert provided companies data into the database"""55        output = StringIO()56        command = Command()57        data = command._collect_data(DATA_FILE)58        self.assertEqual(Company.objects.count(), 0)59        with patch.object(command, "stdout", new=output):60            command._perform_insertion(data, DATA_FILE)61        self.assertEqual(Company.objects.count(), len(data))62        self.assertIn(63            f"Successfully imported companies data from {DATA_FILE}",64            output.getvalue(),65        )66        companies = Company.objects.all()67        cnpjs = [piece["cnpj"] for piece in data]68        for company in companies:69            self.assertIn(company.cnpj, cnpjs)70    def test_perform_insertion_failure(self):71        """72        Should properly write to stdout when failing to import companies data73        """74        output = StringIO()75        command = Command()76        error_message = "Simulated error message"77        data = command._collect_data(DATA_FILE)78        self.assertEqual(Company.objects.count(), 0)79        with patch.object(command, "stdout", new=output):80            with patch(81                "companies.management.commands.import_companies."82                "import_companies",83                side_effect=IOError(error_message),84            ):85                command._perform_insertion(data, DATA_FILE)86        self.assertIn(87            f"Error trying to import companies data from {DATA_FILE}",88            output.getvalue(),89        )90        self.assertIn(error_message, output.getvalue())91        self.assertEqual(Company.objects.count(), 0)...

xmlchat.py

Source:xmlchat.py

1"""xmlchat.py is a Python module for reading CHILDES files2in XML format written by Mark Johnson, 15th March 2005.3You should use this if you need the utterance word targets.4It may also be handy for obtaining morphological forms.5CHILDES data files in XML format are available from:6http://childes.psy.cmu.edu/data-xml/7You will need to download and unzip these before this program8can read them.9The actual XML format is described in an XML Schema document:10http://xml.talkbank.org/talkbank.xsd11The Schema is described in:12http://xml.talkbank.org:8888/talkbank/talkbank.html13This document makes extensive reference to the original CLAN14documentation, available from:15http://childes.psy.cmu.edu/manuals/CHAT.pdf16"""17import re, xml.parsers.expat18# fields to save from a mor entry19_morph_features = ["c","s","stem","mk","mpfx","menx"]20_filename = None21_participants = None        # dictionary of participant information22_utterances = None          # sequence of utterances in file23_utterance = None           # utterance dictionary of current utterance24_words = None               # words of current utterance25_word = None                # current word26_morphs = None              # morphemes of current utterance27_morph = None               # current morpheme28_collect_data = False       # when True, collect char data29_data = None                # variable to collect char data in30_age_re = re.compile(r"P(?P<Year>\d+)Y(?P<Month>\d+)M((?P<Day>\d+)D)?")31# These are my handler functions, which will be called by the32# Expat XML parser.33def _start_element(name, attrs):34    global _collect_data, _data35    global _context, _morph, _morph_features, _morphs36    global _participants, _utterance, _utterances37    global _word, _words38    global _a_type39    if name == "w" or (name in _morph_features and _morph != None):40        _collect_data = True41        _data = ""42    elif name == "u":43        _words = []44        attrs["words"] = _words45        _utterance = attrs46        _utterances.append(attrs)47    elif name == "t":48        _utterance["t"] = attrs["type"]49    elif name == "a":50        _a_type = attrs["type"]51        assert(_collect_data==False)52        _collect_data = True53        _data = ""54    elif name == "participant":55        _participants[attrs["id"]] = attrs56        agestr = attrs.get("age", None)57        if agestr:58            age_match = _age_re.match(agestr)59            if age_match:60                attrs["months"] = ( 12*int(age_match.group('Year')) + 61                                    int(age_match.group('Month')) )62    elif name == "mor":63        _morphs = []64        _utterance["mor"] = _morphs65    elif name == "mw" and _morphs != None:66        _context = "mw"67        _morph = {}68        _morphs.append(_morph)69        70def _char_data(data):71    global _collect_data, _data72    if _collect_data:73        _data += data.strip()74def _end_element(name):75    global _collect_data, _data, _words, _morph76    global _morphs, _morph_features77    if name == "w":78        _words.append(_data)79        _data = None80        _collect_data = False81    elif name == "a":82        _utterance[_a_type] = _data83        _data = None84        _collect_data = False85    elif name == "mw":86        _morph = None87    elif name == "mor":88        _morphs = None89    elif name in _morph_features and _morph != None:90        _morph[name] = _data91        _collect_data = False92        _data = None93def readfile(filename):94    """readfile reads the contents of a CHAT file in XML format.95    It returns a tuple (participants,utterances), where participants96    is a dictionary mapping from identifiers to information about97    these identifiers, and a list of utterances."""98    global _filename, _participants, _utterances, _context99    _participants = {}100    _utterances = []101    _filename = filename102    f = open(filename, "rb")103    parser = xml.parsers.expat.ParserCreate('UTF-8')104    parser.StartElementHandler = _start_element105    parser.EndElementHandler = _end_element106    parser.CharacterDataHandler = _char_data107    parser.buffer_text = True108    # parser.returns_unicode = False109    parser.ParseFile(f)110    _filename = None...

scrobble_player.py

Source:scrobble_player.py

...10    @classmethod11    def run(cls):12        return player.on("metadata", cls._collect_data)13    @classmethod14    def _collect_data(cls, player, data):15        data = dict(data)16        json = {17            "date_listen": int(time.time()),18            "song": {"title": data["xesam:title"], "length": data["mpris:length"]},19            "album": {"title": data["xesam:album"]},20            "artist": {"name": data["xesam:artist"][0]},21        }22        with open("audio.log", "a+") as log:23            log.writelines(str(json) + "\n")24        return cls._post(json)25    @classmethod26    def _post(cls, json):27        return requests.post(API, json=json, auth=(USER, PASSWORD))28Scrobbler.run()...

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.