Best Python code snippet using lisa_python
objects.py
Source:objects.py
1# Copyright 2019 Katteli Inc.2# TestFlows.com Open-Source Software Testing Framework (http://testflows.com)3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15import re16import base6417import hashlib18import threading19import testflows.settings as settings20from collections import namedtuple21from .utils.enum import IntEnum22from .exceptions import SpecificationError, RequirementError, ResultException23from .baseobject import TestObject, Table24from .baseobject import get, hash25from .testtype import TestType26import testflows._core.contrib.rsa as rsa27class Result(TestObject, ResultException):28 _fields = ("message", "reason", "type", "test")29 _defaults = (None,) * 630 metrics = []31 tickets = []32 values = []33 type = None34 class Type(IntEnum):35 OK = 136 Fail = 237 Error = 338 Null = 439 Skip = 540 XOK = 641 XFail = 742 XError = 843 XNull = 944 def __repr__(self):45 return f"Result.Type.{self._name_}"46 def __init__(self, message=None, reason=None, type=None, test=None, metrics=None, tickets=None, values=None, start_time=None, test_time=None):47 from .funcs import current48 self.test = test if test is not None else current().name49 if not isinstance(self.test, str):50 raise TypeError("test must be of 'str' type")51 self.type = get(type, self.type)52 if self.type is None:53 raise TypeError("type must be defined")54 self.message = message55 self.reason = reason56 self.metrics = get(metrics, list(self.metrics))57 self.tickets = get(tickets, list(self.tickets))58 self.values = get(values, list(self.values))59 self.start_time = start_time60 self.test_time = test_time61 return super(Result, self).__init__()62 def __reduce__(self):63 return (self.__class__, (self.message, self.reason, self.type, self.test, self.metrics, self.tickets, self.values, self.start_time, self.test_time))64 @property65 def value(self):66 return self.values[-1].value if self.values else None67 def __call__(self, result=None):68 if result is None:69 result = getattr(self.__module__, str(self.type))70 obj = result.__class__(*[getattr(result, field) for field in result._fields])71 obj.metrics = self.metrics72 obj.tickets = self.tickets73 obj.values = self.values74 obj.start_time = self.start_time75 obj.test_time = self.test_time76 return obj77 def xout(self, reason=None):78 if type(self) in [Result, XResult]:79 return self().xout(reason=reason)80 return self81 def __str__(self):82 return str(self.type)83 def __bool__(cls):84 return True85 def __eq__(self, o):86 return self.type == o.type87 def __ne__(self, o):88 return not self == o89class XResult(Result):90 pass91class OK(Result):92 type = Result.Type.OK93 def xout(self, reason):94 return self(XOK(test=self.test, message=self.message, reason=reason))95class XOK(XResult):96 type = Result.Type.XOK97class Fail(Result):98 type = Result.Type.Fail99 def xout(self, reason):100 return self(XFail(test=self.test, message=self.message, reason=reason))101 def __bool__(self):102 return False103class XFail(XResult):104 type = Result.Type.XFail105class Skip(Result):106 type = Result.Type.Skip107class Error(Result):108 type = Result.Type.Error109 def xout(self, reason):110 return self(XError(test=self.test, message=self.message, reason=reason))111 def __bool__(self):112 return False113class XError(XResult):114 type = Result.Type.XError115class Null(Result):116 type = Result.Type.Null117 def xout(self, reason):118 return self(XNull(test=self.test, message=self.message, reason=reason))119 def __bool__(self):120 return False121class XNull(XResult):122 type = Result.Type.XNull123XoutResults = (XOK, XFail, XError, XNull)124FailResults = (Fail, Error, Null)125PassResults = (OK,) + XoutResults126NonFailResults = (Skip,) + PassResults127class Node(TestObject):128 _fields = ("name", "module", "uid", "nexts", "ins", "outs")129 _defaults = (None,) * 3130 NodeAttributes = namedtuple("NodeAttributes", "name module uid")131 def __init__(self, name, module, uid, nexts=None, ins=None, outs=None):132 self.name = name133 self.module = module134 self.uid = uid135 self.nexts = get(nexts, [])136 self.ins = get(ins, [])137 self.outs =get(outs, [])138 return super(Node, self).__init__()139 @classmethod140 def attributes(cls, test):141 name = test.__name__142 module = ".".join([test.__module__, test.__name__])143 uid = hash(module, short=True)144 return cls.NodeAttributes(name, module, uid)145class Tag(TestObject):146 _fields = ("value",)147 _defaults = ()148 def __init__(self, value):149 self.value = value150 return super(Tag, self).__init__()151 def __str__(self):152 return str(self.value)153class Argument(TestObject):154 _fields = ("name", "value", "type", "group", "uid")155 _defaults = (None,) * 4156 uid = None157 type = None158 group = None159 def __init__(self, name, value=None, type=None, group=None, uid=None):160 self.name = name161 self.value = value162 self.type = get(type, self.type)163 self.group = get(group, self.group)164 self.uid = get(uid, self.uid)165 return super(Argument, self).__init__()166class Attribute(TestObject):167 _fields = ("name", "value", "type", "group", "uid")168 _defaults = (None,) * 3169 uid = None170 type = None171 group = None172 def __init__(self, name, value, type=None, group=None, uid=None):173 self.name = name174 self.value = value175 self.type = get(type, self.type)176 self.group = get(group, self.group)177 self.uid = get(uid, self.uid)178 return super(Attribute, self).__init__()179class Requirement(TestObject):180 _fields = ("name", "version", "description",181 "link", "priority", "type", "group", "uid", "level", "num")182 _defaults = (None,) * 8183 uid = None184 link = None185 priority = None186 type = None187 group = None188 def __init__(self, name, version, description=None, link=None,189 priority=None, type=None, group=None, uid=None, level=None, num=None):190 self.name = name191 self.version = version192 self.description = get(description, self.__doc__)193 self.link = get(link, self.link)194 self.priority = get(priority, self.priority)195 self.type = get(type, self.type)196 self.group = get(group, self.group)197 self.uid = get(uid, self.uid)198 self.level = level199 self.num = num200 return super(Requirement, self).__init__()201 def __call__(self, *version):202 if not self.version in version:203 raise RequirementError("requirement version %s is not in %s" % (self.version, list(version)))204 return self205class Metric(TestObject):206 _fields = ("name", "value", "units", "type", "group", "uid")207 _defaults = (None,) * 3208 uid = None209 type = None210 group = None211 def __init__(self, name, value, units, type=None, group=None, uid=None):212 self.name = name213 self.value = value214 self.units = units215 self.type = get(type, self.type)216 self.group = get(group, self.group)217 self.uid = get(uid, self.uid)218 return super(Metric, self).__init__()219class Value(TestObject):220 _fields = ("name", "value", "type", "group", "uid")221 _defaults = (None,) * 3222 uid = None223 type = None224 group = None225 def __init__(self, name, value, type=None, group=None, uid=None):226 self.name = name227 self.value = str(value)228 self.type = get(type, self.type)229 self.group = get(group, self.group)230 self.uid = get(uid, self.uid)231 return super(Value, self).__init__()232class Ticket(TestObject):233 _fields = ("name", "link", "type", "group", "uid")234 _defaults = (None,) * 4235 uid = None236 link = None237 type = None238 group = None239 def __init__(self, name, link=None, type=None, group=None, uid=None):240 self.name = name241 self.link = get(link, self.link)242 self.type = get(type, self.type)243 self.group = get(group, self.group)244 self.uid = get(uid, self.uid)245 return super(Ticket, self).__init__()246class Specification(TestObject):247 _fields = ("name", "content", "description", "link", "author", "version",248 "date", "status", "approved_by", "approved_date", "approved_version",249 "type", "group", "uid", "parent", "children", "headings", "requirements")250 _defaults = (None,) * 16251 uid = None252 link = None253 type = None254 group = None255 Heading = namedtuple("Heading", "name level num")256 def __init__(self, name, content, description=None, link=None, author=None, version=None,257 date=None, status=None, approved_by=None, approved_date=None, approved_version=None,258 type=None, group=None, uid=None, parent=None, children=None, headings=None,259 requirements=None):260 self.name = name261 self.content = content262 self.description = description263 self.author = author264 self.version = version265 self.date = date266 self.status = status267 self.approved_by = approved_by268 self.approved_date = approved_date269 self.approved_version = approved_version270 self.link = get(link, self.link)271 self.type = get(type, self.type)272 self.group = get(group, self.group)273 self.uid = get(uid, self.uid)274 self.parent = parent275 self.children = children276 self.headings = headings277 self.requirements = requirements278 def __call__(self, *version):279 if not self.version in version:280 raise SpecificationError("specification version %s is not in %s" % (self.version, list(version)))281 return self282class ExamplesRow(TestObject):283 _fields = ("row", "columns", "values", "row_format")284 _defaults = (None,)285 def __init__(self, row, columns, values, row_format=None):286 self.row = row287 self.columns = columns288 self.values = [str(value) for value in values]289 self.row_format = row_format290class Secrets:291 """Secrets registry.292 """293 def __init__(self, secrets=None):294 self._secrets = secrets or {}295 self._filter_regex = re.compile(r"")296 self._filter_secrets = []297 self._lock = threading.Lock()298 self._update_filter()299 300 def __reduce__(self):301 return (Secrets, (self._secrets,))302 def is_empty(self):303 """Return True if registry is empty.304 """305 with self._lock:306 return not bool(self._filter_secrets)307 def register(self, secret):308 """Register secret object.309 """310 with self._lock:311 if secret.name in self._secrets:312 raise ValueError(f"secret '{secret.name}' already registered")313 self._secrets[secret.name] = secret314 self._update_filter()315 def unregister(self, secret):316 """Unregister secret object.317 """318 with self._lock:319 self._secrets.pop(secret.name, None)320 self._update_filter()321 def _update_filter(self):322 """Update filter regex.323 """324 self._filter_secrets = [s for s in self._secrets.values() if s.is_set()]325 self._filter_regex = re.compile("|".join([f"(?P<{s.name}>{re.escape(s.value)})" for s in self._filter_secrets]))326 def filter(self, message):327 """Filter all secret values from message.328 """329 def _filter(s):330 if not isinstance(s, str):331 return message332 return self._filter_regex.sub(lambda m: f"[masked]:{self._filter_secrets[m.lastindex-1]}", s)333 334 with self._lock:335 if isinstance(message, (list, tuple)):336 for i, e in enumerate(message):337 message[i] = self.filter(e)338 return message339 elif isinstance(message, dict):340 _message = {}341 for k, v in message.items():342 _message[filter(k)] = filter(v)343 return _message344 return _filter(message)345class Secret(TestObject):346 """Secret value.347 """348 _fields = ("name", "type", "group", "uid")349 _defaults = (None,) * 4350 351 uid = None352 type = None353 name = None354 group = None355 def __init__(self, name=None, type=None, group=None, uid=None):356 self.name = get(name, self.name)357 if self.name is None:358 raise TypeError("name must be specified")359 else:360 try:361 re.compile(rf"(?P<{self.name}>)")362 except re.error as e:363 raise ValueError("invalid secret name, " + str(e).replace("group name ", "")) from None364 self.type = get(type, self.type)365 self.group = get(group, self.group)366 self.uid = get(uid, self.uid)367 self._value = None368 def __enter__(self):369 return self370 371 def __exit__(self, exc_value, exc_type, exc_tb):372 self.clear()373 def clear(self):374 self._value = None375 if not settings.secrets_registry:376 raise RuntimeError("no secrets registry")377 settings.secrets_registry.unregister(self)378 return self379 def __call__(self, value=None):380 if value is not None:381 self._value = str(value)382 if not settings.secrets_registry:383 raise RuntimeError("no secrets registry")384 settings.secrets_registry.register(self)385 return self386 def is_set(self):387 """Return true if value has been set.388 """389 return self._value is not None390 @property391 def value(self):392 """Return plaintext value of the secret.393 """394 if self._value is None:395 raise ValueError("no value")396 return self._value397 def __str__(self):398 return self.__repr__()399 def __repr__(self):400 """Custom object representation.401 """402 kwargs = []403 for field in self._fields:404 value = getattr(self, field)405 if value is None:406 continue407 kwargs.append(f"{field}={repr(value)}")408 return f"Secret({','.join(kwargs)})"409class RSASecret(Secret):410 """RSA encrypted secret value.411 """412 _fields = ("name", "type", "group", "uid", "code", "pubkey_id",)413 _defaults = (None,) * 6414 uid = None415 name = None416 type = None417 group = None418 encoding = "utf-8"419 def __init__(self, name=None, type=None, group=None, uid=None, code=None, pubkey_id=None):420 self.name = get(name, None)421 self.type = get(type, self.type)422 self.group = get(group, self.group)423 self.uid = get(uid, self.uid)424 self.pubkey_id = get(pubkey_id, None)425 if self.pubkey_id is not None:426 self.pubkey_id = base64.b64decode(pubkey_id.encode(self.encoding))427 self.code = get(code, None)428 self._value = None429 self.__call__(code=code)430 def __call__(self, value=None, code=None, private_key=None, public_key=None):431 if value is not None:432 self._value = str(value)433 self.code = None434 if code is not None:435 self.code = base64.b64decode(code.encode(self.encoding))436 self._value = None437 if public_key is not None:438 self.pubkey_id = hashlib.sha1(public_key.save_pkcs1()).digest()439 self.code = None440 if self._value is None:441 raise ValueError("no value")442 self.code = rsa.encrypt(self._value.encode(self.encoding), public_key)443 if private_key is not None:444 if self.pubkey_id is not None:445 pk_id = hashlib.sha1(private_key.pubkey.save_pkcs1()).digest()446 if self.pubkey_id != pk_id:447 raise ValueError("wrong private key")448 if self.code is None:449 raise ValueError("no code")450 self._value = rsa.decrypt(self.code, private_key).decode(self.encoding)451 return self452 def __repr__(self):453 """Custom object representation.454 """455 kwargs = []456 for field in self._fields:457 value = getattr(self, field)458 if value is None:459 continue460 if field in ("code", "pubkey_id"):461 kwargs.append(f"{field}='{base64.b64encode(value).decode('utf-8')}'")462 else:463 kwargs.append(f"{field}={repr(value)}")464 return f"Secret({','.join(kwargs)})"465class ExamplesTable(Table):466 _row_type_name = "Example"467 def __new__(cls, header=None, rows=None, row_format=None, args=None):468 if rows is None:469 rows = []470 if header is None:471 header = ""472 if args is None:473 args = {}474 else:475 args = dict(args)476 row_type = namedtuple(cls._row_type_name, header)477 class ExampleRow(row_type):478 def __new__(cls, *args):479 _args = {}480 args = list(args)481 len_header = len(header.split(" "))482 if len(args) > len_header:483 _args = {k:v for arg in args[len_header:] for k, v in dict(arg).items()}484 if "type" in args:485 raise TypeError("can't specify 'type' using example arguments")486 if "args" in args:487 raise TypeError("can't specify 'args' using example arguments")488 del args[len_header:]489 obj = super(ExampleRow, cls).__new__(cls, *args)490 obj._args = _args491 return obj492 obj = super(ExamplesTable, cls).__new__(cls, header, rows, row_format, ExampleRow)493 for idx, row in enumerate(obj):494 row._idx = idx495 row._row_format = obj.row_format496 obj.args = args497 return obj498class NamedValue(object):499 name = None500 def __init__(self, value):501 self.value = value502 def keys(self):503 return [self.name]504 def __getitem__(self, key):505 if key == self.name:506 return self.value507 raise KeyError(key)508 def __call__(self, func):509 setattr(func, self.name, self.value)510 return func511class NamedString(NamedValue):512 def __str__(self):513 return str(self.value)514class NamedList(list):515 name = None516 def __init__(self, *items):517 super(NamedList, self).__init__(items)518 def keys(self):519 return [self.name]520 def __getitem__(self, key):521 if key == self.name:522 return list(self)523 return super(NamedList, self).__getitem__(key)524 def __call__(self, func):525 setattr(func, self.name, list(self))526 return func527class Onlys(NamedList):528 """only container.529 ```python530 @Only(531 pattern,532 ...533 )534 ```535 """536 name = "only"537 def __init__(self, *items):538 super(Onlys, self).__init__(*items)539class Skips(NamedList):540 """skip container.541 ```python542 @Skips(543 pattern,544 ...545 )546 ```547 """548 name = "skip"549 def __init__(self, *items):550 super(Skips, self).__init__(*items)551class _FilterTags(NamedValue):552 """filter tags object.553 """554 def __init__(self, test=None, suite=None, module=None, any=None):555 test = set(test) if test is not None else set()556 suite = set(suite) if suite is not None else set()557 module = set(module) if module is not None else set()558 any = set(any) if any is not None else set()559 if any:560 test = test.union(any)561 suite = suite.union(any)562 module = module.union(any)563 super(_FilterTags, self).__init__({TestType.Test: test, TestType.Suite: suite, TestType.Module: module})564class OnlyTags(_FilterTags):565 """only_tags filter object.566 ```python567 @OnlyTags(568 test=[tagA,(tagA,tagB),...],569 suite=[...],570 module=[...],571 any=[...]572 )573 ```574 """575 name = "only_tags"576class SkipTags(_FilterTags):577 """skip_tags filter object.578 ```python579 @SkipTags(580 test=[tagA,(tagA,tagB),...],581 suite=[...],582 module=[...],583 any=[...]584 )585 ```586 """587 name = "skip_tags"588class Setup(NamedValue):589 name = "setup"590class XFails(NamedValue):591 """xfails container.592 xfails = {593 "pattern": [("result", "reason"[, when])],594 ...595 }596 """597 name = "xfails"598 def __init__(self, value):599 super(XFails, self).__init__(dict(value))600 def items(self):601 return self.value.items()602 def add(self, pattern, *results):603 """Add an entry to the xfails.604 :param pattern: test name pattern to match605 :param *results: one or more results to cross out606 where each result is a two-tuple or three-tuple of (result, reason[, when])607 """608 self.value[pattern] = results609 return self610class FFails(NamedValue):611 """ffails (forced fails) container.612 ffails = {613 "pattern": (result, "reason"[, when]),614 ...615 }616 """617 name = "ffails"618 def __init__(self, value):619 value = dict(value)620 value = {p: list(FFail(*v).value.values())[0] for p, v in value.items()}621 super(FFails, self).__init__(value)622 def items(self):623 return self.value.items()624 def add(self, pattern, reason, result=Fail, when=None):625 """Add an entry to the ffails (force fails).626 :param pattern: test name pattern to match627 :param reason: reason628 :param result: forced result, default: Fails629 :param when: when filter630 """631 if when is not None and not callable(when):632 raise TypeError(f"invalid when type '{type(when)}'; must be callable")633 self.value[pattern] = (result, reason, when)634 return self635class FFail(NamedValue):636 """ffails (forced fails) container with single result.637 """638 name = "ffails"639 def __init__(self, result, reason, when=None, pattern=""):640 if not issubclass(result, Result):641 raise TypeError(f"invalid result '{result}' type")642 if not type(reason) in (str,):643 raise TypeError(f"reason '{type(reason)}' must be str")644 if when is not None and not callable(when):645 raise TypeError(f"when '{type(when)}' must be callable")646 647 super(FFail, self).__init__({pattern: (result, reason, when)})648class Skipped(FFail):649 """ffails (forced fails) container with single Skip result.650 """651 def __init__(self, reason, when=None, pattern=""):652 super(Skipped, self).__init__(reason=reason, when=when, pattern=pattern, result=Skip)653class Failed(FFail):654 """ffails (forced fails) container with single Fail result.655 """656 def __init__(self, reason, when=None, pattern=""):657 super(Failed, self).__init__(reason=reason, when=when, pattern=pattern, result=Fail)658class XFailed(FFail):659 """ffails (forced fails) container with single XFail result.660 """661 def __init__(self, reason, when=None, pattern=""):662 super(XFailed, self).__init__(reason=reason, when=when, pattern=pattern, result=XFail)663class XErrored(FFail):664 """ffails (forced fails) container with single XError result.665 """666 def __init__(self, reason, when=None, pattern=""):667 super(XErrored, self).__init__(reason=reason, when=when, pattern=pattern, result=XError)668class Okayed(FFail):669 """ffails (forced fails) container with single OK result.670 """671 def __init__(self, reason, when=None, pattern=""):672 super(Okayed, self).__init__(reason=reason, when=when, pattern=pattern, result=OK)673class XOkayed(FFail):674 """ffails (forced fails) container with single XOK result.675 """676 def __init__(self, reason, when=None, pattern=""):677 super(XOkayed, self).__init__(reason=reason, when=when, pattern=pattern, result=XOK)678class XFlags(NamedValue):679 """xflags container.680 xflags = {681 "filter": (set_flags, clear_flags[, when]),682 ...683 }684 """685 name = "xflags"686 def __init__(self, value):687 super(XFlags, self).__init__(dict(value))688 def items(self):689 return self.value.items()690 def add(self, pattern, set_flags=0, clear_flags=0, when=None):691 """Add an entry to the xflags.692 :param pattern: test name pattern to match693 :param set_flags: flags to set694 :param clear_flags: flags to clear, default: None695 :param when: condition function, default: None696 """697 self.value[pattern] = [Flags(set_flags), Flags(clear_flags), when]698 return self699class Repeats(NamedValue):700 """repeats containers.701 repeats={702 "pattern": (count, until),703 ...704 }705 """706 name = "repeats"707 def __init__(self, value):708 super(Repeats, self).__init__(dict(value))709class Repeat(NamedValue):710 """single repetition container.711 """712 name = "repeats"713 def __init__(self, count, until="complete", pattern=""):714 self.count = int(count)715 self.pattern = str(pattern)716 self.until = str(until)717 if self.count < 1:718 raise ValueError("count must be > 0")719 if self.until not in ("fail", "pass", "complete"):720 raise ValueError("invalid until value")721 return super(Repeat, self).__init__({self.pattern: (self.count, self.until)})722class Retries(NamedValue):723 """retries containers.724 retries={725 "pattern": count[,timeout[,delay[,backoff[,jitter]]]] ,726 ...727 }728 """729 name = "retries"730 def __init__(self, value):731 value = dict(value)732 value = {p: list(Retry(*r).value.values())[0] for p, r in value.items()}733 super(Retries, self).__init__(value)734class Retry(NamedValue):735 """single retry container.736 """737 name = "retries"738 def __init__(self, count=None, timeout=None, delay=0, backoff=1, jitter=None, pattern=""):739 """740 :param count: number of retries, default: None741 :param timeout: timeout in sec, default: None742 :param delay: delay in sec between retries, default: 0 sec743 :param backoff: backoff multiplier that is applied to the delay, default: 1744 :param jitter: jitter added to delay between retries specified as745 a tuple(min, max), default: (0,0)746 """747 self.count = int(count) if count is not None else None748 self.timeout = float(timeout) if timeout is not None else None749 self.delay = float(delay)750 self.backoff = backoff751 self.jitter = tuple(jitter) if jitter else tuple([0, 0])752 753 self.pattern = str(pattern)754 if self.count is not None and self.count < 1:755 raise ValueError("count must be > 0")756 if self.timeout is not None and self.timeout < 0:757 raise ValueError("timeout must be >= 0")758 return super(Retry, self).__init__({self.pattern: (self.count, self.timeout, self.delay, self.backoff, self.jitter)})759class Args(dict):760 def __init__(self, **args):761 super(Args, self).__init__(**args)762 def __call__(self, func):763 for k, v in self.items():764 if not k.startswith("_"):765 setattr(func, k, v)766 return func767class Attributes(NamedList):768 name = "attributes"769 def __init__(self, *attributes):770 super(Attributes, self).__init__(*[Attribute(*a) for a in attributes])771class Requirements(NamedList):772 name = "requirements"773 def __init__(self, *requirements):774 super(Requirements, self).__init__(*[Requirement(*r) for r in requirements])775class Specifications(NamedList):776 name = "specifications"777 def __init__(self, *specifications):778 super(Specifications, self).__init__(*[Specification(*r) for r in specifications])779class Tags(NamedList):780 name = "tags"781 def __init__(self, *tags):782 super(Tags, self).__init__(*tags)783class Uid(NamedString):784 name = "uid"785class Parallel(NamedValue):786 name = "parallel"787 def __init__(self, value):788 self.value = bool(value)789class Executor(NamedValue):790 name = "executor"791 def __init__(self, executor):792 self.value = executor793class ArgumentParser(NamedValue):794 name = "argparser"795 def __init__(self, parser):796 self.value = parser797class Name(NamedString):798 name = "name"799class Description(NamedString):800 name = "description"801class Examples(ExamplesTable):802 def __new__(cls, header, rows, row_format=None, args=None):803 return super(Examples, cls).__new__(cls, header=header,804 rows=rows, row_format=row_format, args=args)805 def __call__(self, func):806 func.examples = self...
logger.py
Source:logger.py
...58 ) -> None:59 """60 Low-level log implementation, proxies to allow nested logger adapters.61 """62 msg = self._filter_secrets(msg)63 args = self._filter_secrets(args)64 return super()._log(65 level,66 msg,67 args,68 exc_info=exc_info,69 extra=extra,70 stack_info=stack_info,71 stacklevel=stacklevel,72 )73 def _filter_secrets(self, value: Any) -> Any:74 if isinstance(value, str):75 value = mask(value)76 elif isinstance(value, Exception):77 value_args = list(value.args)78 for index, arg_item in enumerate(value.args):79 if isinstance(value_args[index], str):80 value_args[index] = mask(arg_item)81 value.args = tuple(value_args)82 elif isinstance(value, tuple):83 value_list = self._filter_secrets(list(value))84 value = tuple(value_list)85 elif isinstance(value, list):86 for index, item in enumerate(value):87 value[index] = self._filter_secrets(item)88 return value89class LogWriter(object):90 def __init__(self, logger: Logger, level: int):91 self._level = level92 self._log = logger93 self._buffer: str = ""94 def write(self, message: str) -> None:95 self._buffer = "".join([self._buffer, message])96 if "\n" in message:97 self.flush()98 def flush(self) -> None:99 if len(self._buffer) > 0:100 self._log.lines(self._level, self._buffer)101 self._buffer = ""...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!