Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use kill_daemon method in pyatom

Best Python code snippet using pyatom_python

host.py

Source:host.py

...77        self.ipfs_path = ipfs_path78        self.devnull = open(os.devnull, "w")79        self._ensure_path()80    def teardown(self):81        self.kill_daemon()82        self._unset_path()83    def launch_daemon(self):84        if self.daemon_available():85            return86        elif self.daemon_running():87            self.kill_daemon()88        print("Launching daemon...")89        while not self.daemon_running():90            try:91                subprocess.Popen(92                    self._get_command("daemon &"), shell=True, stdout=self.devnull)93            except:94                self.kill_daemon()95        while not self.daemon_available():96            print("Waiting for daemon... [may take a min, do not quit]")97            time.sleep(1)98        print("Success.")99    def daemon_available(self):100        try:101            subprocess.check_call(102                self._get_command("stats bitswap"),103                shell=True,104                stdout=self.devnull,105                stderr=self.devnull,106            )107            return True108        except:109            return False110    def daemon_running(self):111        try:112            return int(subprocess.check_output("pgrep ipfs", shell=True).strip())113        except:114            return False115    def kill_daemon(self):116        pid = self.daemon_running()117        while self.daemon_running():118            print("Killing daemon...")119            try:120                os.kill(pid, signal.SIGKILL)121            except:122                return123    def init(self):124        try:125            subprocess.check_call(126                self._get_command("init"),127                shell=True,128                stdout=self.devnull,129                stderr=self.devnull,130            )131        except:132            pass133    def is_connected(self, ipfs_node):134        addrs = self.check_output("swarm addrs").decode("utf8")135        return ipfs_node.id in addrs136    def ensure_connected(self, ipfs_node):137        if not self.is_connected(ipfs_node):138            self.check_output("swarm connect {}".format(ipfs_node.address))139    def ensure_disconnected(self, ipfs_node):140        if self.is_connected(ipfs_node):141            self.check_output("swarm disconnect {}".format(ipfs_node.address))142    def call(self, command):143        return subprocess.call(self._get_command(command), shell=True)144    def time_get(self, hash):145        return float(self.time("get {}".format(hash)).strip())146    def time(self, command):147        # TIMEFORMAT=%R148        if os.environ.get("TIMEFORMAT") != "%R":149            os.environ["TIMEFORMAT"] = "%R"150        try:151            return subprocess.check_output("(time " + self._get_command(command) + "&> /dev/null ) 2>&1", shell=True, executable='bash').decode("utf8")152        except:153            print("ERROR: Failed to run ipfs command: {}. Please report this error".format(command))154            exit()155        os.environ.pop("TIMEFORMAT", None)156        os.unsetenv("TIMEFORMAT")157    def check_output(self, command):158        try:159            return subprocess.check_output(self._get_command(command), shell=True)160        except:161            print("ERROR: Failed to run ipfs command: {}. Please report this error".format(command))162            exit()163    def _ensure_path(self):164        if os.environ.get("IPFS_PATH") != self.ipfs_path:165            os.environ["IPFS_PATH"] = self.ipfs_path166    def _unset_path(self):167        os.environ.pop("IPFS_PATH", None)168        os.unsetenv("IPFS_PATH")169    def _get_command(self, command):170        return "{} {}".format(self.ipfs_binary, command)171    def get_stats(self, file, hosts, samples=10):172        print("Collecting ipfs stats...")173        gets = []174        tries = 0175        while len(gets) < samples and tries < samples*2:176            tries += 1177            print("Attempt {} out of (min: {}, max: {})".format(tries, samples, samples*2))178            for h in hosts:179                self.ensure_connected(h)180            self.check_output("repo gc")181            if os.path.exists(file.hash):182                os.remove(file.hash)183            t = self.time_get(file.hash)184            if all(self.is_connected(h) for h in hosts):185                gets.append(t)186        if os.path.exists(file.hash):187            os.remove(file.hash)188        return {"tries": tries, "gets": gets, "average": sum(gets) / len(gets)}189    # all nodes in our CDN swarm repeatedly generate or request this ever-changing file,190    #   using the existing IPFS DHT to discover other peers who are doing the same191    # hosts generate the next two files in advance to be ready to be found192    def genHostSwarmFiles(self):193        # gen session id based on only known piece of shared* data: time194        #   *yes I know time is hard, but it's approximately shared195        sessionId = int(time.time() / 60) # new session every minute196        # generate the token for this session and the next two197        sharedTokens = ["Member of P2P CDN Session: {}".format(sessionId + i) for i in range(3)]198        # add and pin those tokens199        for t in sharedTokens:200            cmd = 'echo -n "{0}" | {1} add | awk  {2} | {1} pin add'.format(t, self.ipfs_binary, "'{print $2}'")201            try:202                subprocess.check_output(cmd, shell=True)203            except:204                print("ERROR: Failed to run command: {}. Please report this error".format(cmd))205class IPFSDownloader:206    BASE_URL = "https://dist.ipfs.io/go-ipfs/v0.5.0/go-ipfs_v0.5.0_"207    FOLDER_NAME = "go-ipfs"208    PATH_TO_FOLDER = os.path.join(209        os.path.dirname(os.path.realpath(__file__)), FOLDER_NAME)210    @classmethod211    def run(cls):212        if os.path.exists(cls.PATH_TO_FOLDER):213            return cls.PATH_TO_FOLDER214        cls.download_and_extract(cls.get_ipfs_download_link())215        return cls.PATH_TO_FOLDER216    @classmethod217    def delete(cls):218        print("Deleting folder...")219        if os.path.exists(cls.PATH_TO_FOLDER):220            shutil.rmtree(cls.PATH_TO_FOLDER)221    @classmethod222    def get_ipfs_download_link(cls):223        return cls.BASE_URL + cls.get_ipfs_download_postfix()224    @staticmethod225    def get_ipfs_download_postfix():226        is_64bits = sys.maxsize > 2 ** 32227        is_arm = "arm" in platform.machine()228        platform_name = sys.platform229        if platform_name == "linux" or platform_name == "linux2":230            # linux231            if is_arm:232                if is_64bits:233                    return "linux-arm64.tar.gz"234                return "linux-arm.tar.gz"235            if is_64bits:236                return "linux-amd64.tar.gz"237            return "linux-386.tar.gz"238        elif platform_name == "darwin":239            # OS X240            if is_64bits:241                return "darwin-amd64.tar.gz"242            return "darwin-386.tar.gz"243        elif platform_name == "win32":244            # Windows...245            sys.exit("Windows is not supported")246    @staticmethod247    def download_and_extract(url):248        print("Downloading...")249        file_tmp = urlretrieve(url, filename=None)[0]250        tar = tarfile.open(file_tmp)251        print("Extracting...")252        tar.extractall()253def main():254    parser = argparse.ArgumentParser(255        description="Experiments. Please contact the sender for questions.")256    parser.add_argument("--kill",257                        help="Kill the background node",258                        action="store_true")259    parser.add_argument("-d", "--dotipfs",260                        help="/path/to/.ipfs/ (default: .ipfs)")261    args = parser.parse_args()262    if sys.platform == "win32":263        sys.exit("Windows is not supported")264    ipfs_folder = IPFSDownloader.run()265    if args.dotipfs:266        dotipfs = args.dotipfs267    else:268        dotipfs = os.path.join(ipfs_folder, ".ipfs/")269    ipfs = IPFSClient(os.path.join(ipfs_folder, "ipfs"), dotipfs)270    ipfs.init()271    if args.kill:272        ipfs.kill_daemon()273    else:274        print(275"""276Thank you for volunteering! You've helping us beat commercial CDNs277with P2P technology!278Please keep your computer powered on and connected to the internet. Do279not tamper with the ./go-ipfs directory or the running ipfs daemon280process.281If you wish to turn off the host (we hope you'll stay!) you may run:282./host.py --kill283 """)284        ipfs.launch_daemon()285        # pin the files286        for f in FILES.values():287            print("pinning {}...".format(f.hash))288            ipfs.check_output("pin add {}".format(f.hash))289        print("\nIPFS client info (using {}):".format(dotipfs))290        ipfs.call("id")291        print(292"""293Host is running! Pleae do not close this shell.294To keep running long term, we recommend using tmux295(https://github.com/tmux/tmux/wiki) or screen296(https://www.gnu.org/software/screen/). Feel free to close and rerun297this script as needed.298 """)299        while True:300            ipfs.launch_daemon()301            ipfs.genHostSwarmFiles()302            time.sleep(60)303    ipfs.kill_daemon()304    ipfs.teardown()305    IPFSDownloader.delete()306if __name__ == "__main__":...

run_tests.py

Source:run_tests.py

...50        if kill:51            os.kill(pid, signal.SIGKILL)52        else:53            os.kill(pid, signal.SIGTERM)54def kill_daemon(daemon_exe):55    stop_daemon(daemon_exe, True)56def daemon_exists(daemon_exe):57    if get_daemon_pid(daemon_exe) > 0:58        return True59    return False60def check_daemons_log(from_time, check_str, times):61    p1 = subprocess.Popen(['journalctl', '-q', '--since', from_time], stdout=subprocess.PIPE)62    p2 = subprocess.Popen(["grep", "crc"], stdin=p1.stdout, stdout=subprocess.PIPE)63    p1.stdout.close()64    output, error = p2.communicate()65    return output.decode('utf-8').count(check_str) == times66if __name__ == "__main__":67    source_dir = os.path.dirname(os.path.realpath(__file__))68    build_dir = os.path.join(os.getcwd(), 'build')69    if os.path.exists(build_dir):70        try:71            shutil.rmtree(build_dir)72        except OSError as e:73            print("remove build directory error: %s - %s." % (e.filename, e.strerror))74    os.mkdir(build_dir)75    build_project(source_dir, build_dir)76    daemon_exe = 'crc32_check_daemon'77    #78    print('TEST 1: start daemon with args... ')79    run_daemon(build_dir, daemon_exe, '/proc', 1)80    result = daemon_exists(daemon_exe)81    kill_daemon(daemon_exe)82    assert result == True, 'failed: ' + daemon_exe + " process doesn't exist"83    print('Ok')84    #85    print('TEST 2: stop daemon with SIGTERM... ')86    run_daemon(build_dir, daemon_exe, '/proc', 1)87    time.sleep(1)88    stop_daemon(daemon_exe)89    time.sleep(1)90    result = daemon_exists(daemon_exe)91    assert result == False, (kill_daemon(daemon_exe) and False) or 'failed: ' + daemon_exe + " process still exists"92    print('Ok')93    #94    print("TEST 3: run daemon for 3 seconds without dir's modification... ")95    test_dir = generate_test_dir(build_dir, 5)96    start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")97    run_daemon(build_dir, daemon_exe, test_dir, 1)98    time.sleep(3)99    kill_daemon(daemon_exe)100    rm_dir(test_dir)101    result = check_daemons_log(start_time, "Integrity check: OK", 3)102    assert result == True, 'failed: ' + daemon_exe + " wrong logs"103    print('Ok')104    #105    time.sleep(2) # wait old daemon stops106    print("TEST 4: request check with SIGUSR1... ")107    test_dir = generate_test_dir(build_dir, 5)108    start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")109    run_daemon(build_dir, daemon_exe, test_dir, 100) #BIG timeout110    for _ in range(10): request_dir_check(daemon_exe)111    kill_daemon(daemon_exe)112    rm_dir(test_dir)113    result = check_daemons_log(start_time, "Integrity check: OK", 10)114    assert result == True, 'failed: ' + daemon_exe + " wrong logs"115    print('Ok')116    #117    time.sleep(2) # wait old daemon stops118    print("TEST 5: add new file... ")119    test_dir = generate_test_dir(build_dir, 5)120    start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")121    run_daemon(build_dir, daemon_exe, test_dir, 100) #BIG timeout122    #create new file123    time.sleep(1)124    with open(os.path.join(test_dir, '6.txt'), 'w') as f:125        f.write('6'*6)126    time.sleep(1)127    request_dir_check(daemon_exe)128    kill_daemon(daemon_exe)129    rm_dir(test_dir)130    result = check_daemons_log(start_time, "Integrity check: FAIL (6.txt - new file)", 1)131    assert result == True, 'failed: ' + daemon_exe + " wrong logs"132    print('Ok')133    #134    time.sleep(2) # wait old daemon stops135    print("TEST 6: change existing file... ")136    test_dir = generate_test_dir(build_dir, 5)137    start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")138    run_daemon(build_dir, daemon_exe, test_dir, 100) #BIG timeout139    #create new file140    time.sleep(1)141    with open(os.path.join(test_dir, '3.txt'), 'w') as f:142        f.write('7'*7)143    time.sleep(1)144    request_dir_check(daemon_exe)145    kill_daemon(daemon_exe)146    rm_dir(test_dir)147    result = check_daemons_log(start_time, "Integrity check: FAIL (3.txt - new crc", 1)148    assert result == True, 'failed: ' + daemon_exe + " wrong logs"149    print('Ok')150    #151    time.sleep(2) # wait old daemon stops152    print("TEST 7: delete existing file... ")153    test_dir = generate_test_dir(build_dir, 5)154    start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")155    run_daemon(build_dir, daemon_exe, test_dir, 100) #BIG timeout156    #create new file157    time.sleep(1)158    os.remove(os.path.join(test_dir, '2.txt'))159    time.sleep(1)160    request_dir_check(daemon_exe)161    kill_daemon(daemon_exe)162    rm_dir(test_dir)163    result = check_daemons_log(start_time, "Integrity check: FAIL (2.txt - file was deleted)", 1)164    assert result == True, 'failed: ' + daemon_exe + " wrong logs"165    print('Ok')166    #TODO check crc32...

daemonizer.py

Source:daemonizer.py

...5    @staticmethod6    def handle():7        signal.signal(signal.SIGINT, DaemonKiller.kill_daemon)8        signal.signal(signal.SIGTERM, DaemonKiller.kill_daemon)9    def kill_daemon(signum, frame):10        GPIO.cleanup()11        print("Exiting...")...

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.