Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use check_coros method in pandera

Best Python code snippet using pandera_python

test_decorators.py

Source:test_decorators.py

...699        async def static_coroutine(700            df1: DataFrame[Schema],701        ) -> DataFrame[Schema]:702            return df1703    async def check_coros() -> None:704        good_df: DataFrame[Schema] = DataFrame({Schema.col1: [1]})705        bad_df: DataFrame[Schema] = DataFrame({"bad_schema": [1]})706        instance = SomeClass()707        for coro in [708            coroutine,709            instance.regular_coroutine,710            SomeClass.class_coroutine,711            instance.static_coroutine,712            SomeClass.static_coroutine,713            SomeClass.class_meta_coroutine,714            SomeClass.static_meta_coroutine,715            SomeClass.regular_meta_coroutine,716        ]:717            res = await coro(good_df)718            pd.testing.assert_frame_equal(good_df, res)719            with pytest.raises(errors.SchemaError):720                await coro(bad_df)...

check.py

Source:check.py

1import os2import ssl3import enum4import socket5import asyncio6import certifi7from contextlib import redirect_stderr8import attr9from .url import parse_url10async def check_hostname(hostname):11    ssl_context = ssl.create_default_context(cafile=certifi.where())12    try:13        # server_hostname is not needed, because by default,14        # hostname is used for the server cert verification15        _, writer = await asyncio.open_connection(hostname, 443, ssl=ssl_context)16        writer.close()17    except socket.timeout as e:18        return 'Timed out'19    except OSError as e:20        return parse_socket_error_message(e.args[1])21    except ssl.CertificateError as e:22        return str(e)23    else:24        return None25def parse_socket_error_message(message):26    """Cut off brackets from the message, make it human readable."""27    # example: "[SSL: SSLV3_ALERT_HANDSHAKE_FAILURE] sslv3 alert handshake failure (_ssl.c:749)"28    start_ind = message.find(']')29    if start_ind != -1:30        start = start_ind + 231        end = message.find('(_ssl') - 132        message = message[start:end]33    return message34class CheckSiteManager:35    def __init__(self, redirect, timeout, retries, max_threads):36        self.redirect = redirect37        self.timeout = timeout38        self.retries = retries39        self.skipped = []40        self.succeeded = []41        self.failed = []42        self._devnull = open(os.devnull, "w")43    @property44    def success_count(self):45        return len(self.succeeded)46    @property47    def skip_count(self):48        return len(self.skipped)49    @property50    def fail_count(self):51        return len(self.failed)52    async def check_sites(self, urls):53        self.skipped, skipped_urls = self._skip_urls(urls)54        # we deduplicate hostnames, because they are fed in the form of URLs55        hostnames = {parse_url(url).host for url in urls if url not in skipped_urls}56        # we enforce task so they will be started right away, so we can yield from skipped57        check_coros = [self._check_hostname(hostname) for hostname in hostnames]58        # we start yielding after starting requests, so the perceived speed might be better59        # if the client does something with the return values60        for skipped in self.skipped:61            yield skipped62        for future in asyncio.as_completed(check_coros):63            result = await future64            if result.succeeded:65                self.succeeded.append(result)66            elif result.failed:67                self.failed.append(result)68            yield result69    def _skip_urls(self, urls):70        skipped_sites, skipped_urls = [], set()71        for url in urls:72            parsed = parse_url(url)73            if parsed.host is None or not parsed.host.strip():74                checked = CheckedSite(url, CheckResult.SKIPPED, 'invalid_hostname')75            # any other protocoll will be None and as we cannot make a difference,76            # we will check those. Maybe we shouldn't?77            elif parsed.scheme == 'http':78                checked = CheckedSite(url, CheckResult.SKIPPED, 'not https://')79            else:80                continue81            skipped_sites.append(checked)82            skipped_urls.add(url)83        return skipped_sites, skipped_urls84    async def _check_hostname(self, hostname):85        # OpenSSL is more strict about misconfigured servers, e.g. it recognizes missing chains86        with redirect_stderr(self._devnull):87            openssl_error = await check_hostname(hostname)88        result = CheckResult.FAILED if openssl_error else CheckResult.SUCCEEDED89        return CheckedSite(hostname, result, openssl_error)90class CheckResult(enum.Enum):91    SUCCEEDED = 'SUCCEEDED'92    SKIPPED = 'SKIPPED'93    FAILED = 'FAILED'94@attr.s(slots=True, cmp=False)95class CheckedSite:96    url = attr.ib()97    result = attr.ib(convert=CheckResult)98    message = attr.ib(default=None)99    succeeded = attr.ib(init=False)100    skipped = attr.ib(init=False)101    failed = attr.ib(init=False)102    def __attrs_post_init__(self):103        self.succeeded = (self.result == CheckResult.SUCCEEDED)104        self.skipped = (self.result == CheckResult.SKIPPED)...

metadata_utils.py

Source:metadata_utils.py

1import asyncio2import os3from pathlib import Path4from typing import Union5import faraday_agent_dispatcher.logger as logging6from faraday_agent_parameters_types.utils import get_manifests7from faraday_agent_dispatcher import __version__ as current_version8logger = logging.get_logger()9MANDATORY_METADATA_KEYS = [10    "cmd",11    "check_cmds",12    "arguments",13    "environment_variables",14]15INFO_METADATA_KEYS = [16    "category",17    "name",18    "title",19    "website",20    "description",21    "image",22]23# Path can be treated as str24def executor_folder() -> Union[Path, str]:25    folder = Path(__file__).parent.parent / "static" / "executors"26    if "WIZARD_DEV" in os.environ:27        return folder / "dev"28    else:29        return folder / "official"30def executor_metadata(executor_name: str) -> dict:31    return get_manifests(current_version).get(executor_name)32def check_metadata(metadata) -> bool:33    return all(k in metadata for k in MANDATORY_METADATA_KEYS)34def full_check_metadata(metadata) -> bool:35    return all(k in metadata for k in INFO_METADATA_KEYS) and check_metadata(metadata)36async def check_commands(metadata: dict) -> bool:37    async def run_check_command(cmd: str) -> int:38        proc = await asyncio.create_subprocess_shell(39            cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE40        )41        while True:42            stdout, stderr = await proc.communicate()43            if len(stdout) > 0:44                logger.debug(f"Dependency check {cmd} prints: {stdout.decode()}")45            if len(stderr) > 0:46                logger.error(f"Dependency check {cmd} prints to " f"error: {stderr.decode()}")47            if len(stdout) == 0 and len(stderr) == 0:48                break49        return proc.returncode50    for check_cmd in metadata["check_cmds"]:51        response = await run_check_command(check_cmd)52        if response != 0:53            return False54    logger.info("Dependency check ended. Ready to go")55    return True56    # Async check if needed57    # check_coros = [run_check_command(cmd) for cmd in metadata["check_cmds"]]58    # responses = await asyncio.gather(*check_coros)...

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.