Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use funcobj method in pytest-play

Best Python code snippet using pytest-play_python

blueprint.py

Source:blueprint.py

...57gRPCMethodsType = List[AbstractRpcMethod]58class UnaryUnaryRpcMethod(AbstractRpcMethod):59    if TYPE_CHECKING:  # pragma: no cover60        @staticmethod61        def funcobj(bp: "Blueprint", request: Message, contest: Context) -> Message:62            pass63    def to_rpc_method(self) -> str:64        return f"rpc {self.name} ({self.request_cls.__name__}) returns ({self.response_cls.__name__}) {{}}"65    def handle_call(66        self, bp: "Blueprint", message: GeneratedProtocolMessageType, context: Context67    ) -> GeneratedProtocolMessageType:68        current_request = self.request_cls()69        current_request.init_grpc_message(grpc_message=message)70        with bp.current_app.app_context(bp, self.funcobj, context):71            # TODO: using cygrpc.install_context_from_request_call_event to prepare context72            try:73                current_request = bp.current_app.process_request(74                    current_request, context75                )76                current_request = bp.before_request(current_request, context)77                response = self.funcobj(bp, current_request, context)78                bp_response = bp.after_request(response, context)79                app_response = bp.current_app.process_response(bp_response, context)80                return app_response.__message__81            except Exception as e:82                response = bp.current_app.handle_exception(e, context)83                if response:84                    return response.__message__85class UnaryStreamRpcMethod(AbstractRpcMethod):86    if TYPE_CHECKING:  # pragma: no cover87        @staticmethod88        def funcobj(89            bp: "Blueprint", request: Message, contest: Context90        ) -> Iterator[Message]:91            pass92    def to_rpc_method(self) -> str:93        return f"rpc {self.name} ({self.request_cls.__name__}) returns (stream {self.response_cls.__name__}) {{}}"94    def handle_call(95        self, bp: "Blueprint", message: GeneratedProtocolMessageType, context: Context96    ) -> Iterable[GeneratedProtocolMessageType]:97        current_request = self.request_cls()98        current_request.init_grpc_message(grpc_message=message)99        with bp.current_app.app_context(bp, self.funcobj, context):100            # TODO: using cygrpc.install_context_from_request_call_event to prepare context101            try:102                current_request = bp.current_app.process_request(103                    current_request, context104                )105                current_request = bp.before_request(current_request, context)106                for response in self.funcobj(bp, current_request, context):107                    yield response.__message__108            except Exception as e:109                yield from map(110                    attrgetter("__message__"),111                    bp.current_app.handle_exception(e, context),112                )113class StreamUnaryRpcMethod(AbstractRpcMethod):114    if TYPE_CHECKING:  # pragma: no cover115        @staticmethod116        def funcobj(117            bp: "Blueprint", request: Iterator[Message], contest: Context118        ) -> Message:119            pass120    def to_rpc_method(self) -> str:121        return f"rpc {self.name} (stream {self.request_cls.__name__}) returns ({self.response_cls.__name__}) {{}}"122    def handle_call(123        self,124        bp: "Blueprint",125        message: Iterable[GeneratedProtocolMessageType],126        context: Context,127    ) -> GeneratedProtocolMessageType:128        request_iterator = self.request_iterator(message)129        with bp.current_app.app_context(bp, self.funcobj, context):130            # TODO: using cygrpc.install_context_from_request_call_event to prepare context131            try:132                response = self.funcobj(bp, request_iterator, context)133                bp_response = bp.after_request(response, context)134                app_response = bp.current_app.process_response(bp_response, context)135                return app_response.__message__136            except Exception as e:137                response = bp.current_app.handle_exception(e, context)138                if response:139                    return response.__message__140class StreamStreamRpcMethod(AbstractRpcMethod):141    if TYPE_CHECKING:  # pragma: no cover142        @staticmethod143        def funcobj(144            bp: "Blueprint", request: Iterator[Message], contest: Context145        ) -> Iterator[Message]:146            pass147    def to_rpc_method(self) -> str:148        return f"rpc {self.name} (stream {self.request_cls.__name__}) returns (stream {self.response_cls.__name__}) {{}}"149    def handle_call(150        self,151        bp: "Blueprint",152        message: Iterable[GeneratedProtocolMessageType],153        context: Context,154    ) -> Iterable[GeneratedProtocolMessageType]:155        request_iterator = self.request_iterator(message)156        with bp.current_app.app_context(bp, self.funcobj, context):157            # TODO: using cygrpc.install_context_from_request_call_event to prepare context158            try:159                for response in self.funcobj(bp, request_iterator, context):160                    yield response.__message__161            except Exception as e:162                yield from map(163                    attrgetter("__message__"),164                    bp.current_app.handle_exception(e, context),165                )166RequestType = TypeVar("RequestType", bound=Message)167ResponseType = TypeVar("ResponseType", bound=Message)168class InvalidRPCMethod(Exception):169    pass170class Blueprint:171    """gRPCAlchemy uses a concept of blueprints for making gRPC services and172    supporting common patterns within an application or across applications.173    Blueprints can greatly simplify how large applications work. A Blueprint object...

pytest_.py

Source:pytest_.py

...18            if event == 'return':19                funclocals.update(frame.f_locals.copy())20    sys.setprofile(_tracefunc)21    try:22        funcobj(*args, **kwargs)23    finally:24        sys.setprofile(None)25    return funclocals26def make_module_from_function(funcobj):27    """Evaluates the local scope of a function, as if it was a module"""28    module = imp.new_module(funcobj.__name__)29    scope = marks.get(funcobj, 'scope')30    funclocals = trace_function(funcobj, scope)31    module.__dict__.update(funclocals)32    return module33def copy_markinfo(module, funcobj):34    from _pytest.mark import MarkInfo35    marks = {}36    for name, val in funcobj.__dict__.items():...

ls_sgd_solver.py

Source:ls_sgd_solver.py

1import argparse2import numpy as np3class RosenBroke:4    def __init__(self, N) -> None:5        self.N = N6        self.x1 = np.zeros(N)7        self.x2 = np.zeros(N)8    9    def fx(self):10        z = 100 * np.square(np.square(self.x1) - self.x2) + np.square(self.x1 - 1)11        return z.sum()12    def get_x(self):13        return np.stack((self.x1, self.x2), axis=1).reshape(2*self.N)14    def get_gradiant_T(self):15        odd_vec = 100 * (4 * np.power(self.x1, 3) - 4 * self.x2 * self.x1) + 2 * (self.x1 - 1)16        even_vec = 200 * self.x2 - 200 * np.square(self.x1)17        return np.stack((odd_vec, even_vec), axis=1).reshape(2*self.N)18    def get_hessian(self):19        A = 100 * (12 * np.square(self.x1) - 4 * self.x2) + 220        B = 100 * (-4 * self.x1)21        C = -4 * self.x122        D = 2 * np.ones(self.N)23        H = np.zeros((2 * self.N, 2 * self.N))24        for i in range(self.N):25            H[2*i, 2*i] = A[i]26            H[2*i + 1, 2*i + 1] = D[i]27            H[2*i, 2*i + 1] = B[i]28            H[2*i + 1, 2*i] = C[i]29        return H30    def set_x(self, x):31        self.x1, self.x2 = x.reshape(self.N, 2).transpose()32        return self33class LinearSearchSGDSolver:34    def __init__(self, funcObj, c=0.5, max_itr=1000) -> None:35        self.funcObj = funcObj36        self.max_itr = max_itr37        self.c = c38    def solve(self):39        for _ in range(self.max_itr):40            tau = 141            G_T = self.funcObj.get_gradiant_T()42            # print("G_T: ", G_T)43            d_T = - self._norm(G_T)44            # d_T = - G_T45            x = self.funcObj.get_x()46            # print("x: ", x)47            fx = self.funcObj.fx()48            # print("fx: ", fx)49            while tau > 1e-6:50                x_next = x + tau * d_T51                # print("x_next: ", x_next)52                fx_next = self.funcObj.set_x(x_next).fx()53                # print("fx_next: ", fx_next)54                impro_factor = - self.c * tau * np.dot(d_T, G_T)55                # print("if: ", impro_factor)56                if fx - fx_next >= impro_factor:57                    break58                tau = tau / 259            print("===> ", x)60            if impro_factor < 1e-16:61                break62        return self.funcObj.get_x()63    def _norm(self, x):64        return x / np.sqrt(np.sum(x**2))65def main(args):66    funcObj = RosenBroke(args.N)67    # print(funcObj.get_x())68    # print(funcObj.get_gradiant())69    # print(funcObj.get_hessian())70    solver = LinearSearchSGDSolver(funcObj, max_itr=10000)71    result = solver.solve()72if __name__ == "__main__":73    parser = argparse.ArgumentParser()74    parser.add_argument('-N', type=int, default=1)75    args = parser.parse_args()...

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.