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

How to use raise_first method in localstack

Best Python code snippet using localstack_python

fixed_thread_pool_executor.py

Source:fixed_thread_pool_executor.py

...25                executor.submit(sum, value, value)26            executor.drain()27        except:28            executor.close()29        executor.raise_first()30        print executor.returns3132    You can also use it with the Python "with" keyword, in which case you don't need to call "close"33    explicitly::3435        with FixedThreadPoolExecutor(10) as executor:36            for value in range(100):37                executor.submit(sum, value, value)38            executor.drain()39            executor.raise_first()40            print executor.returns41    """4243    def __init__(self, size = multiprocessing.cpu_count() * 2 + 1, timeout = None, print_exceptions = False):44        """45        :param size: Number of threads in the pool (fixed).46        :param timeout: Timeout in seconds for all blocking operations. (Defaults to none, meaning no timeout)47        :param print_exceptions: Set to true in order to print exceptions from tasks. (Defaults to false)48        """49        self.size = size50        self.timeout = timeout51        self.print_exceptions = print_exceptions5253        self._tasks = Queue()54        self._returns = {}55        self._exceptions = {}56        self._id_creator = itertools.count()57        self._lock = Lock() # for console output5859        self._workers = [FixedThreadPoolExecutor._Worker(self, index) for index in range(size)]6061    def submit(self, fn, *args, **kwargs):62        """63        Submit a task for execution.6465        The task will be called ASAP on the next available worker thread in the pool.66        """67        self._tasks.put((next(self._id_creator), fn, args, kwargs), timeout=self.timeout)6869    def close(self):70        """71        Blocks until all current tasks finish execution and all worker threads are dead.7273        You cannot submit tasks anymore after calling this.7475        This is called automatically upon exit if you are using the "with" keyword.76        """77        self.drain()78        while self.is_alive:79            self._tasks.put(FixedThreadPoolExecutor._CYANIDE, timeout=self.timeout)80        self._workers =  None8182    def drain(self):83        """84        Blocks until all current tasks finish execution, but leaves the worker threads alive.85        """86        self._tasks.join() # oddly, the API does not support a timeout parameter8788    @property89    def is_alive(self):90        """91        True if any of the worker threads are alive.92        """93        for worker in self._workers:94            if worker.is_alive():95                return True96        return False9798    @property99    def returns(self):100        """101        The returned values from all tasks, in order of submission.102        """103        return [self._returns[k] for k in sorted(self._returns)]104105    @property106    def exceptions(self):107        """108        The raised exceptions from all tasks, in order of submission.109        """110        return [self._exceptions[k] for k in sorted(self._exceptions)]111112    def raise_first(self):113        """114        If exceptions were thrown by any task, then the first one will be raised.115116        This is rather arbitrary: proper handling would involve iterating all the117        exceptions. However, if you want to use the "raise" mechanism, you are118        limited to raising only one of them.119        """120        exceptions = self.exceptions121        if exceptions:122            raise exceptions[0]123124    _CYANIDE = object()125    """126    Special task marker used to kill worker threads.
...

test_parser_validate.py

Source:test_parser_validate.py

...20                body='{"ARN":"somearn"}',21            )22        )23        with pytest.raises(MissingRequiredField) as e:24            validate_request(op, params).raise_first()25        assert e.value.error.reason == "missing required field"26        assert e.value.required_name == "TagList"27    def test_missing_required_field_query(self):28        parser = create_parser(load_service("sqs"))29        op, params = parser.parse(30            HttpRequest(31                "POST",32                "/",33                body=(34                    "Action=SendMessage&Version=2012-11-05&"35                    "QueueUrl=http%3A%2F%2Flocalhost%3A4566%2F000000000000%2Ftf-acc-test-queue&"36                ),37                headers={"Content-Type": "application/x-www-form-urlencoded"},38            )39        )40        validator = ParamValidator()41        errors = validator.validate(params, op.input_shape)42        assert errors.has_errors()43        with pytest.raises(MissingRequiredField) as e:44            errors.raise_first()45        assert e.match("MessageBody")46        assert e.value.error.reason == "missing required field"47        assert e.value.required_name == "MessageBody"48    def test_missing_required_field_restxml(self):49        parser = create_parser(load_service("route53"))50        op, params = parser.parse(51            HttpRequest(52                "POST",53                "/2013-04-01/hostedzone",54                body="<CreateHostedZoneRequest><Name>foobar.com</Name></CreateHostedZoneRequest>",55            )56        )57        with pytest.raises(MissingRequiredField) as e:58            validate_request(op, params).raise_first()59        assert e.value.error.reason == "missing required field"60        assert e.value.required_name == "CallerReference"61    def test_invalid_range_query(self):62        parser = create_parser(load_service("sts"))63        op, params = parser.parse(64            HttpRequest(65                "POST",66                "/",67                body=urlencode(68                    query={69                        "Action": "AssumeRole",70                        "RoleArn": "arn:aws:iam::000000000000:role/foobared",71                        "RoleSessionName": "foobared",72                        "DurationSeconds": "100",73                    }74                ),75                headers={"Content-Type": "application/x-www-form-urlencoded"},76            )77        )78        with pytest.raises(InvalidRange) as e:79            validate_request(op, params).raise_first()80        e.match("DurationSeconds")81    def test_invalid_length_query(self):82        parser = create_parser(load_service("sts"))83        op, params = parser.parse(84            HttpRequest(85                "POST",86                "/",87                body=urlencode(88                    query={89                        "Action": "AssumeRole",90                        "RoleArn": "arn:aws",  # min=891                        "RoleSessionName": "foobared",92                    }93                ),94                headers={"Content-Type": "application/x-www-form-urlencoded"},95            )96        )97        with pytest.raises(InvalidLength) as e:98            validate_request(op, params).raise_first()...

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.