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

How to use t_default method in avocado

Best Python code snippet using avocado_python

fields.py

Source:fields.py

1import inspect2from decimal import Decimal3from functools import cache4from types import MappingProxyType5from typing import Pattern, Iterable, TypeVar, Callable, Type, Any, Optional6from django.core import validators as dj_validators7from django.core.exceptions import ValidationError8from django.db.models import Model9from django.utils.translation import gettext_lazy as _10import validators as _validators11T = TypeVar('T')12U = TypeVar('U')13T_DEFAULT = Optional[T | Callable[[], T]]14T_VALUE_MSG = T | tuple[T, str]15# todo: simplify regexes?16T_REGEXES = (17    str |                                # non-compiled regex18    Pattern |                            # compiled regex19    Iterable[Pattern | str] |            # iterable of compiled or non-compiled regexes20    Iterable[tuple[Pattern | str, str]]  # iterable of compiled or non-compiled regexes and their errors21)22T_VALIDATORS = Iterable[Callable[[Any], Any]]23T_DUNDER_CALL = Callable[[Optional[T]], None]24# use this instead of 'None' when 'None' is also a valid value25_UNDEFINED = object()26# empty immutable dict27_EMPTY_DICT = MappingProxyType({})28def get_value_and_error(v: T_VALUE_MSG[T], default_error: str = None) -> tuple[T, str | None]:29    if isinstance(v, Iterable) and not isinstance(v, str):30        val = tuple(v)31        val, error = val32        return val, error or default_error33    return v, default_error34def merge_serializer_errors(errors: list['SerializerError']):35    errors_dict = {}36    for err in errors:37        errors_dict.update(err.parse_data())  # todo: warning or exception on conflicts38    return SerializerError(errors_dict)39class SerializerError(ValidationError):40    def __init__(self, data: dict, message: str = 'Serializer error.', code: str = 'serializer_error'):41        super().__init__(message, code)42        self.data = data43    def parse_data(self, data: dict = None, only_first: bool = False, only_messages: bool = False) -> dict:44        data = data if data else self.data45        def errors_from_exception(error: ValidationError):46            _errors = error.error_list[0:(1 if only_first else None)]47            _errors_list = []48            for e in _errors:49                msg = e.message % (e.params or _EMPTY_DICT)50                _errors_list.append(msg if only_messages else {'code': e.code, 'message': msg})51            return _errors_list52        errors = {}53        for key, value in data.items():54            if isinstance(value, SerializerError):55                errors[key] = self.parse_data(value.data, only_first, only_messages)56            else:57                current_errors = sum([errors_from_exception(v) for v in value], [])58                errors[key] = current_errors[0] if only_first else current_errors59            continue60        return errors61# todo: more elegant error definition62class BaseField:63    ERROR_NULL = _('Field value can not be null.')64    ERROR_REQUIRED = _('This field is required.')65    ERROR_READ_ONLY = _('This field is read only.')66    ERROR_WRITE_ONLY = _('This field is write only')67    ERROR_TYPE = _('Field is in incorrect type.')  # todo: add list of supported types68    def __init__(69            self, *, source: str = None, getter: str = None, setter: str = None, default: T_DEFAULT[Any] = _UNDEFINED,70            allow_null: bool = False, is_required: bool = True, is_snippet: bool = True,71            read_only: bool = False, write_only: bool = False,72            validators: Iterable[Callable[[Any], Any]] = None, json_type: Type | tuple[type, ...] = None73    ):74        assert not (source and (getter or setter)), "'source' can not be combined with 'getter' or 'setter'"75        assert not (write_only and read_only), "Field can not be 'write_only' and 'read_only' at the same time"76        assert not (read_only and is_required), "Field can not be 'read_only' and 'is_required' at the same time"77        assert not (write_only and is_snippet), "Field can not be 'write_only' and 'is_snippet' at the same time"78        assert not (is_required and default != _UNDEFINED), "Field can not be 'is_required' and have a 'default' set"79        assert not (not allow_null and default is None), "'default' can not be 'None' if 'allow_null' is 'False'"80        self.source = source  # acts as getter and setter at the same time81        self.getter = getter  # model field name to get value82        self.setter = setter  # model field name to store value83        # list of django validators84        self.validators: list[Callable[[Any], Any]] = list(validators) if validators else []85        self.json_type: tuple[type] = json_type86        if self.json_type:87            self.json_type = (self.json_type,) if isinstance(self.json_type, type) else tuple(self.json_type)88        self.allow_null = allow_null89        self.is_required = is_required90        self.read_only = read_only91        self.write_only = write_only92        self.is_snippet = is_snippet93        # default value, can be a function with no mandatory arguments94        self.default: T_DEFAULT[Any] = default95    @staticmethod96    def to_python(json_value):97        """Prepares JSON data to be converted to Python"""98        return json_value99    @staticmethod100    def to_json(python_value):101        """Prepares Python data to be converted to JSON"""102        return python_value103    def is_valid_field_base(self, value) -> None:104        """Base validation of a field (type, 'allow_null', etc.)"""105        if value is None:106            if not self.allow_null:107                raise ValidationError([ValidationError(message=self.ERROR_NULL, code='null_not_allowed')])108            return109        if self.json_type and not isinstance(value, self.json_type):110            raise ValidationError([ValidationError(message=self.ERROR_TYPE, code='wrong_type')])111    def is_valid_field(self, value) -> None:112        """Validates a field"""113        self.is_valid_field_base(value)114        errors = []115        for validator in self.validators:116            try:117                validator(value)118            except ValidationError as e:119                errors.append(e)120        if len(errors) > 0 and isinstance(errors[0], SerializerError):121            if len(errors) == 1:122                raise errors[0]123            serializer_error = merge_serializer_errors([i for i in errors if isinstance(i, SerializerError)])124            raise serializer_error125        elif errors:126            raise ValidationError(errors)127class Serializer(BaseField):128    ERROR_FORBIDDEN_ELEMENTS = _('Mapping contains forbidden elements.')  # todo: specify which?129    def __init__(self, *, instance: Model = None, data: dict = None, is_partial: bool = False, **kwargs):130        assert not (kwargs and instance), "You can not use a serializer as a field with passed 'instance' to it"131        assert not (kwargs and data is not None), "You can not use a serializer as a field with passed 'data' to it"132        assert not (data and is_partial), "It seems you wanted to partially validate data you passed as 'data' argument. " \133                                          "If so, 'is_partial' must be passed into `.validate` method"134        super().__init__(json_type=dict, **kwargs)135        self.is_partial = is_partial136        self.instance = instance137        self.data = data138    def is_valid_field(self, value: dict, only_first: bool = False, is_partial: bool = False) -> None:139        """Acts as a validator of a serializer as a field"""140        self.is_valid_field_base(value)141        fields = self.get_fields()142        errors = {}143        for field_name, field in fields.items():144            field: BaseField145            if field_name in value:146                try:147                    field.is_valid_field(value[field_name])148                except SerializerError as error:149                    errors[field_name] = error150                except ValidationError as error:151                    errors[field_name] = errors.get(field_name, []) + [error]152            else:153                if field.is_required and not is_partial:154                    errors[field_name] = errors.get(field_name, []) + [ValidationError(field.ERROR_REQUIRED)]155        if errors:156            raise SerializerError(errors)157    def is_valid(self, *, is_partial: bool = False, raise_error: bool = False) -> bool:158        """Validates a serializer"""159        if self.data is None:160            raise ValueError("Could not validate, because 'data' was not provided")161        try:162            self.is_valid_field(self.data, is_partial=is_partial)163            return True164        except ValidationError as e:165            if raise_error:166                raise e167            return False168    @classmethod169    @cache170    def get_fields(cls) -> MappingProxyType[str, BaseField]:171        """Returns an immutable dict of fields in form of {field_name: Field}"""172        fields = inspect.getmembers(cls, lambda x: not inspect.isroutine(x) and isinstance(x, BaseField))173        return MappingProxyType(dict(fields))174    @classmethod175    def get_required_fields(cls) -> set[str]:176        """Returns a set of required field names"""177        fields = cls.get_fields()178        required_fields = {k for k, v in fields.items() if v.is_required}179        return required_fields180    @classmethod181    def get_snippet_fields(cls) -> set[str]:182        """Returns a set of snippet field names"""183        fields = cls.get_fields()184        snippet_fields = {k for k, v in fields.items() if v.is_snippet}185        return snippet_fields186class BooleanField(BaseField):187    def __init__(self, *, default: T_DEFAULT[bool] = _UNDEFINED, **kwargs):188        super().__init__(**kwargs, json_type=bool, default=default)189class CharField(BaseField):190    def __init__(191            self, *, min_length: int = None, max_length: int = None, regexes: T_REGEXES = None,192            default: T_DEFAULT[str] = _UNDEFINED, **kwargs193    ):194        super().__init__(**kwargs, json_type=str, default=default)195        self.default: T_DEFAULT[str]196        self.min_length = min_length197        self.max_length = max_length198        if self.min_length and self.max_length:199            assert self.min_length <= self.max_length, "'max_length' must be greater or equal to 'min_length'"200        if self.min_length:201            self.validators.append(dj_validators.MinLengthValidator(min_length))202        if self.max_length:203            self.validators.append(dj_validators.MaxLengthValidator(max_length))204        if regexes:205            if isinstance(regexes, str):206                regexes_and_errors = get_value_and_error(regexes),207            else:208                vals = tuple(get_value_and_error(val) for val in regexes)209                regexes_and_errors = vals210            for regex, error in regexes_and_errors:211                self.validators.append(dj_validators.RegexValidator(regex, error))212class NumericField(BaseField):213    def __init__(214            self, *, min_value: int = None, max_value: int = None,215            default: T_DEFAULT[int] = _UNDEFINED, **kwargs216    ):217        super().__init__(json_type=kwargs.pop('json_type', (int, float)), **kwargs, default=default)218        self.default: T_DEFAULT[int]219        self.min_value = min_value220        self.max_value = max_value221        if self.min_value and self.max_value:222            assert self.min_value <= self.max_value, "'max_value' must be greater or equal to 'min_value'"223        if self.min_value:224            self.validators.append(dj_validators.MinValueValidator(min_value))225        if self.max_value:226            self.validators.append(dj_validators.MaxValueValidator(max_value))227class IntegerField(NumericField):228    def __init__(self, *, default: T_DEFAULT[int] = _UNDEFINED, **kwargs):229        super().__init__(**kwargs, json_type=int, default=default)230        self.default: T_DEFAULT[int]231class FloatField(NumericField):232    def __init__(self, *, default: T_DEFAULT[float] = _UNDEFINED, **kwargs):233        super().__init__(**kwargs, json_type=float, default=default)234        self.default: T_DEFAULT[float]235class DecimalField(NumericField):236    def __init__(237            self, *, max_digits: int = None, decimal_places: int = None,238            default: T_DEFAULT[Decimal] = _UNDEFINED, **kwargs239    ):240        super().__init__(**kwargs, json_type=float, default=default)  # todo: cast float into decimal241        self.max_digits = max_digits242        self.decimal_places = decimal_places243        if self.max_digits or self.decimal_places:244            self.validators.append(dj_validators.DecimalValidator(self.max_digits, self.decimal_places))245    @staticmethod246    def to_json(python_value: Decimal):247        return float(python_value)248    @staticmethod249    def to_python(json_value: float):250        return Decimal(json_value)251class DictField(BaseField):252    def __init__(253            self, *, min_items: int, max_items: int, key_field: CharField = None, value_field: BaseField = None,254            default: T_DEFAULT[dict] = _UNDEFINED, **kwargs255    ):256        super().__init__(**kwargs, json_type=dict, default=default)257        self.default: T_DEFAULT[dict]258        self.min_items = min_items259        self.max_items = max_items260        if self.min_items and self.max_items:261            assert self.min_items <= self.max_items, "'max_items' must be greater or equal to 'min_items'"262        if self.min_items:263            self.validators.append(dj_validators.MinLengthValidator(self.min_items))  # todo: correct error264        if self.max_items:265            self.validators.append(dj_validators.MaxLengthValidator(self.max_items))  # todo: correct error266        self.key_field = key_field267        self.value_field = value_field268        if self.key_field:269            assert not self.key_field.allow_null, "Dict keys can not be 'None', please edit 'allow_null' argument"270            self.validators.append(_validators.DictFieldKeyValidator(self.key_field))271        if self.value_field:272            self.validators.append(_validators.DictFieldValueValidator(self.value_field))273class ListField(BaseField):274    def __init__(275            self, *, min_length=None, max_length=None, unique_items=False, item_field: BaseField = None,276            default: T_DEFAULT[list | tuple] = _UNDEFINED, **kwargs277    ):278        super().__init__(**kwargs, json_type=list, default=default)279        self.default: T_DEFAULT[list | tuple]280        self.min_length: Optional[int]281        self.max_length: Optional[int]282        self.unique_items: bool283        self.min_length = min_length284        self.max_length = max_length285        self.unique_items = unique_items286        if self.min_length and self.max_length:287            assert self.min_length <= self.max_length, "'max_length' must be greater or equal to 'min_length'"288        if self.min_length:289            self.validators.append(dj_validators.MinLengthValidator(self.min_length))  # todo: correct error290        if self.max_length:291            self.validators.append(dj_validators.MaxLengthValidator(self.max_length))  # todo: correct error292        if self.unique_items:293            self.validators.append(_validators.UniqueItemsValidator())294        self.item_field = item_field295        if self.item_field:...

test_transport.py

Source:test_transport.py

1# coding=utf-82import os3import mock4import unittest5from winrm import transport6from winrm.exceptions import WinRMError, InvalidCredentialsError7class TestTransport(unittest.TestCase):8    maxDiff = 20489    _old_env = None10    def setUp(self):11        super(TestTransport, self).setUp()12        self._old_env = {}13        os.environ.pop('REQUESTS_CA_BUNDLE', None)14        os.environ.pop('TRAVIS_APT_PROXY', None)15        os.environ.pop('CURL_CA_BUNDLE', None)16        os.environ.pop('HTTPS_PROXY', None)17        os.environ.pop('HTTP_PROXY', None)18        os.environ.pop('NO_PROXY', None)19        transport.DISPLAYED_PROXY_WARNING = False20        transport.DISPLAYED_CA_TRUST_WARNING = False21    def tearDown(self):22        super(TestTransport, self).tearDown()23        os.environ.pop('REQUESTS_CA_BUNDLE', None)24        os.environ.pop('TRAVIS_APT_PROXY', None)25        os.environ.pop('CURL_CA_BUNDLE', None)26        os.environ.pop('HTTPS_PROXY', None)27        os.environ.pop('HTTP_PROXY', None)28        os.environ.pop('NO_PROXY', None)29    def test_build_session_cert_validate_default(self):30        t_default = transport.Transport(endpoint="https://example.com",31                                        username='test',32                                        password='test',33                                        auth_method='basic',34                                        )35        t_default.build_session()36        self.assertEqual(True, t_default.session.verify)37    def test_build_session_cert_validate_default_env(self):38        os.environ['REQUESTS_CA_BUNDLE'] = 'path_to_REQUESTS_CA_CERT'39        t_default = transport.Transport(endpoint="https://example.com",40                                        username='test',41                                        password='test',42                                        auth_method='basic',43                                        )44        t_default.build_session()45        self.assertEqual('path_to_REQUESTS_CA_CERT', t_default.session.verify)46    def test_build_session_cert_validate_1(self):47        os.environ['REQUESTS_CA_BUNDLE'] = 'path_to_REQUESTS_CA_CERT'48        t_default = transport.Transport(endpoint="https://example.com",49                                        server_cert_validation='validate',50                                        username='test',51                                        password='test',52                                        auth_method='basic',53                                        )54        t_default.build_session()55        self.assertEqual('path_to_REQUESTS_CA_CERT', t_default.session.verify)56    def test_build_session_cert_validate_2(self):57        os.environ['CURL_CA_BUNDLE'] = 'path_to_CURL_CA_CERT'58        t_default = transport.Transport(endpoint="https://example.com",59                                        server_cert_validation='validate',60                                        username='test',61                                        password='test',62                                        auth_method='basic',63                                        )64        t_default.build_session()65        self.assertEqual('path_to_CURL_CA_CERT', t_default.session.verify)66    def test_build_session_cert_override_1(self):67        os.environ['REQUESTS_CA_BUNDLE'] = 'path_to_REQUESTS_CA_CERT'68        t_default = transport.Transport(endpoint="https://example.com",69                                        server_cert_validation='validate',70                                        username='test',71                                        password='test',72                                        auth_method='basic',73                                        ca_trust_path='overridepath',74                                        )75        t_default.build_session()76        self.assertEqual('overridepath', t_default.session.verify)77    def test_build_session_cert_override_2(self):78        os.environ['CURL_CA_BUNDLE'] = 'path_to_CURL_CA_CERT'79        t_default = transport.Transport(endpoint="https://example.com",80                                        server_cert_validation='validate',81                                        username='test',82                                        password='test',83                                        auth_method='basic',84                                        ca_trust_path='overridepath',85                                        )86        t_default.build_session()87        self.assertEqual('overridepath', t_default.session.verify)88    def test_build_session_cert_override_3(self):89        os.environ['CURL_CA_BUNDLE'] = 'path_to_CURL_CA_CERT'90        t_default = transport.Transport(endpoint="https://example.com",91                                        server_cert_validation='validate',92                                        username='test',93                                        password='test',94                                        auth_method='basic',95                                        ca_trust_path=None,96                                        )97        t_default.build_session()98        self.assertEqual(True, t_default.session.verify)99    def test_build_session_cert_ignore_1(self):100        os.environ['REQUESTS_CA_BUNDLE'] = 'path_to_REQUESTS_CA_CERT'101        os.environ['CURL_CA_BUNDLE'] = 'path_to_CURL_CA_CERT'102        t_default = transport.Transport(endpoint="https://example.com",103                                        server_cert_validation='ignore',104                                        username='test',105                                        password='test',106                                        auth_method='basic',107                                        )108        t_default.build_session()109        self.assertIs(False, t_default.session.verify)110    def test_build_session_cert_ignore_2(self):111        os.environ['REQUESTS_CA_BUNDLE'] = 'path_to_REQUESTS_CA_CERT'112        os.environ['CURL_CA_BUNDLE'] = 'path_to_CURL_CA_CERT'113        t_default = transport.Transport(endpoint="https://example.com",114                                        server_cert_validation='ignore',115                                        username='test',116                                        password='test',117                                        auth_method='basic',118                                        ca_trust_path='boguspath'119                                        )120        t_default.build_session()121        self.assertIs(False, t_default.session.verify)122    def test_build_session_proxy_none(self):123        os.environ['HTTP_PROXY'] = 'random_proxy'124        os.environ['HTTPS_PROXY'] = 'random_proxy_2'125        t_default = transport.Transport(endpoint="https://example.com",126                                        server_cert_validation='validate',127                                        username='test',128                                        password='test',129                                        auth_method='basic',130                                        proxy=None131                                        )132        t_default.build_session()133        self.assertEqual({'no_proxy': '*'}, t_default.session.proxies)134    def test_build_session_proxy_defined(self):135        t_default = transport.Transport(endpoint="https://example.com",136                                        server_cert_validation='validate',137                                        username='test',138                                        password='test',139                                        auth_method='basic',140                                        proxy='test_proxy'141                                        )142        t_default.build_session()143        self.assertEqual({'http': 'test_proxy', 'https': 'test_proxy'}, t_default.session.proxies)144    def test_build_session_proxy_defined_and_env(self):145        os.environ['HTTPS_PROXY'] = 'random_proxy'146        t_default = transport.Transport(endpoint="https://example.com",147                                        server_cert_validation='validate',148                                        username='test',149                                        password='test',150                                        auth_method='basic',151                                        proxy='test_proxy'152                                        )153        t_default.build_session()154        self.assertEqual({'http': 'test_proxy', 'https': 'test_proxy'}, t_default.session.proxies)155    def test_build_session_proxy_with_env_https(self):156        os.environ['HTTPS_PROXY'] = 'random_proxy'157        t_default = transport.Transport(endpoint="https://example.com",158                                        server_cert_validation='validate',159                                        username='test',160                                        password='test',161                                        auth_method='basic',162                                        )163        t_default.build_session()164        self.assertEqual({'https': 'random_proxy'}, t_default.session.proxies)165    def test_build_session_proxy_with_env_http(self):166        os.environ['HTTP_PROXY'] = 'random_proxy'167        t_default = transport.Transport(endpoint="https://example.com",168                                        server_cert_validation='validate',169                                        username='test',170                                        password='test',171                                        auth_method='basic',172                                        )173        t_default.build_session()174        self.assertEqual({'http': 'random_proxy'}, t_default.session.proxies)175    def test_build_session_server_cert_validation_invalid(self):176        with self.assertRaises(WinRMError) as exc:177            transport.Transport(endpoint="Endpoint",178                                server_cert_validation='invalid_value',179                                username='test',180                                password='test',181                                auth_method='basic',182                                )183        self.assertEqual('invalid server_cert_validation mode: invalid_value', str(exc.exception))184    def test_build_session_krb_delegation_as_str(self):185        winrm_transport = transport.Transport(endpoint="Endpoint",186                                              server_cert_validation='validate',187                                              username='test',188                                              password='test',189                                              auth_method='kerberos',190                                              kerberos_delegation='True'191                                              )192        self.assertTrue(winrm_transport.kerberos_delegation)193    def test_build_session_krb_delegation_as_invalid_str(self):194        with self.assertRaises(ValueError) as exc:195            transport.Transport(endpoint="Endpoint",196                                server_cert_validation='validate',197                                username='test',198                                password='test',199                                auth_method='kerberos',200                                kerberos_delegation='invalid_value'201                                )202        self.assertEqual("invalid truth value 'invalid_value'", str(exc.exception))203    def test_build_session_no_username(self):204        with self.assertRaises(InvalidCredentialsError) as exc:205            transport.Transport(endpoint="Endpoint",206                                server_cert_validation='validate',207                                password='test',208                                auth_method='basic',209                                )210        self.assertEqual("auth method basic requires a username", str(exc.exception))211    def test_build_session_no_password(self):212        with self.assertRaises(InvalidCredentialsError) as exc:213            transport.Transport(endpoint="Endpoint",214                                server_cert_validation='validate',215                                username='test',216                                auth_method='basic',217                                )218        self.assertEqual("auth method basic requires a password", str(exc.exception))219    def test_build_session_invalid_auth(self):220        winrm_transport = transport.Transport(endpoint="Endpoint",221                                              server_cert_validation='validate',222                                              username='test',223                                              password='test',224                                              auth_method='invalid_value',225                                              )226        with self.assertRaises(WinRMError) as exc:227            winrm_transport.build_session()228        self.assertEqual("unsupported auth method: invalid_value", str(exc.exception))229    def test_build_session_invalid_encryption(self):230        with self.assertRaises(WinRMError) as exc:231            transport.Transport(endpoint="Endpoint",232                                server_cert_validation='validate',233                                username='test',234                                password='test',235                                auth_method='basic',236                                message_encryption='invalid_value'237                                )238        self.assertEqual("invalid message_encryption arg: invalid_value. Should be 'auto', 'always', or 'never'", str(exc.exception))239    @mock.patch('requests.Session')240    def test_close_session(self, mock_session):241        t_default = transport.Transport(endpoint="Endpoint",242                                        server_cert_validation='ignore',243                                        username='test',244                                        password='test',245                                        auth_method='basic',246                                        )247        t_default.build_session()248        t_default.close_session()249        mock_session.return_value.close.assert_called_once_with()250        self.assertIsNone(t_default.session)251    @mock.patch('requests.Session')252    def test_close_session_not_built(self, mock_session):253        t_default = transport.Transport(endpoint="Endpoint",254                                        server_cert_validation='ignore',255                                        username='test',256                                        password='test',257                                        auth_method='basic',258                                        )259        t_default.close_session()260        self.assertFalse(mock_session.return_value.close.called)...

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.