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

How to use _is_executed method in localstack

Best Python code snippet using localstack_python

Interpreter.py

Source:Interpreter.py

1class InfiniteLoopException(Exception):2    pass3class Instruction:4    def __init__(self, operation_map, operation, argument):5        self._operation_map = operation_map6        self._operation = operation7        self._argument = argument8        self._is_executed = False9    def execute(self):10        # print("Executing", self._operation, self._argument)11        if self._is_executed:12            raise InfiniteLoopException13        self._operation_map[self._operation](self._argument)14        self._is_executed = True15    def get_operation(self):16        return self._operation17    def set_operation(self, new_operation):18        self._operation = new_operation19    def get_argument(self):20        return self._argument21    def set_argument(self, new_argument):22        self._argument = new_argument23class Interpreter:24    def __init__(self, instructions):25        self.operation_map = {26            "nop": self._nop,27            "jmp": self._jmp,28            "acc": self._acc29        }30        self._instructions = []31        for instruction in instructions:32            operation, argument = instruction.split(" ")33            self._instructions.append(Instruction(self.operation_map, operation, int(argument)))34        self._instructionPointer = 035        self._accumulator = 036    def reset(self):37        self._instructionPointer = 038        self._accumulator = 039        for instruction in self._instructions:40            instruction._is_executed = False41    def run(self):42        self.reset()43        while self._instructionPointer < len(self._instructions):44            self._instructions[self._instructionPointer].execute()45    def get_accumulator(self):46        return self._accumulator47    def get_instruction_set(self):48        return self._instructions49    def _nop(self, argument):50        self._instructionPointer += 151        pass52    def _jmp(self, argument):53        self._instructionPointer += argument54    def _acc(self, argument):55        self._accumulator += argument56        self._instructionPointer += 157    def print(self):58        print("----------- interpreter ----------")59        print("Accumulator:", self._accumulator)60        print("Instruction pointer:", self._instructionPointer)61        print("Instructions:")62        for instruction in self._instructions:...

node.py

Source:node.py

1from __future__ import annotations2from typing import Any, Callable, Dict, List, Optional3import threading4from tinydag.exceptions import NodeUnexecutedError5class Node:6    def __init__(7        self,8        name: str,9        func: Callable,10        depends: Optional[List[str]] = None,11    ) -> None:12        self._name = name13        self._depends = depends or []14        self._func = func15        self._next_nodes: List[Node] = []16        self._inputs: Dict[str, Any] = {}17        self._output: Optional[Any] = None18        self._is_executed = False19        self._lock = threading.RLock()20    @property21    def name(self) -> str:22        return self._name23    @property24    def depends(self) -> List[str]:25        return self._depends26    @property27    def next_nodes(self) -> List[Node]:28        return self._next_nodes29    @property30    def is_ready(self) -> bool:31        with self._lock:32            return len(self._depends) == len(self._inputs)33    @property34    def is_executed(self) -> bool:35        return self._is_executed36    @property37    def output(self) -> Any:38        if not self.is_executed:39            raise NodeUnexecutedError(f"Node <{self.name}> is not executed.")40        return self._output41    def add_next_node(self, next_node: Node) -> None:42        self._next_nodes.append(next_node)43    def set_input(self, depname: str, data: Any):44        with self._lock:45            self._inputs[depname] = data46    def run(self) -> Any:47        inputs: List[Any] = [48            self._inputs[depname] for depname in self._depends49            if depname in self._inputs50        ]51        if len(self._inputs) != len(self._depends):52            raise RuntimeError("Numbers of required and given "53                               "inputs is not matched.")54        self._output = self._func(*inputs)55        self._is_executed = True56        return self._output57    def __repr__(self) -> str:...

computation.py

Source:computation.py

1"""Contains code related to computations."""2from __future__ import annotations3import textwrap4import warnings5from dataclasses import dataclass6from typing import Callable, NewType7from .environment import Environment8from .record import Record9Identifier = NewType("Identifier", str)10class Computation:11    """Represents a computation."""12    def __init__(13        self,14        identifier: Identifier,15        environment: Environment,16        trigger: Callable[[], None],17    ) -> None:18        """Initialize the computation."""19        self.identifier = identifier20        self._environment = environment21        self._trigger = trigger22        self._is_executed = False23    def execute(self) -> ComputationRecord:24        """Execute the computation."""25        if self._is_executed:26            raise RuntimeError("Computation already executed!")27        with self._environment.consistency_check() as check:28            self._trigger()29        if not check.success:30            warnings.warn("Environment changed during execution!")31        self._is_executed = True32        return ComputationRecord(self.identifier, check.record)33    def __repr__(self) -> str:34        """Return a string representation of the computation."""35        return (36            f"{self.__class__.__name__}("37            f"identifier={repr(self.identifier)}, "38            f"environment={repr(self._environment)}, "39            f"trigger={repr(self._trigger)})"40        )41@dataclass(frozen=True)42class ComputationRecord:43    """Represents the association between an executed computation and its environmental record."""44    identifier: Identifier45    record: Record46    def __str__(self) -> str:47        """Return a human-readable representation of the computation record."""48        indent = 4 * " "49        string = "Computation Record:\n"50        string += textwrap.indent("Identifier: " + self.identifier + "\n" + str(self.record), indent)...

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.