How to use domain_prefix method in localstack

Best Python code snippet using localstack_python

gen_doc.py

Source:gen_doc.py

1#!/usr/bin/env python2from __future__ import absolute_import3from __future__ import division4from __future__ import print_function5from __future__ import unicode_literals6import io7import os8from collections import defaultdict9from onnx import defs10from onnx.defs import OpSchema11from onnx.backend.test.case import collect_snippets12from typing import Text, Sequence, Dict, List, Type, Set13SNIPPETS = collect_snippets()14ONNX_ML = bool(os.getenv('ONNX_ML') == '1')15if ONNX_ML:16    ext = '-ml.md'17else:18    ext = '.md'19def display_number(v):  # type: (int) -> Text20    if defs.OpSchema.is_infinite(v):21        return '&#8734;'22    return Text(v)23def should_render_domain(domain):  # type: (Text) -> bool24    if domain == 'ai.onnx.ml' and not ONNX_ML:25        return False26    elif ONNX_ML and domain != 'ai.onnx.ml':27        return False28    return True29def display_attr_type(v):  # type: (OpSchema.AttrType) -> Text30    assert isinstance(v, OpSchema.AttrType)31    s = Text(v)32    s = s[s.rfind('.') + 1:].lower()33    if s[-1] == 's':34        s = 'list of ' + s35    return s36def display_domain(domain):  # type: (Text) -> Text37    if domain:38        return "the '{}' operator set".format(domain)39    else:40        return "the default ONNX operator set"41def display_version_link(name, version):  # type: (Text, int) -> Text42    changelog_md = 'Changelog' + ext43    name_with_ver = '{}-{}'.format(name, version)44    return '<a href="{}#{}">{}</a>'.format(changelog_md, name_with_ver, name_with_ver)45def display_schema(schema, versions):  # type: (OpSchema, Sequence[OpSchema]) -> Text46    s = ''47    if schema.domain:48        domain_prefix = '{}.'.format(schema.domain)49    else:50        domain_prefix = ''51    # doc52    if schema.doc:53        s += '\n'54        s += '\n'.join('  ' + line55                       for line in schema.doc.lstrip().splitlines())56        s += '\n'57    # since version58    s += '\n#### Version\n'59    s += '\nThis version of the operator has been available since version {}'.format(schema.since_version)60    s += ' of {}.\n'.format(display_domain(schema.domain))61    if len(versions) > 1:62        # TODO: link to the Changelog.md63        s += '\nOther versions of this operator: {}\n'.format(64            ', '.join(display_version_link(domain_prefix + v.name, v.since_version) for v in versions[:-1]))65    # attributes66    if schema.attributes:67        s += '\n#### Attributes\n\n'68        s += '<dl>\n'69        for _, attr in sorted(schema.attributes.items()):70            s += '<dt><tt>{}</tt> : {}{}</dt>\n'.format(71                attr.name,72                display_attr_type(attr.type),73                ' (required)' if attr.required else '')74            s += '<dd>{}</dd>\n'.format(attr.description)75        s += '</dl>\n'76    # inputs77    s += '\n#### Inputs'78    if schema.min_input != schema.max_input:79        s += ' ({} - {})'.format(display_number(schema.min_input),80                                 display_number(schema.max_input))81    s += '\n\n'82    if schema.inputs:83        s += '<dl>\n'84        for input in schema.inputs:85            option_str = ""86            if OpSchema.FormalParameterOption.Optional == input.option:87                option_str = " (optional)"88            elif OpSchema.FormalParameterOption.Variadic == input.option:89                option_str = " (variadic)"90            s += '<dt><tt>{}</tt>{} : {}</dt>\n'.format(input.name, option_str, input.typeStr)91            s += '<dd>{}</dd>\n'.format(input.description)92        s += '</dl>\n'93    # outputs94    s += '\n#### Outputs'95    if schema.min_output != schema.max_output:96        s += ' ({} - {})'.format(display_number(schema.min_output),97                                 display_number(schema.max_output))98    s += '\n\n'99    if schema.outputs:100        s += '<dl>\n'101        for output in schema.outputs:102            option_str = ""103            if OpSchema.FormalParameterOption.Optional == output.option:104                option_str = " (optional)"105            elif OpSchema.FormalParameterOption.Variadic == output.option:106                option_str = " (variadic)"107            s += '<dt><tt>{}</tt>{} : {}</dt>\n'.format(output.name, option_str, output.typeStr)108            s += '<dd>{}</dd>\n'.format(output.description)109        s += '</dl>\n'110    # type constraints111    s += '\n#### Type Constraints'112    s += '\n\n'113    if schema.type_constraints:114        s += '<dl>\n'115        for type_constraint in schema.type_constraints:116            allowedTypes = type_constraint.allowed_type_strs117            if (len(allowedTypes) > 0):118                allowedTypeStr = allowedTypes[0]119            for allowedType in allowedTypes[1:]:120                allowedTypeStr += ', ' + allowedType121            s += '<dt><tt>{}</tt> : {}</dt>\n'.format(122                type_constraint.type_param_str, allowedTypeStr)123            s += '<dd>{}</dd>\n'.format(type_constraint.description)124        s += '</dl>\n'125    return s126def support_level_str(level):  # type: (OpSchema.SupportType) -> Text127    return \128        "<sub>experimental</sub> " if level == OpSchema.SupportType.EXPERIMENTAL else ""129def main(args):  # type: (Type[Args]) -> None130    with io.open(args.changelog, 'w', newline='') as fout:131        fout.write('## Operator Changelog\n')132        fout.write(133            "*This file is automatically generated from the\n"134            "            [def files](/onnx/defs) via [this script](/onnx/defs/gen_doc.py).\n"135            "            Do not modify directly and instead edit operator definitions.*\n")136        # domain -> version -> [schema]137        dv_index = defaultdict(lambda: defaultdict(list))  # type: Dict[Text, Dict[int, List[OpSchema]]]138        for schema in defs.get_all_schemas_with_history():139            dv_index[schema.domain][schema.since_version].append(schema)140        fout.write('\n')141        for domain, versionmap in sorted(dv_index.items()):142            if not should_render_domain(domain):143                continue144            if domain:145                s = '# {}\n'.format(domain)146                domain_prefix = '{}.'.format(domain)147            else:148                s = '# ai.onnx (default)\n'149                domain_prefix = ''150            for version, unsorted_schemas in sorted(versionmap.items()):151                s += '## Version {} of {}\n'.format(version, display_domain(domain))152                for schema in sorted(unsorted_schemas, key=lambda s: s.name):153                    name_with_ver = '{}-{}'.format(domain_prefix +154                                                   schema.name, schema.since_version)155                    s += '### <a name="{}"></a>**{}**</a>\n'.format(name_with_ver, name_with_ver)156                    s += display_schema(schema, [schema])157                    s += '\n'158            fout.write(s)159    with io.open(args.output, 'w', newline='') as fout:160        fout.write('## Operator Schemas\n')161        fout.write(162            "*This file is automatically generated from the\n"163            "            [def files](/onnx/defs) via [this script](/onnx/defs/gen_doc.py).\n"164            "            Do not modify directly and instead edit operator definitions.*\n")165        # domain -> support level -> name -> [schema]166        index = defaultdict(lambda: defaultdict(lambda: defaultdict(list)))  # type: Dict[Text, Dict[int, Dict[Text, List[OpSchema]]]]167        for schema in defs.get_all_schemas_with_history():168            index[schema.domain][int(schema.support_level)][schema.name].append(schema)169        fout.write('\n')170        # Table of contents171        exsting_ops = set()  # type: Set[Text]172        for domain, supportmap in sorted(index.items()):173            if not should_render_domain(domain):174                continue175            if domain:176                s = '* {}\n'.format(domain)177                domain_prefix = '{}.'.format(domain)178            else:179                s = '* ai.onnx (default)\n'180                domain_prefix = ''181            fout.write(s)182            for _, namemap in sorted(supportmap.items()):183                for n, unsorted_versions in sorted(namemap.items()):184                    versions = sorted(unsorted_versions, key=lambda s: s.since_version)185                    schema = versions[-1]186                    if schema.name in exsting_ops:187                        continue188                    exsting_ops.add(schema.name)189                    s = '  * {}<a href="#{}">{}</a>\n'.format(190                        support_level_str(schema.support_level),191                        domain_prefix + n, domain_prefix + n)192                    fout.write(s)193        fout.write('\n')194        for domain, supportmap in sorted(index.items()):195            if not should_render_domain(domain):196                continue197            if domain:198                s = '## {}\n'.format(domain)199                domain_prefix = '{}.'.format(domain)200            else:201                s = '## ai.onnx (default)\n'202                domain_prefix = ''203            fout.write(s)204            for _support, namemap in sorted(supportmap.items()):205                for op_type, unsorted_versions in sorted(namemap.items()):206                    versions = sorted(unsorted_versions, key=lambda s: s.since_version)207                    schema = versions[-1]208                    # op_type209                    s = '### {}<a name="{}"></a><a name="{}">**{}**</a>\n'.format(210                        support_level_str(schema.support_level),211                        domain_prefix + op_type, domain_prefix + op_type.lower(),212                        domain_prefix + op_type)213                    s += display_schema(schema, versions)214                    s += '\n\n'215                    if op_type in SNIPPETS:216                        s += '#### Examples\n\n'217                        for summary, code in sorted(SNIPPETS[op_type]):218                            s += '<details>\n'219                            s += '<summary>{}</summary>\n\n'.format(summary)220                            s += '```python\n{}\n```\n\n'.format(code)221                            s += '</details>\n'222                            s += '\n\n'223                    fout.write(s)224if __name__ == '__main__':225    base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))226    docs_dir = os.path.join(base_dir, 'docs')227    class Args(object):228        output = os.path.join(docs_dir, 'Operators' + ext)229        changelog = os.path.join(docs_dir, 'Changelog' + ext)...

test_domain.py

Source:test_domain.py

1# Licensed under the Apache License, Version 2.0 (the "License");2# you may not use this file except in compliance with the License.3# You may obtain a copy of the License at4#5#    http://www.apache.org/licenses/LICENSE-2.06#7# Unless required by applicable law or agreed to in writing, software8# distributed under the License is distributed on an "AS IS" BASIS,9# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.10# See the License for the specific language governing permissions and11# limitations under the License.12"""13test_domain14----------------------------------15Functional tests for `shade` keystone domain resource.16"""17import SDK.openstack.cloud18from SDK.openstack.tests.functional import base19class TestDomain(base.BaseFunctionalTest):20    def setUp(self):21        super(TestDomain, self).setUp()22        i_ver = self.operator_cloud.config.get_api_version('identity')23        if i_ver in ('2', '2.0'):24            self.skipTest('Identity service does not support domains')25        self.domain_prefix = self.getUniqueString('domain')26        self.addCleanup(self._cleanup_domains)27    def _cleanup_domains(self):28        exception_list = list()29        for domain in self.operator_cloud.list_domains():30            if domain['name'].startswith(self.domain_prefix):31                try:32                    self.operator_cloud.delete_domain(domain['id'])33                except Exception as e:34                    exception_list.append(str(e))35                    continue36        if exception_list:37            # Raise an error: we must make users aware that something went38            # wrong39            raise SDK.openstack.cloud.OpenStackCloudException(40                '\n'.join(exception_list))41    def test_search_domains(self):42        domain_name = self.domain_prefix + '_search'43        # Shouldn't find any domain with this name yet44        results = self.operator_cloud.search_domains(45            filters=dict(name=domain_name))46        self.assertEqual(0, len(results))47        # Now create a new domain48        domain = self.operator_cloud.create_domain(domain_name)49        self.assertEqual(domain_name, domain['name'])50        # Now we should find only the new domain51        results = self.operator_cloud.search_domains(52            filters=dict(name=domain_name))53        self.assertEqual(1, len(results))54        self.assertEqual(domain_name, results[0]['name'])55        # Now we search by name with name_or_id, should find only new domain56        results = self.operator_cloud.search_domains(name_or_id=domain_name)57        self.assertEqual(1, len(results))58        self.assertEqual(domain_name, results[0]['name'])59    def test_update_domain(self):60        domain = self.operator_cloud.create_domain(61            self.domain_prefix, 'description')62        self.assertEqual(self.domain_prefix, domain['name'])63        self.assertEqual('description', domain['description'])64        self.assertTrue(domain['enabled'])65        updated = self.operator_cloud.update_domain(66            domain['id'], name='updated name',67            description='updated description', enabled=False)68        self.assertEqual('updated name', updated['name'])69        self.assertEqual('updated description', updated['description'])70        self.assertFalse(updated['enabled'])71        # Now we update domain by name with name_or_id72        updated = self.operator_cloud.update_domain(73            None,74            name_or_id='updated name',75            name='updated name 2',76            description='updated description 2',77            enabled=True)78        self.assertEqual('updated name 2', updated['name'])79        self.assertEqual('updated description 2', updated['description'])80        self.assertTrue(updated['enabled'])81    def test_delete_domain(self):82        domain = self.operator_cloud.create_domain(self.domain_prefix,83                                                   'description')84        self.assertEqual(self.domain_prefix, domain['name'])85        self.assertEqual('description', domain['description'])86        self.assertTrue(domain['enabled'])87        deleted = self.operator_cloud.delete_domain(domain['id'])88        self.assertTrue(deleted)89        # Now we delete domain by name with name_or_id90        domain = self.operator_cloud.create_domain(91            self.domain_prefix, 'description')92        self.assertEqual(self.domain_prefix, domain['name'])93        self.assertEqual('description', domain['description'])94        self.assertTrue(domain['enabled'])95        deleted = self.operator_cloud.delete_domain(None, domain['name'])96        self.assertTrue(deleted)97        # Finally, we assert we get False from delete_domain if domain does98        # not exist99        domain = self.operator_cloud.create_domain(100            self.domain_prefix, 'description')101        self.assertEqual(self.domain_prefix, domain['name'])102        self.assertEqual('description', domain['description'])103        self.assertTrue(domain['enabled'])104        deleted = self.operator_cloud.delete_domain(None, 'bogus_domain')...

cctv.py

Source:cctv.py

1from news_crawler import NewsCrawler2from bs4 import BeautifulSoup3import datefinder4from urllib.parse import urlparse5import re6def clean_up(text: str) -> str:7    tab = re.compile('#TAB#')8    rchar = re.compile('#R#')9    newline = re.compile('#N#')10    text = tab.sub('\t', text)11    text = rchar.sub('\r', text)12    text = newline.sub('\n', text)13    return text14class CCTVNewsCrawler(NewsCrawler):15    def __init__(self, store_in_memory: bool = True, store_in_file: str = '../../../result/news/cctv_news.json', domain_prefix: str = 'news'):16        """17        news.cctv.com18        m.news.cctv.com19        tv.cctv.com (don't consider)20        """21        super().__init__(store_in_memory, store_in_file)22        self.domain_prefix = domain_prefix23    def _get_raw_html(self, url: str) -> str:24        # return super()._get_raw_html(url, specific_encoding='ISO-8859-1')25        return super()._get_raw_html(url, force_encode=True)26    def _get_title(self, html_body_soup: BeautifulSoup):27        """28        Seems there might have some dirty content here29        """30        if self.domain_prefix == 'news':31            title_area = html_body_soup.find('div', {'class': 'title_area'})32            title = title_area.find('h1').text33        elif self.domain_prefix == 'm.news':34            title = html_body_soup.find('h1').text35        return clean_up(title).strip()36    def _get_author(self, html_body_soup: BeautifulSoup):37        """38        TODO: need some NLP method to refine or extract39        """40        if self.domain_prefix == 'news':41            author_info = html_body_soup.find('div', {'class': 'zebian'})42            author = author_info.text43        elif self.domain_prefix == 'm.news':44            function_info = html_body_soup.find('div', {'class': 'function'}).find(45                'span', {'class', 'info'})46            author = function_info.find('a').text47        return clean_up(author).strip()48    def _get_date(self, html_body_soup: BeautifulSoup):49        if self.domain_prefix == 'news':50            title_area = html_body_soup.find('div', {'class': 'title_area'})51            text_contain_date = title_area.find('div', {'class', 'info1'}).text52        elif self.domain_prefix == 'm.news':53            function_info = html_body_soup.find('div', {'class': 'function'}).find(54                'span', {'class', 'info'})55            text_contain_date = function_info.text56        article_info = html_body_soup.find('div', {'class': 'article-info'})57        return next(datefinder.find_dates(text_contain_date))58    def _get_content(self, html_body_soup: BeautifulSoup):59        if self.domain_prefix == 'news':60            content_area = html_body_soup.find(61                'div', {'class': 'content_area'})62            article = content_area.text63        elif self.domain_prefix == 'm.news':64            content_area = html_body_soup.find('div', {'class': 'cnt_bd'})65            paragraphs = content_area.find_all('p')66            article = '\n\n'.join((paragraph.text for paragraph in paragraphs))67        return clean_up(article)68if __name__ == "__main__":69    crawler = CCTVNewsCrawler()70    crawler.domain_prefix = 'news'71    crawler.crawl_single_url(72        'https://news.cctv.com/2020/08/19/ARTIJapDsvNdGtH0AF4k8yGf200819.shtml')  # success!!!73    crawler.domain_prefix = 'm.news'74    crawler.crawl_single_url(75        'http://m.news.cctv.com/2020/08/19/ARTIzpQnphaF1ZC0DFzMNxvd200819.shtml')  # success!!!76    domains_files = {77        'news': 'https://news.cctv.com/2020/08/19/ARTIJapDsvNdGtH0AF4k8yGf200819.shtml',78        'm.news': 'http://m.news.cctv.com/2020/08/19/ARTIzpQnphaF1ZC0DFzMNxvd200819.shtml'79    }80    # TODO: write Testcase by evaluate parse result with test_html_body/cctv.json81    for domain, url in domains_files.items():82        with open(f'test_html_body/cctv_{domain}.html', 'r') as fp:83            html = fp.read()84        _parsed_uri = urlparse(url)85        domain = '{uri.netloc}'.format(uri=_parsed_uri)86        domain = domain.replace('.cctv.com', '')87        crawler.domain_prefix = domain88        crawler.crawl_html(89            html, url=url)90    print(crawler.data)91    import ipdb...

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.