How to use pid_is_alive method in autotest

Best Python code snippet using autotest_python

views.py

Source:views.py

1import os2import json3import signal4from subprocess import Popen, PIPE5from django.http import HttpResponse6from django.core.serializers.json import DjangoJSONEncoder7from django.forms.models import model_to_dict8from django.core.exceptions import ObjectDoesNotExist9from management_app.models import Trade, Orders_History, Pair_Timeframe, Candle10from server.settings import BASE_DIR, VIRTUALENV_DIR11from trader.bitfinex_client import post_balances12def check_pid(pid):13    """ Check For the existence of a unix pid. """14    try:15        os.kill(pid, 0)16    except OSError:17        return False18    else:19        return True20def settings():21    work_dir = os.path.join(BASE_DIR, "trader")22    log_dir = os.path.join(BASE_DIR, "logs", "log")23    misc_dir = os.path.join(BASE_DIR, "logs", ".misc")24    if not os.path.exists(log_dir):25        os.makedirs(log_dir)26    if not os.path.exists(misc_dir):27        os.makedirs(misc_dir)28    return work_dir, log_dir, misc_dir29def json_response(data):30    return HttpResponse(json.dumps(data, cls=DjangoJSONEncoder, sort_keys=True, indent=5, separators=(',', ': ')), content_type='application/json', status=200)31def run_command_in_virtualenv(command):32    commands = """33    . %s/bin/activate34    %s35    deactivate36    """ % (VIRTUALENV_DIR, command)37    process = Popen(commands, stdout=PIPE, shell=True)38    return process.communicate()39def file_read_last_lines(fname, lines):40    s = ""41    with open(fname) as f:42        x = f.readlines()43        s += ''.join(x[-lines:])44    return s45def collector_misc(action):46    work_dir, log_dir, misc_dir = settings()47    log_file = os.path.join(log_dir, "collect_data.log")48    pid_file = os.path.join(log_dir, "collect_data.pid")49    collector_py = os.path.join(work_dir, "collect_data.py")50    if (action == "start") or (action == "stop"):51        # Clear52        if os.path.exists(pid_file):53            with open(pid_file) as f:54                pids = f.read()55            try:56                os.kill(int(pids), signal.SIGTERM) #or signal.SIGKILL57            except ProcessLookupError:58                pass59            os.remove(pid_file)60        if os.path.exists(log_file):61            os.remove(log_file)62        if action == "start":63            command = "nohup python3 " + collector_py + " > /dev/null & echo $! >> " + pid_file64            run_command_in_virtualenv(command)65        return dict()66    if action == "status":67        if os.path.exists(pid_file):68            with open(pid_file) as f:69                pids = f.read()70            pid_is_alive = check_pid(int(pids))71        else:72            pid_is_alive = False73        if os.path.exists(log_file):74            with open(log_file) as f:75                log_text = f.read()76        else:77            log_text = "No log file"78        data = {79            "is_alive": pid_is_alive,80            "log": str(log_text)81        }82        return data83def collector(request):84    """85    .. http:post:: /management/collector/86        Popen(87            ["nohup", 88                    "python3", "/root/crypto_trading/bitfinex/collect_data.py", 89                    ">", "/root/crypto_trading/bitfinex/log/collect_data.out", "&", 90                    "echo", "$!", ">>", "/root/crypto_trading/bitfinex/.misc/collect_data.pid"]91        )92    **Request**:93    .. sourcecode:: json94        {95          "action": "start",  # "stop", "status"96        }97    **Response**:98    .. sourcecode:: json99        {100            "log": "kokokokoko"101        }102        or103        {}104    """105    received_json_data = json.loads(request.body.decode())106    r = collector_misc(action=received_json_data["action"])107    return json_response(r)108def crypto_trader_bot(request):109    """110    .. http:post:: /management/crypto_trader_bot/111        Start/Stop Crypto tradering bot112                or113        Status of alive Crypto tradering bot114    **Request**:115    .. sourcecode:: json116        {117          "action": "status", # "stop", "short_status"118          "pair": "XMRUSD"119        }120        or121        {122          "action": "start",123                    "pair": "XMRUSD",124                    "timeframe": "1m",125                    "timeperiod": 9,126                    "init_last_sell_price": 345.3, 127                    "init_last_buy_price": 345.3,128                    "DIV_PARTS": 2, 129                    "LPP_count": 1, 130                    "EP_gradient_threshold": 0.0, 131                    "LPP_gradients_threshold": 0.0, 132                    "sell_threshold": 0.8,133                    "buy_threshold": 0.8,134                    "independent_last_sell_buy_price_checking": false,135                    "forse_commit_sell_buy_status_distace": none136        }137    **Response**:138    .. sourcecode:: json139        140        {141            "log": "kokokokoko"142        }143        or144        {}145    """146    received_json_data = json.loads(request.body.decode())147    work_dir, log_dir, misc_dir = settings()148    log_file = os.path.join(log_dir, "crypto_trader_bot_%s.log" % received_json_data["pair"])149    pid_file = os.path.join(log_dir, "trader_%s.pid" % received_json_data["pair"])150    trader_py = os.path.join(work_dir, "crypto_trader_bot.py")151    if (received_json_data["action"] == "start") or (received_json_data["action"] == "stop"):152        # Clear153        if os.path.exists(pid_file):154            with open(pid_file) as f:155                pids = f.read()156            try:157                os.kill(int(pids), signal.SIGTERM) #or signal.SIGKILL158            except ProcessLookupError:159                pass160            os.remove(pid_file)161        if received_json_data["action"] == "start":162            # if os.path.exists(out_file):163            #     os.remove(out_file)164            if os.path.exists(log_file):165                os.remove(log_file)166        if received_json_data["action"] == "start":167            command = "nohup python3 " + trader_py + \168                  " " + str(received_json_data["pair"]) + \169                  " " + str(received_json_data["timeframe"]) + \170                  " " + str(received_json_data["timeperiod"]) + \171                  " " + str(received_json_data["independent_last_sell_buy_price_checking"]) + \172                  " " + str(received_json_data["forse_commit_sell_buy_status_distace"]) + \173                  " " + str(received_json_data["LPP_count"]) + \174                  " " + str(received_json_data["EP_gradient_threshold"]) + \175                  " " + str(received_json_data["LPP_gradients_threshold"]) + \176                  " " + str(received_json_data["sell_threshold"]) + \177                  " " + str(received_json_data["buy_threshold"]) + \178                  " " + str(received_json_data["DIV_PARTS"]) + \179                  " " + str(received_json_data["init_last_sell_price"]) + \180                  " " + str(received_json_data["init_last_buy_price"]) + \181                      " > /dev/null & echo $! >> " + pid_file182            run_command_in_virtualenv(command)183    if received_json_data["action"] == "status":184        if os.path.exists(pid_file):185            with open(pid_file) as f:186                pids = f.read()187            pid_is_alive = check_pid(int(pids))188        else:189            pid_is_alive = False190        # if os.path.exists(out_file):191        #     with open(out_file) as f:192        #         out_file_text = f.read()193        # else:194        #     out_file_text = "No out file"195        if os.path.exists(log_file):196            with open(log_file) as f:197                log_file_text = f.read()198        else:199            log_file_text = "No log file"200        data = {201            "is_alive": pid_is_alive,202            # "out_file": str(out_file_text),203            "log_file": str(log_file_text)204        }205        return json_response(data)206    if received_json_data["action"] == "short_status":207        if os.path.exists(pid_file):208            with open(pid_file) as f:209                pids = f.read()210            pid_is_alive = check_pid(int(pids))211        else:212            pid_is_alive = False213        # if os.path.exists(out_file):214        #     with open(out_file) as f:215        #         out_file_text = f.read()216        # else:217        #     out_file_text = "No out file"218        if os.path.exists(log_file):219            log_file_text = file_read_last_lines(log_file, 228)220        else:221            log_file_text = "No log file"222        data = {223            "is_alive": pid_is_alive,224            # "out_file": str(out_file_text),225            "log_file": str(log_file_text)226        }227        return json_response(data)228    return json_response(dict())229def orders_history(request):230    """231    .. http:post:: /management/orders_history/232    **Request**:233    .. sourcecode:: json234        {235          "trade_id": 45,236          "action": "get_all" #, "delete", "get"237        }238    **Response**:239    .. sourcecode:: json240        [241           {}, {}, {}, ...242        ]243    """244    received_json_data = json.loads(request.body.decode())245    trade_id = received_json_data["trade_id"]246    action = received_json_data["action"]247    if action == "get":248        orders_history = Orders_History.objects.filter(trade=trade_id)249        data = list(orders_history.values())250        return json_response(data)251    elif action == "delete":252        Orders_History.objects.filter(trade=trade_id).delete()253        data = list()254        return json_response(data)255    elif action == "get_all":256        orders_history = Orders_History.objects.all()257        data = list(orders_history.values())258        return json_response(data)259def trade(request):260    """261    .. http:post:: /management/trade/262    Delete/Get/Get All263    **Request**:264    .. sourcecode:: json265        {266          "id": 56,267          "action": "get_all" #, "delete", "get"268        }269    **Response**:270    .. sourcecode:: json271        [272           {}, {}, {}, ...273        ]274    """275    received_json_data = json.loads(request.body.decode())276    trade_id = received_json_data["id"]277    action = received_json_data["action"]278    if action == "get":279        trade = Trade.objects.filter(id=trade_id)280        data = list(trade.values())281        return json_response(data)282    elif action == "delete":283        Trade.objects.filter(id=trade_id).delete()284        data = list()285        return json_response(data)286    elif action == "get_all":287        trades = Trade.objects.all()288        all_data = []289        for trade in trades:290            data = model_to_dict(trade)291            BTC_in_the_start = float(trade.initial_BTC_wallet)292            USD_in_the_start = float(trade.initial_USD_wallet)293            pair = trade.pair_timeframe.pair294            timeframe = trade.pair_timeframe.timeframe295            try:296                orders_history_count = Orders_History.objects.filter(trade=trade).count()297                orders_history_last = Orders_History.objects.filter(trade=trade).latest('mts')298                BTC_in_the_end = sum(orders_history_last.misc['BTC'])299                USD_in_the_end = sum(orders_history_last.misc['USD'])300                data.update({301                    "orders_history_count": orders_history_count,302                    "profit_BTC": str(BTC_in_the_start) + " --> " + str(BTC_in_the_end),303                    "profit_USD": str(USD_in_the_start) + " --> " + str(USD_in_the_end),304                    "pair": pair,305                    "timeframe": timeframe306                })307            except ObjectDoesNotExist:308                data.update({309                    "orders_history_count": 0,310                    "profit_BTC": str(BTC_in_the_start) + " --> N/A",311                    "profit_USD": str(USD_in_the_start) + " --> N/A",312                    "pair": pair,313                    "timeframe": timeframe314                })315            all_data.append(data)316        return json_response(all_data)317def pairs(request):318    """319    .. http:post:: /management/pairs/320        Create/Delete/Get/Get All321    **Request**:322    .. sourcecode:: json323        {324          "pair": "XMRUSD",325          "timeframe": "1m",326          "action": "get_all" #, "delete", "get", "put"327        }328    **Response**:329    .. sourcecode:: json330        "put":331        [332           23333        ]334        "get":335        [336           23337        ]338        "get_all":339        [340             {341                  "id": 1,342                  "pair": "XMRUSD",343                  "timeframe": "1m"344             },345             {346                  "id": 4,347                  "pair": "XMRUSD",348                  "timeframe": "5m"349             },350             {351                  "id": 6,352                  "pair": "XMRUSD",353                  "timeframe": "15m"354             }355        ]356        "delete":357        []358    """359    received_json_data = json.loads(request.body.decode())360    pair = received_json_data["pair"]361    timeframe = received_json_data["timeframe"]362    action = received_json_data["action"]363    if action == "get":364        pair_timeframe = Pair_Timeframe.objects.get(pair=pair, timeframe=timeframe)365        data = [pair_timeframe.id]366        return json_response(data)367    elif action == "put":368        pair_timeframe = Pair_Timeframe(pair=pair, timeframe=timeframe)369        pair_timeframe.save()370        data = [pair_timeframe.id]371        return json_response(data)372    elif action == "delete":373        Pair_Timeframe.objects.filter(pair=pair, timeframe=timeframe).delete()374        data = list()375        return json_response(data)376    elif action == "get_all":377        pair_timeframe = Pair_Timeframe.objects.all()378        if pair_timeframe.count() == 0:379            return json_response([])380        else:381            data = list(pair_timeframe.values())382            return json_response(data)383def candles(request):384    """385    .. http:post:: /management/candles/386        Create/Delete/Get/Get All/Delete All/Count387    **Request**:388    .. sourcecode:: json389        {390          "pair": "XMRUSD",391          "timeframe": "1m",392          "action": "get" #, "delete", "delete_all", "count"393        }394    **Response**:395    .. sourcecode:: json396        "get":397        [398             {399                  "close": "272.4500000000000000000000000000000000000000",400                  "high": "274.3400000000000000000000000000000000000000",401                  "id": 419,402                  "low": "272.4500000000000000000000000000000000000000",403                  "mts": "2018-03-08T22:50:00Z",404                  "open": "274.3300000000000000000000000000000000000000",405                  "pair_timeframe_id": 4,406                  "volume": "447.7072914500000000000000000000000000000000"407             },408             {409                  "close": "274.2100000000000000000000000000000000000000",410                  "high": "276.6500000000000000000000000000000000000000",411                  "id": 420,412                  "low": "273.1700000000000000000000000000000000000000",413                  "mts": "2018-03-08T22:55:00Z",414                  "open": "275.2600000000000000000000000000000000000000",415                  "pair_timeframe_id": 4,416                  "volume": "1032.2220301600000000000000000000000000000000"417             },418             ...419         ]420        "delete":421        []422    """423    received_json_data = json.loads(request.body.decode())424    action = received_json_data["action"]425    if action != "delete_all":426        pair = received_json_data["pair"]427        timeframe = received_json_data["timeframe"]428    if action == "get":429        candles = Candle.objects.filter(pair_timeframe__pair=pair, pair_timeframe__timeframe=timeframe)430        data = list(candles.order_by('-mts').values())431        return json_response(data)432    elif action == "delete":433        Candle.objects.filter(pair_timeframe__pair=pair, pair_timeframe__timeframe=timeframe).delete()434        data = list()435        return json_response(data)436    elif action == "delete_all":437        Candle.objects.all().delete()438    elif action == "count":439        if (pair != "") and (timeframe != ""):440            pairs_timeframes = Pair_Timeframe.objects.filter(pair=pair, timeframe=timeframe).values()441        else:442            pairs_timeframes = Pair_Timeframe.objects.all().values()443        data = []444        for pair_timeframe in pairs_timeframes:445            pair = pair_timeframe["pair"]446            timeframe = pair_timeframe["timeframe"]447            pair_timeframe_id = pair_timeframe["id"]448            count = Candle.objects.filter(pair_timeframe_id=pair_timeframe_id).count()449            last = Candle.objects.filter(pair_timeframe_id=pair_timeframe_id).latest('mts')450            data.append({451                "pair": pair,452                "timeframe": timeframe,453                "count": count,454                "last": {455                    "id": last.id,456                    "mts": last.mts,457                    "open": float(last.open),458                    "close": float(last.close),459                    "low": float(last.low),460                    "high": float(last.high),461                    "volume": float(last.volume)462                }463            })464        data = sorted(data, key=lambda x: x["timeframe"])465        return json_response(data)466    return json_response([])467def balance(request):468    """469    .. http:post:: /management/balance/470    **Request**:471    .. sourcecode:: json472    **Response**:473    .. sourcecode:: json474            [{475              "type":"deposit",476              "currency":"btc",477              "amount":"0.0",478              "available":"0.0"479            },{480              "type":"deposit",481              "currency":"usd",482              "amount":"1.0",483              "available":"1.0"484            },485            ...]486    """487    r = post_balances()...

base_syncdata_unittest.py

Source:base_syncdata_unittest.py

...61        sync_ls = syncdata.SyncListenServer()62        os.kill(sync_ls.server_pid, 0)63        time.sleep(2)64        sync_ls.close()65        self.assertEqual(utils.pid_is_alive(sync_ls.server_pid), False,66                         "Server should be dead.")67    def test_SyncListenServer_start_with_tmp(self):68        tmpdir = "/tmp"69        sync_ls = syncdata.SyncListenServer(tmpdir=tmpdir)70        os.kill(sync_ls.server_pid, 0)71        time.sleep(2)72        sync_ls.close()73        self.assertEqual(utils.pid_is_alive(sync_ls.server_pid), False,74                         "Server should be dead.")75        self.assertEqual(os.path.exists(tmpdir), True,76                         "Tmpdir should exists after SyncListenServer"77                         " finish with predefined tmpdir.")78    def test_SyncData_with_listenServer(self):79        sync_ls = syncdata.SyncListenServer()80        sync = syncdata.SyncData("127.0.0.1", "127.0.0.1", ["127.0.0.1"],81                                 "127.0.0.1#1", sync_ls)82        data = sync.sync("test1")83        sync.close()84        sync_ls.close()85        self.assertEqual(data, {'127.0.0.1': 'test1'})86    def test_SyncData_with_self_listen_server(self):87        sync = syncdata.SyncData("127.0.0.1", "127.0.0.1", ["127.0.0.1"],88                                 "127.0.0.1#1")89        os.kill(sync.listen_server.server_pid, 0)90        data = sync.sync("test2")91        sync.close()92        self.assertEqual(utils.pid_is_alive(sync.listen_server.server_pid),93                         False, "Server should be dead.")94        self.assertEqual(data, {'127.0.0.1': 'test2'})95    def test_SyncData_with_listenServer_auto_close(self):96        sync = syncdata.SyncData("127.0.0.1", "127.0.0.1", ["127.0.0.1"],97                                 "127.0.0.1#1")98        os.kill(sync.listen_server.server_pid, 0)99        data = sync.single_sync("test3")100        self.assertEqual(utils.pid_is_alive(sync.listen_server.server_pid),101                         False, "Server should be dead.")102        self.assertEqual(data, {'127.0.0.1': 'test3'})103    def test_SyncData_with_closed_listenServer(self):104        sync_ls = syncdata.SyncListenServer()105        sync_ls.close()106        time.sleep(2)107        sync = syncdata.SyncData("127.0.0.1", "127.0.0.1", ["127.0.0.1"],108                                 "127.0.0.1#1", sync_ls)109        self.assertRaises(error.DataSyncError, lambda: sync.sync("test1", 2))110    def test_SyncData_with_listenServer_client_wait_timeout(self):111        sync = syncdata.SyncData("127.0.0.1", "127.0.0.1",112                                 ["127.0.0.1", "192.168.0.1"],113                                 "127.0.0.1#1")114        self.assertRaises(error.DataSyncError,...

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.