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

How to use save_status method in lettuce-tools

Best Python code snippet using lettuce-tools_python

consumers.py

Source:consumers.py

...13        await self.channel_layer.group_add(14            self.room_group_name,15            self.channel_name16        )17        await self.save_status(self.user_name,"connected")18        await self.accept()19    async def disconnect(self, close_code):20        # Leave room group21        await self.channel_layer.group_discard(22            self.room_group_name,23            self.channel_name24        )25        await self.save_status(self.user_name,"disconnected")26    # Receive message from WebSocket27    async def receive(self, text_data):28        text_data_json = json.loads(text_data)29        message = text_data_json['message']30        time=await self.save_message(self.user_name,self.room_name,message)31        time=str(time)32        # Send message to room group33        await self.channel_layer.group_send(34            self.room_group_name,35            {36                'type': 'chat_message',37                'user_name':self.user_name,38                'message': message,39                'time':time,40            }41        )42    # Receive message from room group43    async def chat_message(self, event):44        username = event['user_name']45        message = event['message']46        time=event['time']47        # Send message to WebSocket48        await self.send(text_data=json.dumps({49            'user_name':username,50            'message': message,51            'time':time,52        }))53    54    @sync_to_async55    def save_message(self,username,room,message):56        mes=Messages.objects.create(username=username,room_id=room,message=message)57        return mes.datetime58    @sync_to_async59    def save_status(self,username,status):60        stat=OnlineOffline.objects.get(user=username)61        stat.status=status62        stat.save()63class PersonalConsumer(AsyncWebsocketConsumer):64    async def connect(self):65        self.me = self.scope['url_route']['kwargs']['me']66        self.you = self.scope['url_route']['kwargs']['you']67        chat_id=await self.get_id(self.me,self.you)68        self.room_group_name = 'personalchat_%d' % chat_id69        # Join room group70        await self.channel_layer.group_add(71            self.room_group_name,72            self.channel_name73        )74        await self.save_status(self.me,"connected")75        await self.accept()76    async def disconnect(self, close_code):77        self.me = self.scope['url_route']['kwargs']['me']78        # Leave room group79        await self.channel_layer.group_discard(80            self.room_group_name,81            self.channel_name82        )83        await self.save_status(self.me,"disconnected")84    # Receive message from WebSocket85    async def receive(self, text_data):86        text_data_json = json.loads(text_data)87        message = text_data_json['message']88        me = text_data_json['me']89        you = text_data_json['you']90        chat_id=await self.get_id(me,you)91        chat_time=await self.save_personal_message(chat_id,me,message)92        chat_time=str(chat_time)93        # Send message to room group94        await self.channel_layer.group_send(95            self.room_group_name,96            {97                'type': 'chat_message',98                'me':me,99                'you':you,100                'message': message,101                'time':chat_time,102            }103        )104    # Receive message from room group105    async def chat_message(self, event):106        print(event)107        me = event['me']108        you=event['you']109        message = event['message']110        time=event['time']111        # Send message to WebSocket112        await self.send(text_data=json.dumps({113            'me':me,114            'you':you,115            'message': message,116            'time':time,117        }))118    @sync_to_async119    def get_id(self,me,you):120        try:121            id=Personal.objects.get(me=me,user=you)122        except:123            id=Personal.objects.get(me=you,user=me)124        return(id.id)125    @sync_to_async126    def save_personal_message(self,person_id,me,messages):127        personal=PersonalMessage.objects.create(people_id=person_id,sender=me,messages=messages)128        return personal.datetime129    @sync_to_async130    def save_status(self,username,status):131        stat=OnlineOffline.objects.get(user=username)132        stat.status=status133        stat.save()134class LoginStatus(AsyncWebsocketConsumer):135    async def connect(self):136        self.username = self.scope['url_route']['kwargs']['username']137        self.room_group_name = 'connected'138        # Join room group139        await self.channel_layer.group_add(140            self.room_group_name,141            self.channel_name142        )143        await self.accept()144        await self.save_status(self.username,"connected")145        # Send message to room group146        await self.channel_layer.group_send(147            self.room_group_name,148            {149                'type': 'chat_message_disconnect',150                'username':self.username,151                'status':"connected"152            }153        )154    async def disconnect(self, close_code):155        self.username = self.scope['url_route']['kwargs']['username']156        # Leave room group157        await self.channel_layer.group_discard(158            self.room_group_name,159            self.channel_name160        )161        await self.save_status(self.username,"disconnected")162        # Send message to room group163        await self.channel_layer.group_send(164            self.room_group_name,165            {166                'type': 'chat_message_disconnect',167            }168        )169    async def chat_message_disconnect(self, event):170        stapeople=await self.show_status(self.username)171        # Send message to WebSocket172        await self.send(text_data=stapeople)173    174    # Receive message from room group175    async def chat_message(self, event):176        # Send message to WebSocket177        await self.send(text_data="hello")178    @sync_to_async179    def save_status(self,username,status):180        stat=OnlineOffline.objects.get(user=username)181        stat.status=status182        stat.save()183    @sync_to_async184    def show_status(self,user):185        stapeople=OnlineOffline.objects.all()186        stapeople1=serializers.serialize('json', stapeople)...

emop_payload.py

Source:emop_payload.py

1import json2import logging3import os4from emop.lib.utilities import mkdirs_exists_ok5logger = logging.getLogger('emop')6class EmopPayload(object):7    def __init__(self, settings, proc_id):8        self.settings = settings9        self.input_path = self.settings.payload_input_path10        self.output_path = self.settings.payload_output_path11        self.completed_output_path = self.settings.payload_completed_path12        self.uploaded_output_path = self.settings.payload_uploaded_path13        self.proc_id = proc_id14        self.input_filename = os.path.join(self.input_path, "%s.json" % self.proc_id)15        self.output_filename = os.path.join(self.output_path, "%s.json" % self.proc_id)16        self.completed_output_filename = os.path.join(self.completed_output_path, "%s.json" % self.proc_id)17        self.uploaded_output_filename = os.path.join(self.uploaded_output_path, "%s.json" % self.proc_id)18    def file_exists(self, filename):19        logger.debug("Checking for payload at %s", filename)20        if os.path.isfile(filename):21            return True22        else:23            return False24    def input_exists(self):25        return self.file_exists(self.input_filename)26    def output_exists(self):27        return self.file_exists(self.output_filename)28    def completed_output_exists(self):29        return self.file_exists(self.completed_output_filename)30    def uploaded_output_exists(self):31        return self.file_exists(self.uploaded_output_filename)32    def save(self, data, dirname, filename, overwrite=False):33        if not os.path.isdir(dirname):34            logger.debug("Creating payload directory %s" % dirname)35            mkdirs_exists_ok(dirname)36        if not overwrite and os.path.exists(filename):37            logger.error("payload file %s already exists" % filename)38            return None39        if overwrite:40            logger.debug("Overwriting payload file at %s" % filename)41        else:42            logger.debug("Saving payload to %s" % filename)43        with open(filename, 'w') as outfile:44            json.dump(data, outfile)45        return True46    def load(self, filename):47        if not os.path.isfile(filename):48            logger.error("payload file %s does not exist" % filename)49            return None50        logger.debug("Loading payload from %s" % filename)51        with open(filename) as datafile:52            try:53                data = json.load(datafile)54            except ValueError:55                logger.error("EmopPayload: Invalid JSON file %s" % filename)56                return None57        return data58    def save_input(self, data):59        dirname = self.input_path60        filename = self.input_filename61        save_status = self.save(data=data, dirname=dirname, filename=filename, overwrite=False)62        return save_status63    def save_output(self, data, overwrite=False):64        dirname = self.output_path65        filename = self.output_filename66        save_status = self.save(data=data, dirname=dirname, filename=filename, overwrite=overwrite)67        return save_status68    def save_completed_output(self, data, overwrite=False):69        dirname = self.completed_output_path70        filename = self.completed_output_filename71        save_status = self.save(data=data, dirname=dirname, filename=filename, overwrite=overwrite)72        if save_status and os.path.isfile(self.output_filename):73            logger.debug("Removing payload file %s" % self.output_filename)74            os.remove(self.output_filename)75        return save_status76    def save_uploaded_output(self, data):77        dirname = self.uploaded_output_path78        filename = self.uploaded_output_filename79        save_status = self.save(data=data, dirname=dirname, filename=filename, overwrite=True)80        if save_status:81            if self.completed_output_exists():82                logger.debug("Removing payload file %s" % self.completed_output_filename)83                os.remove(self.completed_output_filename)84            elif self.output_exists():85                logger.debug("Removing payload file %s" % self.output_filename)86                os.remove(self.output_filename)87        return save_status88    def load_input(self):89        filename = self.input_filename90        data = self.load(filename=filename)91        # TODO Need to move or remove input payloads as there will be many after some time...

strat_manual_trade.py

Source:strat_manual_trade.py

1#-*- coding:utf-8 -*-2from base import *3from misc import *4import logging5from strategy import *6class ManualTrade(Strategy):7    common_params =  dict({'daily_close_buffer': 3, 'price_limit_buffer': 5}, \8                          **Strategy.common_params)9    asset_params = Strategy.asset_params.copy()10    asset_params.update({'long_price': 0.0, 'long_stop': 0.0, 'short_price': 0.0, 'short_stop': 0.0, \11                         'tick_num': 1, 'order_offset': True, 'run_flag': 0, \12                         'max_pos': 1, 'max_vol': 10, 'time_period': 600, 'price_type': OPT_LIMIT_ORDER, \13                         'exec_args': {'max_vol': 10, 'time_period': 600, 'price_type': OPT_LIMIT_ORDER, \14                                  'tick_num': 1, 'order_type': '', 'order_offset': True, 'inst_order': None},})15    def __init__(self, config, agent = None):16        Strategy.__init__(self, config, agent)17        numAssets = len(self.underliers)18        self.tick_base = [0.0] * numAssets19    def set_exec_args(self, idx, direction):20        for key in ['max_vol', 'time_period', 'price_type', 'tick_num', 'order_offset']:21            self.exec_args[idx][key] = getattr(self, key)[idx]22    def open_long(self, idx):23        if len(self.positions[idx]) < self.max_pos[idx]:24            self.set_exec_args(idx, ORDER_BUY)25            self.open_tradepos(idx, 1, self.long_price[idx], int(self.trade_unit[idx]))26            return True27        else:28            return False29    def open_short(self, idx):30        if len(self.positions[idx]) < self.max_pos[idx]:31            self.set_exec_args(idx, ORDER_SELL)32            self.open_tradepos(idx, -1, self.short_price[idx], int(self.trade_unit[idx]))33            return True34        else:35            return False36    def on_tick(self, idx, ctick):37        num_pos = len(self.positions[idx])38        curr_pos = self.curr_pos[idx]39        save_status = False40        if ((self.curr_pos[idx] <= 0) and (self.curr_prices[idx] >= self.long_price[idx])) or \41                ((self.curr_pos[idx] >= 0) and (self.curr_prices[idx] <= self.short_price[idx])):42            save_status = self.liquidate_tradepos(idx) or save_status43            num_pos = 044            curr_pos = 045        if (self.curr_prices[idx] >= self.long_price[idx]):46            save_status = self.open_long(idx) or save_status47        elif (self.curr_prices[idx] <= self.short_price[idx]):48            save_status = self.open_short(idx) or save_status...

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.