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

How to use log_level_to_color method in Lemoncheesecake

Best Python code snippet using lemoncheesecake

console.py

Source:console.py

...14        "disabled": "cyan",15        "in_progress": "cyan",16        None: "cyan"  # None means test is in progress17    }.get(status, "yellow")18def log_level_to_color(level):19    return {20        "debug": "cyan",21        "info": "cyan",22        "warn": "yellow",23        "error": "red"24    }.get(level, "cyan")25def outcome_to_color(outcome):26    return "green" if outcome else "red"27def _serialize_metadata(tags, properties, links, disabled):28    parts = []29    if disabled:30        parts.append("DISABLED")31    parts.extend(tags)32    parts.extend("%s:%s" % (key, properties[key]) for key in sorted(properties))33    parts.extend(link_name or link_url for link_url, link_name in links)34    return ", ".join(parts)35def serialize_metadata(obj, hide_disabled=False):36    return _serialize_metadata(37        obj.tags, obj.properties, obj.links,38        disabled=not hide_disabled and getattr(obj, "disabled", False)39    )40def serialize_hierarchy_metadata(obj, hide_disabled=False):41    return _serialize_metadata(42        obj.hierarchy_tags, obj.hierarchy_properties, obj.hierarchy_links,43        disabled=not hide_disabled and (hasattr(obj, "is_disabled") and obj.is_disabled())44    )45class Renderer(object):46    def __init__(self, max_width, explicit=False, highlight=None, show_debug_logs=False):47        self.max_width = max_width48        self.explicit = explicit49        self.highlight = highlight50        self.show_debug_logs = show_debug_logs51        # "20" is an approximation of the maximal overhead of table border, padding, and table first cell52        self._table_overhead = 2053    def wrap_description_col(self, description):54        return wrap_text(description, int((self.max_width - self._table_overhead) * 0.75))55    def wrap_details_col(self, details):56        return wrap_text(details, int((self.max_width - self._table_overhead) * 0.25))57    def render_check_outcome(self, is_successful):58        if self.explicit:59            check_label = "CHECK %s" % ("OK" if is_successful else "KO")60        else:61            check_label = "CHECK"62        return colored(check_label, color=outcome_to_color(is_successful), attrs=["bold"])63    def render_highlighted(self, content):64        if not self.highlight or not content:65            return content66        return self.highlight.sub(67            lambda m: colored(m.group(0), color="yellow", attrs=["bold", "underline"]), content68        )69    def render_step_log(self, log):70        if isinstance(log, Log):71            if log.level == "debug" and not self.show_debug_logs:72                return None73            else:74                return [75                    colored(log.level.upper(), color=log_level_to_color(log.level), attrs=["bold"]),76                    self.render_highlighted(self.wrap_description_col(log.message))77                ]78        if isinstance(log, Check):79            return [80                self.render_check_outcome(log.is_successful),81                self.render_highlighted(self.wrap_description_col(log.description)),82                self.render_highlighted(self.wrap_details_col(log.details))83            ]84        if isinstance(log, Url):85            if log.description == log.url:86                description = log.url87            else:88                description = "%s (%s)" % (log.url, log.description)89            return [...

test_determinism.py

Source:test_determinism.py

1#!/usr/bin/env python32# Copyright (c) 2020 Graphcore Ltd. All rights reserved.3"""Verify that compilation is deterministic.4This tool runs a command multiple times and checks that the archive generated by5the engine is always the same. If any difference is found then the test is failed.6"""7import argparse8import asyncio9import os10import logging11import json12import subprocess13import sys14import tempfile15LOG = logging.getLogger()16"""The logger instance to use throughout this module."""17class ColouredLoggingFormatter(logging.Formatter):18    """A `logging.Formatter` subclass that colours the level name of the record."""19    RESET = "\033[0m"20    LOG_LEVEL_TO_COLOR = {21        "DEBUG": "\033[1;34m",22        "INFO": "\033[1;37m",23        "WARNING": "\033[1;33m",24        "ERROR": "\033[1;31m",25        "CRITICAL": "\033[1;35m",26    }27    def __init__(self, *args, **kwargs):28        super().__init__(*args, **kwargs)29    def format(self, record) -> str:30        if sys.__stdout__.isatty():31            levelname = record.levelname32            colour = self.LOG_LEVEL_TO_COLOR.get(levelname, None)33            if colour is not None:34                record.levelname = colour + levelname + self.RESET35        return super().format(record)36class SubprocessError(subprocess.CalledProcessError):37    """An extension of `subprocess.CalledProcessError` that prints stdout and stderr."""38    def __init__(self, cmd, returncode, stdout, stderr, kwargs):39        super().__init__(returncode, cmd, stdout, stderr)40        self.cmd = " ".join(cmd)41        self.kwargs = kwargs42    def __str__(self):43        string = super().__str__()44        if self.stdout:45            string += f"\nstdout: {self.stdout}"46        if self.stderr:47            string += f"\nstderr: {self.stderr}"48        if self.kwargs and LOG.getEffectiveLevel() <= logging.DEBUG:49            string += f"\nkwargs: {self.kwargs}"50        return string51    @classmethod52    def from_result(cls, result, kwargs):53        """Construct a SubprocessError instance from a subprocess.CompletedProcess object."""54        assert isinstance(result, subprocess.CompletedProcess)55        return cls(result.args, result.returncode, result.stdout, result.stderr, kwargs)56async def run_async_subprocess(cmd, *args, **kwargs) -> subprocess.CompletedProcess:57    """Run a subprocess asynchronously."""58    # Support some options from `subprocess.run`.59    check = kwargs.pop("check", False)60    input_ = kwargs.pop("input", None)61    decode = kwargs.pop("universal_newlines", True)62    # Use sensible defaults for this script.63    kwargs.setdefault("stdout", asyncio.subprocess.PIPE)64    kwargs.setdefault("stderr", asyncio.subprocess.PIPE)65    # Spawn the subprocess and wait for it to complete asynchronously.66    proc = await asyncio.create_subprocess_exec(*cmd, *args, **kwargs)67    stdout, stderr = await proc.communicate(input_)68    if decode:69        stdout = stdout.strip().decode("utf-8")70        stderr = stderr.strip().decode("utf-8")71    # Make the return value look like that of `subprocess.run`.72    result = subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr)73    if check and result.returncode != 0:74        raise SubprocessError.from_result(result, kwargs)75    LOG.debug(f"Ran: '{' '.join(cmd)}' -> '{result.stdout}'")76    return result77def cli():78    """Define the command line interface of the script."""79    parser = argparse.ArgumentParser(80        description=__doc__,81        formatter_class=argparse.RawTextHelpFormatter82    )83    parser.add_argument("--output", help="the directory to store the archives")84    parser.add_argument("--repeats", type=int, default=2, help="the number of compilations to run")85    parser.add_argument("--overwrite", action="store_true", help="ignore existing archives")86    parser.add_argument("--log-level", choices=("debug", "info", "warning", "error", "critical"),87                        default="info", help="the severity of log messages to print")88    return parser89def main(*args, **kwargs):90    """Entrypoint of the script."""91    parser = cli()92    opts, command = parser.parse_known_args(*args, **kwargs)93    # Prepare the logger.94    global LOG95    LOG.setLevel(getattr(logging, opts.log_level.upper()))96    handler = logging.StreamHandler()97    handler.setFormatter(ColouredLoggingFormatter("%(levelname)s: %(message)s"))98    LOG.addHandler(handler)99    LOG.debug(f"opts={opts}")100    LOG.debug(f"command={command}")101    ref = None # Keep the temporary directory alive for the duration of the test.102    if opts.output:103        out = opts.output104        os.makedirs(out, exist_ok=True)105    else:106        ref = tempfile.TemporaryDirectory()107        out = ref.name108        logging.debug(f"Using temporary directory: {out}")109    # Generate the names of the archives.110    archive_names = {os.path.join(out, f"archive{i}.a") for i in range(opts.repeats)}111    archives_to_generate = archive_names if opts.overwrite else {112        archive113        for archive in archive_names114        if not os.path.exists(archive)115    }116    # Construct the environment variables needed to generate the archive117    # without clobbering any of the users environment variables.118    poplar_engine_options = os.environ.get("POPLAR_ENGINE_OPTIONS", "{}")119    engine_options_dict = json.loads(poplar_engine_options)120    # Check that the user hasn't already specified the engine options we need.121    if "target.saveArchive" in engine_options_dict:122        raise RuntimeError("POPLAR_ENGINE_OPTIONS already contains 'target.saveArchive'")123    envs = []124    for archive in archives_to_generate:125        engine_options_dict["target.saveArchive"] = archive126        env_copy = dict(os.environ)127        env_copy["POPLAR_ENGINE_OPTIONS"] = json.dumps(engine_options_dict)128        envs.append(env_copy)129    # Run the commands asynchronously.130    coros = [131        run_async_subprocess(command, check=True, env=env)132        for archive, env in zip(archives_to_generate, envs)133    ]134    LOG.info(f"Running {len(coros)} commands")135    event_loop = asyncio.get_event_loop()136    event_loop.run_until_complete(asyncio.gather(*coros))137    # Check that the archives were created138    missing_archives = [139        archive140        for archive in archive_names141        if not os.path.exists(archive)142    ]143    if missing_archives:144        raise RuntimeError(f"{len(missing_archives)} compilations failed")145    LOG.debug(f"Generated archives: {archives_to_generate}")146    # Check that the hash of the archives is the same.147    hash_results = event_loop.run_until_complete(asyncio.gather(*[148        run_async_subprocess(["cksum", archive], check=True)149        for archive in archive_names150    ]))151    unique_hashes = {result.stdout.split(" ", 1)[0] for result in hash_results}152    if len(unique_hashes) != 1:153        LOG.error(f"Non-deterministic compilation: {len(unique_hashes)} different archives")154        sys.exit(1)155    LOG.info("Passed :-)")156if __name__ == "__main__":...

tabs.py

Source:tabs.py

1#    Copyright (c) 2013 Mirantis, Inc.2#3#    Licensed under the Apache License, Version 2.0 (the "License"); you may4#    not use this file except in compliance with the License. You may obtain5#    a copy of the License at6#7#         http://www.apache.org/licenses/LICENSE-2.08#9#    Unless required by applicable law or agreed to in writing, software10#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT11#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the12#    License for the specific language governing permissions and limitations13#    under the License.14import logging15from django.core.urlresolvers import reverse16from django.utils.translation import ugettext_lazy as _17from django.utils.datastructures import SortedDict18from horizon import exceptions19from horizon import tabs20from muranoclient.common import exceptions as exc21from muranodashboard.environments import api22from muranodashboard.environments import consts23from muranodashboard.environments.tables import EnvConfigTable, ServicesTable24LOG = logging.getLogger(__name__)25class OverviewTab(tabs.Tab):26    name = _("Component")27    slug = "_service"28    template_name = 'services/_overview.html'29    def get_context_data(self, request):30        """31        :param request:32        :return:33        """34        service_data = self.tab_group.kwargs['service']35        status_name = ''36        for id, name in consts.STATUS_DISPLAY_CHOICES:37            if id == service_data['?']['status']:38                status_name = name39        detail_info = SortedDict([40            ('Name', getattr(service_data, 'name', '')),41            ('ID', service_data['?']['id']),42            ('Type', service_data['?'][consts.DASHBOARD_ATTRS_KEY]['name']),43            ('Status', status_name), ])44        if hasattr(service_data, 'domain'):45            if not service_data.domain:46                detail_info['Domain'] = 'Not in domain'47            else:48                detail_info['Domain'] = service_data.domain49        if hasattr(service_data, 'repository'):50            detail_info['Application repository'] = service_data.repository51        if hasattr(service_data, 'uri'):52            detail_info['Load Balancer URI'] = service_data.uri53        if hasattr(service_data, 'floatingip'):54            detail_info['Floating IP'] = service_data.floatingip55        return {'service': detail_info}56class ServiceLogsTab(tabs.Tab):57    name = _("Logs")58    slug = "service_logs"59    template_name = 'services/_logs.html'60    preload = False61    def get_context_data(self, request):62        service_id = self.tab_group.kwargs['service_id']63        environment_id = self.tab_group.kwargs['environment_id']64        reports = api.get_status_messages_for_service(request, service_id,65                                                      environment_id)66        return {"reports": reports}67class EnvLogsTab(tabs.Tab):68    name = _("Logs")69    slug = "env_logs"70    template_name = 'deployments/_logs.html'71    preload = False72    def get_context_data(self, request):73        reports = self.tab_group.kwargs['logs']74        lines = []75        for r in reports:76            line = format_log(r.created.replace('T', ' ') + ' - ' + r.text,77                              r.level)78            lines.append(line)79        result = '\n'.join(lines)80        if not result:81            result = '\n'82        return {"reports": result}83class EnvConfigTab(tabs.TableTab):84    name = _("Configuration")85    slug = "env_config"86    table_classes = (EnvConfigTable,)87    template_name = 'horizon/common/_detail_table.html'88    preload = False89    def get_environment_configuration_data(self):90        deployment = self.tab_group.kwargs['deployment']91        return deployment.get('services')92class EnvironmentTopologyTab(tabs.Tab):93    name = _("Topology")94    slug = "topology"95    template_name = "services/_detail_topology.html"96    preload = False97    def get_context_data(self, request):98        context = {}99        environment_id = self.tab_group.kwargs['environment_id']100        context['environment_id'] = environment_id101        d3_data = api.load_environment_data(self.request, environment_id)102        context['d3_data'] = d3_data103        return context104class EnvironmentServicesTab(tabs.TableTab):105    name = _("Components")106    slug = "serviceslist"107    table_classes = (ServicesTable,)108    template_name = "services/_service_list.html"109    preload = False110    def get_services_data(self):111        services = []112        self.environment_id = self.tab_group.kwargs['environment_id']113        ns_url = "horizon:murano:environments:index"114        try:115            services = api.services_list(self.request, self.environment_id)116        except exc.HTTPForbidden:117            msg = _('Unable to retrieve list of services. This environment '118                    'is deploying or already deployed by other user.')119            exceptions.handle(self.request, msg, redirect=reverse(ns_url))120        except (exc.HTTPInternalServerError, exc.HTTPNotFound):121            msg = _("Environment with id %s doesn't exist anymore"122                    % self.environment_id)123            exceptions.handle(self.request, msg, redirect=reverse(ns_url))124        except exc.HTTPUnauthorized:125            exceptions.handle(self.request)126        return services127class EnvironmentDetailsTabs(tabs.TabGroup):128    slug = "environemnt_details"129    tabs = (EnvironmentServicesTab, EnvironmentTopologyTab)130class ServicesTabs(tabs.TabGroup):131    slug = "services_details"132    tabs = (OverviewTab, ServiceLogsTab)133class DeploymentTabs(tabs.TabGroup):134    slug = "deployment_details"135    tabs = (EnvConfigTab, EnvLogsTab,)136def format_log(message, level):137    if level == 'warning' or level == 'error':138        frm = "<b><span style='color:#{0}' title='{1}'>{2}</span></b>"139        return frm.format(consts.LOG_LEVEL_TO_COLOR[level],140                          consts.LOG_LEVEL_TO_TEXT[level],141                          message)142    else:...

logger.py

Source:logger.py

1import logging2import logging.handlers3import os4import sys5import colorama6def to_bool(arg):7    '''8    Return a boolean value based on `arg`.9    '''10    if arg is None or isinstance(arg, bool):11        return arg12    if isinstance(arg, str):13        arg = arg.lower()14    if arg in ('yes', 'on', '1', 'true', 1):15        return True16    return False17def interactive_console():18    '''19    Return whether the current console is "interactive". Meaning: Capable of20    user input and not just something like a cron job.21    '''22    return sys.stderr.isatty() and os.environ.get('TERM') != 'dumb'23def should_do_markup(no_color, configs):24    '''25    Given the value of the command-line no-color argument, and a dict of configuration filename to26    corresponding parsed configuration, determine if we should enable colorama marking up.27    '''28    if no_color:29        return False30    if any(config.get('output', {}).get('color') is False for config in configs.values()):31        return False32    py_colors = os.environ.get('PY_COLORS', None)33    if py_colors is not None:34        return to_bool(py_colors)35    return interactive_console()36class Multi_stream_handler(logging.Handler):37    '''38    A logging handler that dispatches each log record to one of multiple stream handlers depending39    on the record's log level.40    '''41    def __init__(self, log_level_to_stream_handler):42        super(Multi_stream_handler, self).__init__()43        self.log_level_to_handler = log_level_to_stream_handler44        self.handlers = set(self.log_level_to_handler.values())45    def flush(self):  # pragma: no cover46        super(Multi_stream_handler, self).flush()47        for handler in self.handlers:48            handler.flush()49    def emit(self, record):50        '''51        Dispatch the log record to the approriate stream handler for the record's log level.52        '''53        self.log_level_to_handler[record.levelno].emit(record)54    def setFormatter(self, formatter):  # pragma: no cover55        super(Multi_stream_handler, self).setFormatter(formatter)56        for handler in self.handlers:57            handler.setFormatter(formatter)58    def setLevel(self, level):  # pragma: no cover59        super(Multi_stream_handler, self).setLevel(level)60        for handler in self.handlers:61            handler.setLevel(level)62LOG_LEVEL_TO_COLOR = {63    logging.CRITICAL: colorama.Fore.RED,64    logging.ERROR: colorama.Fore.RED,65    logging.WARN: colorama.Fore.YELLOW,66    logging.INFO: colorama.Fore.GREEN,67    logging.DEBUG: colorama.Fore.CYAN,68}69class Console_color_formatter(logging.Formatter):70    def format(self, record):71        color = LOG_LEVEL_TO_COLOR.get(record.levelno)72        return color_text(color, record.msg)73def color_text(color, message):74    '''75    Give colored text.76    '''77    if not color:78        return message79    return '{}{}{}'.format(color, message, colorama.Style.RESET_ALL)80def configure_logging(81    console_log_level,82    syslog_log_level=None,83    log_file_log_level=None,84    monitoring_log_level=None,85    log_file=None,86):87    '''88    Configure logging to go to both the console and (syslog or log file). Use the given log levels,89    respectively.90    Raise FileNotFoundError or PermissionError if the log file could not be opened for writing.91    '''92    if syslog_log_level is None:93        syslog_log_level = console_log_level94    if log_file_log_level is None:95        log_file_log_level = console_log_level96    if monitoring_log_level is None:97        monitoring_log_level = console_log_level98    # Log certain log levels to console stderr and others to stdout. This supports use cases like99    # grepping (non-error) output.100    console_error_handler = logging.StreamHandler(sys.stderr)101    console_standard_handler = logging.StreamHandler(sys.stdout)102    console_handler = Multi_stream_handler(103        {104            logging.CRITICAL: console_error_handler,105            logging.ERROR: console_error_handler,106            logging.WARN: console_standard_handler,107            logging.INFO: console_standard_handler,108            logging.DEBUG: console_standard_handler,109        }110    )111    console_handler.setFormatter(Console_color_formatter())112    console_handler.setLevel(console_log_level)113    syslog_path = None114    if log_file is None:115        if os.path.exists('/dev/log'):116            syslog_path = '/dev/log'117        elif os.path.exists('/var/run/syslog'):118            syslog_path = '/var/run/syslog'119        elif os.path.exists('/var/run/log'):120            syslog_path = '/var/run/log'121    if syslog_path and not interactive_console():122        syslog_handler = logging.handlers.SysLogHandler(address=syslog_path)123        syslog_handler.setFormatter(logging.Formatter('borgmatic: %(levelname)s %(message)s'))124        syslog_handler.setLevel(syslog_log_level)125        handlers = (console_handler, syslog_handler)126    elif log_file:127        file_handler = logging.handlers.WatchedFileHandler(log_file)128        file_handler.setFormatter(logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s'))129        file_handler.setLevel(log_file_log_level)130        handlers = (console_handler, file_handler)131    else:132        handlers = (console_handler,)133    logging.basicConfig(134        level=min(console_log_level, syslog_log_level, log_file_log_level, monitoring_log_level),135        handlers=handlers,...

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.