KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pyatom automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
