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

How to use _parse_channel_info method in lisa

Best Python code snippet using lisa_python

test_ytr_service_data_import.py

Source:test_ytr_service_data_import.py

...305        self.assertEqual(service_parsed['areas']['fi'][0]['name'], 'Turku')306        self.assertEqual(len(service_parsed['channelIds']), 2)  307        self.assertEqual(service_parsed['lastUpdated'], datetime(2020, 12, 13, 8, 2, 57, 83000)) 308        309    def test_parse_channel_info(self):310        self.setUp()311        municipalities = self.ytr_importer.get_municipalities()312        self.ytr_importer.municipality_map = self.ytr_importer._parse_municipality_map(municipalities)313        314        # Parse channel315        channel_orig = self.channels_response[0]316        channel_parsed = self.ytr_importer._parse_channel_info(channel_orig)317        self.assertEqual(channel_parsed['organizationId'], str(channel_orig['toimija']))318        self.assertEqual(channel_parsed['name']['fi'], channel_orig['nimi']['fi'])319        self.assertEqual(channel_parsed['descriptions']['fi'][0]['value'], channel_orig['kuvaus']['fi'])320        self.assertEqual(channel_parsed['webPages']['fi'][0], channel_orig['yhteystiedot'][0]['arvo'])321        self.assertEqual(channel_parsed['addresses']['fi'][0]['municipalityName'], 'Turku')322        self.assertEqual(len(channel_parsed['serviceIds']), 0)323        self.assertEqual(channel_parsed['lastUpdated'], datetime(2020, 12, 13, 8, 2, 57, 83000))  324        325    def test_get_new_services_and_channels(self):326        self.setUp()327        municipalities = self.ytr_importer.get_municipalities()328        self.ytr_importer.municipality_map = self.ytr_importer._parse_municipality_map(municipalities)329        # Services330        service_offers = self.ytr_importer.get_service_offers()331        service_offers_parsed = [self.ytr_importer._parse_service_info(ser) for ser in service_offers]332        self.assertEqual(len(service_offers_parsed), 5)333        service_offers_parsed = [ser for ser in service_offers_parsed if self.ytr_importer._is_suitable_service(ser)]334        self.assertEqual(len(service_offers_parsed), 4)335        last_update_time = self.ytr_importer.get_latest_update_time_from_mongo('ytr_services')336        service_offers_new = [so for so in service_offers_parsed if last_update_time is None or so.get('lastUpdated') is None or so.get('lastUpdated') > last_update_time]337        self.assertEqual(len(service_offers_new), 4)338        ytr_original, ptv_recognized_services = self.ytr_importer._filter_and_split_services(service_offers_new, self.ptv_services_response)339        self.assertEqual(len(ytr_original), 2)340        self.assertEqual(len(ptv_recognized_services), 2)341        342        # Channels343        all_channel_ids = []344        all_channels = []345        new_service = ptv_recognized_services[0]346        channels = self.ytr_importer.get_service_channels(new_service.get('channelIds'))347        channels_parsed = [self.ytr_importer._parse_channel_info(cha) for cha in channels]348        channels_parsed_ptv_ids = [cha.get('ptvId') for cha in channels_parsed if cha.get('ptvId') is not None]349            350        new_service_ptv_id = new_service.get('ptvId')351        if new_service_ptv_id is not None:352            current_ptv_channels_by_service = list(self.ytr_importer.mongo_client.service_db.channels.find({'serviceIds': {"$in": [new_service_ptv_id]}}))353            if len(channels_parsed_ptv_ids) > 0:354                channel_referred_ptv_channels = list(self.ytr_importer.mongo_client.service_db.channels.find({'id': {"$in": channels_parsed_ptv_ids}}))355            else:356                channel_referred_ptv_channels = []357            current_ptv_channels = current_ptv_channels_by_service + channel_referred_ptv_channels358        else:359            current_ptv_channels = []360        new_channels, ptv_unrecognized_channels, known_channels = self.ytr_importer._split_channels(channels_parsed, all_channels, current_ptv_channels)361        all_new_service_channels = new_channels + ptv_unrecognized_channels362        # Add reference to the services that have been found earlier363        for known_channel in known_channels:364            known_channels_handle = [cha for cha in all_channels if cha.get('ptvId') == known_channel.get('ptvId') or cha.get('id') == known_channel.get('id')][0]365            known_channels_handle['serviceIds'] = known_channels_handle['serviceIds'] + [new_service.get('id')]366        # Add same reference to the new ones367        for channel in all_new_service_channels:368            channel['serviceIds'] = channel['serviceIds'] + [new_service.get('id')]369        all_channels = all_channels + all_new_service_channels370        all_channel_ids = all_channel_ids + [cha.get('id') for cha in all_new_service_channels if cha.get('id') is not None]371    372        self.assertEqual(len(new_channels), 2)373        self.assertEqual(len(ptv_unrecognized_channels), 0)374        self.assertEqual(len(known_channels), 0)375        376        new_service = ptv_recognized_services[1]377        channels = self.ytr_importer.get_service_channels(new_service.get('channelIds'))378        channels_parsed = [self.ytr_importer._parse_channel_info(cha) for cha in channels]379        channels_parsed_ptv_ids = [cha.get('ptvId') for cha in channels_parsed if cha.get('ptvId') is not None]380        new_service_ptv_id = new_service.get('ptvId')381        if new_service_ptv_id is not None:382            current_ptv_channels_by_service = list(self.ytr_importer.mongo_client.service_db.channels.find({'serviceIds': {"$in": [new_service_ptv_id]}}))383            if len(channels_parsed_ptv_ids) > 0:384                channel_referred_ptv_channels = list(self.ytr_importer.mongo_client.service_db.channels.find({'id': {"$in": channels_parsed_ptv_ids}}))385            else:386                channel_referred_ptv_channels = []387            current_ptv_channels = current_ptv_channels_by_service + channel_referred_ptv_channels388        else:389            current_ptv_channels = []390        new_channels, ptv_unrecognized_channels, known_channels = self.ytr_importer._split_channels(channels_parsed, all_channels, current_ptv_channels)391        all_new_service_channels = new_channels + ptv_unrecognized_channels392        # Add reference to the services that have been found earlier393        for known_channel in known_channels:394            known_channels_handle = [cha for cha in all_channels if cha.get('ptvId') == known_channel.get('ptvId') or cha.get('id') == known_channel.get('id')][0]395            known_channels_handle['serviceIds'] = known_channels_handle['serviceIds'] + [new_service.get('id')]396        # Add same reference to the new ones397        for channel in all_new_service_channels:398            channel['serviceIds'] = channel['serviceIds'] + [new_service.get('id')]399        all_channels = all_channels + all_new_service_channels400        all_channel_ids = all_channel_ids + [cha.get('id') for cha in all_new_service_channels if cha.get('id') is not None]401        402        self.assertEqual(len(new_channels), 0)403        self.assertEqual(len(ptv_unrecognized_channels), 1)404        self.assertEqual(len(known_channels), 1)405        406        new_service = ytr_original[0]407        channels = self.ytr_importer.get_service_channels(new_service.get('channelIds'))408        channels_parsed = [self.ytr_importer._parse_channel_info(cha) for cha in channels]409        channels_parsed_ptv_ids = [cha.get('ptvId') for cha in channels_parsed if cha.get('ptvId') is not None] 410        new_service_ptv_id = new_service.get('ptvId')411        if new_service_ptv_id is not None:412            current_ptv_channels_by_service = list(self.ytr_importer.mongo_client.service_db.channels.find({'serviceIds': {"$in": [new_service_ptv_id]}}))413            if len(channels_parsed_ptv_ids) > 0:414                channel_referred_ptv_channels = list(self.ytr_importer.mongo_client.service_db.channels.find({'id': {"$in": channels_parsed_ptv_ids}}))415            else:416                channel_referred_ptv_channels = []417            current_ptv_channels = current_ptv_channels_by_service + channel_referred_ptv_channels418        else:419            current_ptv_channels = []420        new_channels, ptv_unrecognized_channels, known_channels = self.ytr_importer._split_channels(channels_parsed, all_channels, current_ptv_channels)421        all_new_service_channels = new_channels + ptv_unrecognized_channels422        # Add reference to the services that have been found earlier...

_parser.py

Source:_parser.py

...74        elem = elem.strip()75        if elem == "Time [s]":76            time = index77        elif elem.startswith("Channel"):78            number, type = _parse_channel_info(elem)79            channel = _Channel(number, type, time, index)80            channels.append(channel)81        else:82            raise ValueError(83                'Expected header field of the following form: "Time [s]"\n'84                'or "Channel [0-9]-(Analog|Digital)". Instead received the\n'85                f"following: {header}"86            )87    return channels88def _parse_channel_info(entry: str) -> tuple[int, str]:89    """Parse a header entry of the form ``'Channel90    [0-9]-(Analog|Digital)'``.91    Args:92        entry: The entry to parse93    Returns:94        The number of the channel and its type (``'analog'`` or95        ``'digital'``)96    """97    match = re.match("Channel ([0-9])-(Analog|Digital)", entry)98    number = int(match.group(1))99    type = match.group(2).lower()100    return number, type101@dataclasses.dataclass102class _Channel:...

feed.py

Source:feed.py

...34                raise e35            except Exception, e:36                logger.error(e)37                raise ValueError('Failed to parse data from url, is it a valid RSS feed url?')38    def _parse_channel_info(self):39        if self.channel_info is None:40            self._parse_feed_data()41            updated = self.feed_data.feed.get('updated_parsed')42            # Some feeds do not provide last updated time43            if updated:44                updated = RSSFeed._time_struct_to_datetime(updated)45            self.channel_info = {46                'title': self.feed_data.feed.title,47                'subtitle': self.feed_data.feed.get('subtitle'),48                'url': self.feed_data.feed.get('link'),49                'language': self.feed_data.feed.get('language'),50                'updated': updated,51                'rss_url': self.feed_url52            }53    def _parse_entries_info(self):54        if self.entries is None:55            self._parse_feed_data()56            self.entries = []57            for entry in self.feed_data.entries:58                pub_date = RSSFeed._time_struct_to_datetime(entry.published_parsed)59                title = entry.title60                summary = entry.get('summary')61                url = entry.get('link')62                # Because ATOM feeds may provide more than one content feedparser provides the content as a list63                content = entry.get('content')64                if content and isinstance(content, list):65                    content = content[0]66                if isinstance(content, dict):67                    content = content.get('value')68                authors = list(author['name'] for author in entry.get('authors', []))69                tags = list(tag['term'] for tag in entry.get('tags', []))70                entry_info = {71                    'entry_info': {72                        'title': title,73                        'summary': summary,74                        'url': url,75                        'pub_date': pub_date,76                        'content': content,77                    },78                    'authors': authors,79                    'tags': tags80                }81                self.entries.append(entry_info)82    def get_channel_info(self):83        self._parse_channel_info()84        return self.channel_info85    def get_entries_info(self):86        self._parse_entries_info()...

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.