Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use do_download method in localstack
Best Python code snippet using localstack_python
test_download.py
Source:test_download.py 
1# Copyright (c) 2012-2022 by the GalSim developers team on GitHub2# https://github.com/GalSim-developers3#4# This file is part of GalSim: The modular galaxy image simulation toolkit.5# https://github.com/GalSim-developers/GalSim6#7# GalSim is free software: redistribution and use in source and binary forms,8# with or without modification, are permitted provided that the following9# conditions are met:10#11# 1. Redistributions of source code must retain the above copyright notice, this12# list of conditions, and the disclaimer given in the accompanying LICENSE13# file.14# 2. Redistributions in binary form must reproduce the above copyright notice,15# this list of conditions, and the disclaimer given in the documentation16# and/or other materials provided with the distribution.17#18import sys19import logging20import builtins21import shutil22from unittest import mock23import galsim24import galsim.download_cosmos # Not imported automatically25from galsim_test_helpers import *26# This file tests the galsim_download_cosmos executable.27# We use a lock of mocking to do these tests. We mock user input, urlopen, and more.28# The directory fake_cosmos has a tarball for the 23.5 sample and a directory for 25.2.29# So we use these separately at times when we need one or the other to exist.30@timer31def test_args():32 """Test the argument parsing33 """34 # -h prints description and exits35 print('This should print the description...') # Can only check by inspection, not assert36 with assert_raises(SystemExit):37 args = galsim.download_cosmos.parse_args(['-h'])38 # Check defaults:39 args = galsim.download_cosmos.parse_args([])40 assert args.verbosity == 241 assert args.force is False42 assert args.quiet is False43 assert args.unpack is False44 assert args.save is False45 assert args.dir is None46 assert args.sample == '25.2'47 assert args.nolink is False48 # Check setting each of those to be something else49 args = galsim.download_cosmos.parse_args(['-v', '3'])50 assert args.verbosity == 351 args = galsim.download_cosmos.parse_args(['-v', '0'])52 assert args.verbosity == 053 args = galsim.download_cosmos.parse_args(['-f'])54 assert args.force is True55 args = galsim.download_cosmos.parse_args(['-q'])56 assert args.quiet is True57 args = galsim.download_cosmos.parse_args(['-u'])58 assert args.unpack is True59 args = galsim.download_cosmos.parse_args(['--save'])60 assert args.save is True61 args = galsim.download_cosmos.parse_args(['-d','~/share'])62 assert args.dir == '~/share'63 args = galsim.download_cosmos.parse_args(['-s','23.5'])64 assert args.sample == '23.5'65 args = galsim.download_cosmos.parse_args(['-s','25.2'])66 assert args.sample == '25.2'67 args = galsim.download_cosmos.parse_args(['--nolink'])68 assert args.nolink is True69 # Some invalid parameters70 # To avoid ugly text output during nosetests runs, redirect stderr to stdout for a moment.71 sys_stderr = sys.stderr72 sys.stderr = sys.stdout73 with assert_raises(SystemExit):74 galsim.download_cosmos.parse_args(['-s', '25.9'])75 with assert_raises(SystemExit):76 galsim.download_cosmos.parse_args(['-v', '-1'])77 with assert_raises(SystemExit):78 galsim.download_cosmos.parse_args(['-v', '4'])79 sys.stderr = sys_stderr80# global for the bleh and delay functions81count = 082@timer83def test_query():84 """Test the query_yes_no function85 Need to mock the input function for this86 """87 from galsim.download_cosmos import query_yes_no88 def bleh():89 global count90 count += 191 return 'y' if count % 5 == 0 else 'bleh'92 def delay_y():93 global count94 count += 195 return 'y' if count % 5 == 0 else ''96 def delay_n():97 global count98 count += 199 return 'n' if count % 5 == 0 else ''100 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):101 assert query_yes_no('', 'yes') == 'yes'102 with mock.patch('galsim.download_cosmos.get_input', return_value='yes'):103 assert query_yes_no('', 'yes') == 'yes'104 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):105 assert query_yes_no('', 'yes') == 'no'106 with mock.patch('galsim.download_cosmos.get_input', return_value='no'):107 assert query_yes_no('', 'yes') == 'no'108 with mock.patch('galsim.download_cosmos.get_input', return_value=''):109 assert query_yes_no('', 'yes') == 'yes'110 with mock.patch('galsim.download_cosmos.get_input', bleh):111 assert query_yes_no('', 'yes') == 'yes'112 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):113 assert query_yes_no('', 'no') == 'yes'114 with mock.patch('galsim.download_cosmos.get_input', return_value='yes'):115 assert query_yes_no('', 'no') == 'yes'116 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):117 assert query_yes_no('', 'no') == 'no'118 with mock.patch('galsim.download_cosmos.get_input', return_value='no'):119 assert query_yes_no('', 'no') == 'no'120 with mock.patch('galsim.download_cosmos.get_input', return_value=''):121 assert query_yes_no('', 'no') == 'no'122 with mock.patch('galsim.download_cosmos.get_input', bleh):123 assert query_yes_no('', 'yes') == 'yes'124 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):125 assert query_yes_no('', None) == 'yes'126 with mock.patch('galsim.download_cosmos.get_input', return_value='yes'):127 assert query_yes_no('', None) == 'yes'128 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):129 assert query_yes_no('', None) == 'no'130 with mock.patch('galsim.download_cosmos.get_input', return_value='no'):131 assert query_yes_no('', None) == 'no'132 with mock.patch('galsim.download_cosmos.get_input', delay_n):133 assert query_yes_no('', None) == 'no'134 with mock.patch('galsim.download_cosmos.get_input', delay_y):135 assert query_yes_no('', None) == 'yes'136 with mock.patch('galsim.download_cosmos.get_input', bleh):137 assert query_yes_no('', None) == 'yes'138 with assert_raises(ValueError):139 query_yes_no('', 'invalid')140# Need to call these before each time make_logger is repeated. Else duplicate handles.141def remove_handler():142 logger = logging.getLogger('galsim')143 for handler in logger.handlers:144 logger.removeHandler(handler)145@timer146def test_names():147 """Test the get_names function148 """149 from galsim.download_cosmos import get_names150 args = galsim.download_cosmos.parse_args([])151 remove_handler()152 logger = galsim.download_cosmos.make_logger(args)153 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)154 assert url == 'https://zenodo.org/record/3242143/files/COSMOS_25.2_training_sample.tar.gz'155 assert target_dir == galsim.meta_data.share_dir156 assert do_link is False157 assert target == os.path.join(target_dir, 'COSMOS_25.2_training_sample.tar.gz')158 assert unpack_dir == os.path.join(target_dir, 'COSMOS_25.2_training_sample')159 assert link_dir == os.path.join(galsim.meta_data.share_dir, 'COSMOS_25.2_training_sample')160 args = galsim.download_cosmos.parse_args(['-d','~/share','-s','23.5'])161 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)162 assert url == 'https://zenodo.org/record/3242143/files/COSMOS_23.5_training_sample.tar.gz'163 assert target_dir == os.path.expanduser('~/share')164 assert do_link is True165 assert target == os.path.join(target_dir, 'COSMOS_23.5_training_sample.tar.gz')166 assert unpack_dir == os.path.join(target_dir, 'COSMOS_23.5_training_sample')167 assert link_dir == os.path.join(galsim.meta_data.share_dir, 'COSMOS_23.5_training_sample')168 args = galsim.download_cosmos.parse_args(['-d','share','--nolink'])169 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)170 assert url == 'https://zenodo.org/record/3242143/files/COSMOS_25.2_training_sample.tar.gz'171 assert target_dir == 'share'172 assert do_link is False173 assert target == os.path.join(target_dir, 'COSMOS_25.2_training_sample.tar.gz')174 assert unpack_dir == os.path.join(target_dir, 'COSMOS_25.2_training_sample')175 assert link_dir == os.path.join(galsim.meta_data.share_dir, 'COSMOS_25.2_training_sample')176class fake_urlopen:177 err = None178 # We don't want to actually check things on the internet. So this class fakes179 # up the real urlopen function.180 def __init__(self, url):181 self.n = 10182 pass183 def info(self):184 return {185 "Server": "nginx/1.16.1",186 "Content-Type": "application/octet-stream",187 "Content-Length": "728",188 "Connection": "close",189 "Content-MD5": "e05cfe60c037c645d61ac70545cc2a99",190 "Content-Security-Policy": "default-src 'none';",191 "X-Content-Type-Options": "nosniff",192 "X-Download-Options": "noopen",193 "X-Permitted-Cross-Domain-Policies": "none",194 "X-Frame-Options": "sameorigin",195 "X-XSS-Protection": "1; mode=block",196 "Content-Disposition": "attachment; filename=COSMOS_25.2_training_sample.tar.gz",197 "ETag": "\"md5:e05cfe60c037c645d61ac70545cc2a99\"",198 "Last-Modified": "Sun, 31 May 2020 02:19:18 GMT",199 "Date": "Thu, 11 Jun 2020 16:06:07 GMT",200 "Accept-Ranges": "none",201 "X-RateLimit-Limit": "60",202 "X-RateLimit-Remaining": "59",203 "X-RateLimit-Reset": "1591891628",204 "Retry-After": "60",205 "Strict-Transport-Security": "max-age=0",206 "Referrer-Policy": "strict-origin-when-cross-origin",207 "Set-Cookie": "session=3765c5a1d5211e53_5ee2566f",208 "X-Session-ID": "3765c5a1d5211e53_5ee2566f",209 "X-Request-ID": "2b720f14bdd71a29031a5cb415b391f8"210 }211 def read(self, block_sz):212 if self.err:213 raise OSError(self.err)214 if self.n:215 self.n -= 1216 return b'x' * 80 # ignore block_sz217 else:218 return b''219@timer220def test_check():221 """Test the get_meta and check_existing functions222 The latter of these is really the most interesting, and has the most bits223 of anything in the script that are worth checking with unit tests.224 """225 from galsim.download_cosmos import get_names, get_meta, check_existing226 args = galsim.download_cosmos.parse_args(['-d','fake_cosmos','-q','-v','3'])227 remove_handler()228 logger = galsim.download_cosmos.make_logger(args)229 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)230 # Check get_meta231 with mock.patch('galsim.download_cosmos.urlopen', fake_urlopen):232 meta = get_meta(url, args, logger)233 assert meta['Content-Length'] == "728"234 assert meta['Content-MD5'] == "e05cfe60c037c645d61ac70545cc2a99"235 # File already exists and is current.236 do_download = check_existing(target, unpack_dir, meta, args, logger)237 assert do_download is False238 # Some changes imply it's obsolete239 meta['Server'] = "nginx/1.23.1"240 meta['X-Content-Type-Options'] = "sniff"241 meta['Last-Modified'] = "Tue, 12 Mar 2019 08:12:12 GMT"242 meta['Date'] = "Sun, 14 Jun 2020 20:00:00 GMT"243 meta['X-RateLimit-Remaining'] = "31"244 meta['Retry-After'] = "120"245 meta['Set-Cookie'] = "session=2b720f14bdd71a29031a5cb415b391f8"246 do_download = check_existing(target, unpack_dir, meta, args, logger)247 assert do_download is False248 # Force download anyway249 args.quiet = False250 args.force = True251 do_download = check_existing(target, unpack_dir, meta, args, logger)252 assert do_download is True253 # Ask whether to re-download254 args.force = False255 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):256 do_download = check_existing(target, unpack_dir, meta, args, logger)257 assert do_download is True258 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):259 do_download = check_existing(target, unpack_dir, meta, args, logger)260 assert do_download is False261 # Meta data is obsolete262 meta1 = meta.copy()263 meta1['Content-Length'] = "9999"264 meta1['Content-MD5'] = "f05cfe60c037c645d61ac70545cc2a99"265 args.quiet = True266 do_download = check_existing(target, unpack_dir, meta1, args, logger)267 assert do_download is True268 # If they change the name of the checksum key, we consider it obsolete.269 meta2 = meta1.copy()270 meta2['Content-New-MD5'] = "e05cfe60c037c645d61ac70545cc2a99"271 del meta2['Content-MD5']272 del meta2['ETag']273 do_download = check_existing(target, unpack_dir, meta2, args, logger)274 assert do_download is True275 args.quiet = False276 args.force = True277 do_download = check_existing(target, unpack_dir, meta1, args, logger)278 assert do_download is True279 args.force = False280 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):281 do_download = check_existing(target, unpack_dir, meta1, args, logger)282 assert do_download is True283 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):284 do_download = check_existing(target, unpack_dir, meta1, args, logger)285 assert do_download is False286 # Meta data is missing287 args.quiet = True288 do_download = check_existing(target, 'output', meta, args, logger)289 assert do_download is True290 # Tarball is present, but wrong size291 args = galsim.download_cosmos.parse_args(['-d','fake_cosmos','-s','23.5','-q'])292 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)293 do_download = check_existing(target, 'output', meta1, args, logger)294 assert do_download is True295 args.quiet = False296 args.force = True297 do_download = check_existing(target, 'output', meta1, args, logger)298 assert do_download is True299 args.force = False300 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):301 do_download = check_existing(target, unpack_dir, meta1, args, logger)302 assert do_download is True303 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):304 do_download = check_existing(target, unpack_dir, meta1, args, logger)305 assert do_download is False306 # Tarball is present, and correct size307 args.quiet = True308 do_download = check_existing(target, 'output', meta, args, logger)309 assert do_download is False310 args.quiet = False311 args.force = True312 do_download = check_existing(target, 'output', meta, args, logger)313 assert do_download is True314 args.force = False315 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):316 do_download = check_existing(target, unpack_dir, meta, args, logger)317 assert do_download is True318 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):319 do_download = check_existing(target, unpack_dir, meta, args, logger)320 assert do_download is False321 # Tarball and unpack_dir both missing322 args = galsim.download_cosmos.parse_args(['-d','input'])323 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)324 do_download = check_existing(target, unpack_dir, meta, args, logger)325 assert do_download is True326@timer327def test_download():328 """Test the download function329 This one is a little silly. It's almost completely mocked. But we can at least check330 that there are no bugs that would raise an exception of some sort.331 """332 from galsim.download_cosmos import get_names, get_meta, download333 args = galsim.download_cosmos.parse_args(['-d','output','-q'])334 remove_handler()335 logger = galsim.download_cosmos.make_logger(args)336 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)337 with mock.patch('galsim.download_cosmos.urlopen', fake_urlopen):338 meta = get_meta(url, args, logger)339 print('Start download with verbosity = 2')340 download(True, url, target, meta, args, logger)341 print('Start download with verbosity = 1')342 args.verbosity = 1343 download(True, url, target, meta, args, logger)344 print('Start download with verbosity = 3')345 args.verbosity = 3346 download(True, url, target, meta, args, logger)347 print('Start download with verbosity = 0')348 args.verbosity = 0349 download(True, url, target, meta, args, logger)350 print("Don't download")351 download(False, url, target, meta, args, logger)352 fake_urlopen.err = 'Permission denied'353 with CaptureLog() as cl:354 assert_raises(OSError, download, True, url, target, meta, args, cl.logger)355 assert "Rerun using sudo" in cl.output356 fake_urlopen.err = 'Disk quota exceeded'357 with CaptureLog() as cl:358 assert_raises(OSError, download, True, url, target, meta, args, cl.logger)359 assert "You might need to download this in an alternate location" in cl.output360 fake_urlopen.err = 'gack'361 assert_raises(OSError, download, True, url, target, meta, args, logger)362 fake_urlopen.err = None363@timer364def test_unpack():365 """Test the check_unpack and unpack functions366 """367 from galsim.download_cosmos import get_names, check_unpack, unpack368 # If we downloaded the file, then we usually want to unpack369 args = galsim.download_cosmos.parse_args(['-d','fake_cosmos','-s','23.5','-q'])370 remove_handler()371 logger = galsim.download_cosmos.make_logger(args)372 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)373 meta = fake_urlopen(url).info()374 print('unpack_dir = ',unpack_dir)375 if os.path.exists(unpack_dir):376 shutil.rmtree(unpack_dir)377 # Regular case, downloaded file and not unpacked yet378 do_unpack = check_unpack(True, unpack_dir, target, args)379 assert do_unpack is True380 # If we didn't download, but tarball exists, still unpack381 do_unpack = check_unpack(False, unpack_dir, target, args)382 assert do_unpack is True383 # Now unpack it384 print('unpack with verbose = 2:')385 unpack(True, target, target_dir, unpack_dir, meta, args, logger)386 shutil.rmtree(unpack_dir)387 print('unpack with verbose = 3:')388 args.verbosity = 3389 unpack(True, target, target_dir, unpack_dir, meta, args, logger)390 shutil.rmtree(unpack_dir)391 print('unpack with verbose = 1:')392 args.verbosity = 1393 unpack(True, target, target_dir, unpack_dir, meta, args, logger)394 print("Don't unpack")395 unpack(False, target, target_dir, unpack_dir, meta, args, logger)396 # If it is already unpacked, probably don't unpack it397 do_unpack = check_unpack(False, unpack_dir, target, args)398 assert do_unpack is False399 # Unless we expressly say to on the command line400 args.unpack = True401 do_unpack = check_unpack(False, unpack_dir, target, args)402 assert do_unpack is True403 # Or if we downloaded a new tarball, we will unpack it.404 args.unpack = False405 do_unpack = check_unpack(True, unpack_dir, target, args)406 assert do_unpack is True407 # Or if not quiet, it will ask whether to re-unpack the tarball408 args.quiet = False409 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):410 do_unpack = check_unpack(False, unpack_dir, target, args)411 assert do_unpack is True412 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):413 do_unpack = check_unpack(False, unpack_dir, target, args)414 assert do_unpack is False415 # Finally, if the tarball doesn't exist, then we can't unpack it.416 target2 = target.replace('23.5','25.2')417 do_unpack = check_unpack(True, unpack_dir, target2, args)418 assert do_unpack is False419@timer420def test_remove():421 """Test the check_remove and remove_tarball function422 """423 from galsim.download_cosmos import get_names, check_remove, remove_tarball424 args = galsim.download_cosmos.parse_args(['-d','fake_cosmos','-s','23.5','-q'])425 remove_handler()426 logger = galsim.download_cosmos.make_logger(args)427 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)428 # Normally, we remove the tarball if we unpacked it.429 do_remove = check_remove(True, target, args)430 assert do_remove is True431 # Or if we say to save it, then save it432 args.save = True433 do_remove = check_remove(True, target, args)434 assert do_remove is False435 # If we didn't unpack it, probably don't delete it436 args.save = False437 do_remove = check_remove(False, target, args)438 assert do_remove is False439 # But ask if not quiet to be sure440 args.quiet = False441 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):442 do_remove = check_remove(False, target, args)443 assert do_remove is True444 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):445 do_remove = check_remove(False, target, args)446 assert do_remove is False447 target1 = target + '.tar'448 with open(target1,'w') as f:449 f.write('blah')450 assert os.path.isfile(target1)451 remove_tarball(False, target1, logger)452 assert os.path.isfile(target1)453 remove_tarball(True, target1, logger)454 assert not os.path.isfile(target1)455@timer456def test_link():457 """Test the link_cosmos function458 """459 from galsim.download_cosmos import get_names, make_link460 args = galsim.download_cosmos.parse_args(['-d','fake_cosmos','-q'])461 remove_handler()462 logger = galsim.download_cosmos.make_logger(args)463 url, target, target_dir, link_dir, unpack_dir, do_link = get_names(args, logger)464 link_dir = os.path.join('output', 'COSMOS_25.2_training_sample')465 # If link doesn't exist yet, make it.466 if os.path.lexists(link_dir):467 os.unlink(link_dir)468 assert not os.path.lexists(link_dir)469 make_link(True, unpack_dir, link_dir, args, logger)470 assert os.path.exists(link_dir)471 assert os.path.islink(link_dir)472 # If link already exists, remove and remake473 make_link(True, unpack_dir, link_dir, args, logger)474 assert os.path.exists(link_dir)475 assert os.path.islink(link_dir)476 # If do_link is False, don't make it477 os.unlink(link_dir)478 make_link(False, unpack_dir, link_dir, args, logger)479 assert not os.path.exists(link_dir)480 # If link exists, but is a directory, don't remove it481 os.mkdir(link_dir)482 assert os.path.isdir(link_dir)483 make_link(True, unpack_dir, link_dir, args, logger)484 assert os.path.exists(link_dir)485 assert os.path.isdir(link_dir)486 assert not os.path.islink(link_dir)487 # Unless force488 args.force = True489 make_link(True, unpack_dir, link_dir, args, logger)490 assert os.path.exists(link_dir)491 assert os.path.islink(link_dir)492 # Or ask493 args.force = False494 args.quiet = False495 os.unlink(link_dir)496 os.mkdir(link_dir)497 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):498 make_link(True, unpack_dir, link_dir, args, logger)499 assert os.path.isdir(link_dir)500 assert not os.path.islink(link_dir)501 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):502 make_link(True, unpack_dir, link_dir, args, logger)503 assert os.path.islink(link_dir)504 # If it's a broken link, remove and relink505 os.unlink(link_dir)506 os.symlink('invalid', link_dir)507 assert os.path.lexists(link_dir)508 assert os.path.islink(link_dir)509 assert not os.path.exists(link_dir)510 make_link(True, unpack_dir, link_dir, args, logger)511 assert os.path.exists(link_dir)512 assert os.path.islink(link_dir)513 # If it's a file, probably remove and relink514 os.unlink(link_dir)515 with open(link_dir,'w') as f: f.write('blah')516 args.quiet = True517 assert os.path.exists(link_dir)518 assert not os.path.islink(link_dir)519 make_link(True, unpack_dir, link_dir, args, logger)520 assert os.path.exists(link_dir)521 assert os.path.islink(link_dir)522 os.unlink(link_dir)523 with open(link_dir,'w') as f: f.write('blah')524 args.quiet = False525 args.force = True526 assert os.path.exists(link_dir)527 assert not os.path.islink(link_dir)528 make_link(True, unpack_dir, link_dir, args, logger)529 assert os.path.exists(link_dir)530 assert os.path.islink(link_dir)531 # But ask if appropriate532 os.unlink(link_dir)533 with open(link_dir,'w') as f: f.write('blah')534 args.force = False535 assert os.path.exists(link_dir)536 assert not os.path.islink(link_dir)537 with mock.patch('galsim.download_cosmos.get_input', return_value='n'):538 make_link(True, unpack_dir, link_dir, args, logger)539 assert os.path.exists(link_dir)540 assert not os.path.islink(link_dir)541 with mock.patch('galsim.download_cosmos.get_input', return_value='y'):542 make_link(True, unpack_dir, link_dir, args, logger)543 assert os.path.exists(link_dir)544 assert os.path.islink(link_dir)545@timer546def test_full():547 """Test the full script548 """549 link_dir1 = os.path.join('output', 'COSMOS_23.5_training_sample')550 link_dir2 = os.path.join('output', 'COSMOS_25.2_training_sample')551 if os.path.lexists(link_dir1):552 os.unlink(link_dir1)553 if os.path.lexists(link_dir2):554 os.unlink(link_dir2)555 with mock.patch('galsim.download_cosmos.share_dir', 'output'), \556 mock.patch('galsim.download_cosmos.urlopen', fake_urlopen):557 remove_handler()558 assert not os.path.islink(link_dir1)559 galsim.download_cosmos.main(['-d','fake_cosmos','-q','-s','23.5','--save'])560 assert os.path.islink(link_dir1)561 remove_handler()562 assert not os.path.islink(link_dir2)563 with mock.patch('sys.argv', ['galsim_download_cosmos', '-d', 'fake_cosmos', '-q']):564 galsim.download_cosmos.run_main()565 assert os.path.islink(link_dir2)566if __name__ == "__main__":567 testfns = [v for k, v in vars().items() if k[:5] == 'test_' and callable(v)]568 for testfn in testfns:...
test_manager.py
Source:test_manager.py
...128 def setUp(self):129 self.tempdir = tempfile.mkdtemp()130 self.addCleanup(shutil.rmtree, self.tempdir)131 self.dl_manager = download_manager.DownloadManager(self.tempdir)132 def do_download(self, url, fname, data, wait=0):133 session, response = mock_session()134 mock_response(response, data, wait)135 # patch the session, so the download will use that136 self.dl_manager.session = session137 return self.dl_manager.download(url, fname)138 def test_download(self):139 dl1 = self.do_download('http://foo', 'foo', 'hello' * 4, wait=0.02)140 self.assertIsInstance(dl1, download_manager.Download)141 self.assertTrue(dl1.is_running())142 # with the same fname, no new download is started. The same instance143 # is returned since the download is running.144 dl2 = self.do_download('http://bar', 'foo', 'hello2' * 4, wait=0.02)145 self.assertEquals(dl1, dl2)146 # starting a download with another fname will trigger a new download147 dl3 = self.do_download('http://bar', 'foo2', 'hello you' * 4)148 self.assertIsInstance(dl3, download_manager.Download)149 self.assertNotEquals(dl3, dl1)150 # let's wait for the downloads to finish151 dl3.wait()152 dl1.wait()153 # now if we try to download a fname that exists, None is returned154 dl4 = self.do_download('http://bar', 'foo', 'hello2' * 4, wait=0.02)155 self.assertIsNone(dl4)156 # downloaded files are what is expected157 def content(fname):158 with open(os.path.join(self.tempdir, fname)) as f:159 return f.read()160 self.assertEquals(content('foo'), 'hello' * 4)161 self.assertEquals(content('foo2'), 'hello you' * 4)162 # download instances are removed from the manager (internal test)163 self.assertEquals(self.dl_manager._downloads, {})164 def test_cancel(self):165 dl1 = self.do_download('http://foo', 'foo', 'foo' * 50000, wait=0.02)166 dl2 = self.do_download('http://foo', 'bar', 'bar' * 50000, wait=0.02)167 dl3 = self.do_download('http://foo', 'foobar', 'foobar' * 4)168 # let's cancel only one169 def cancel_if(dl):170 if os.path.basename(dl.get_dest()) == 'foo':171 return True172 self.dl_manager.cancel(cancel_if=cancel_if)173 self.assertTrue(dl1.is_canceled())174 self.assertFalse(dl2.is_canceled())175 self.assertFalse(dl3.is_canceled())176 # wait for dl3177 dl3.wait()178 # cancel everything179 self.dl_manager.cancel()180 self.assertTrue(dl1.is_canceled())181 self.assertTrue(dl2.is_canceled())...
HHTick.py
Source:HHTick.py
...23 print('passwd error')24 return None25 print('ssh login success')26 return child27def do_download(handler, dte, exch):28 #pdb.set_trace()29 try:30 cmd = 'cd /datas/production/HHTICK_C/bin/'31 handler.sendline(cmd)32 print handler.readline()33 index = handler.expect(['bin', '没æé£ä¸ªæ件æç®å½'])34 if index == 1:35 print('path error')36 return None37 cmd = './HHTick ../config/config_204_{}.xml {} \n'.format(exch,dte)38 handler.sendline(cmd)39 while True:40 process = handler.readline()41 print(process)42 if process.find('quit') != -1:43 print('down load SH L2 done')44 handler.sendcontrol('c')45 break46 else:47 try:48 if int(process) < 1000:49 print('down load error')50 break51 except Exception,e:52 continue53 #cmd = './HHTick ../config/config_204_SZ.xml {}'.format(dte)54 #handler.sendline(cmd)55 #while True:56 # process = handler.readline()57 # print(process)58 # if process.find('quit') != -1:59 # print('down load SZ L2 done')60 # handler.sendcontrol('c')61 # break62 # else:63 # try:64 # if int(process) < 1000:65 # print('down load error')66 # break67 # except Exception,e:68 # continue69 70 except Exception,e:71 print 'do_download {} failed, reason:{}'.format(dte, e)72 73def del_data(dte,exch):74 try:75 handler = ssh_login(host, user, passwd)76 cmd = 'cd /datas/production/HHTICK_C/data/day/{}/'.format(exch)77 handler.sendline(cmd)78 print handler.readline()79 # index = handler.expect(['SH', '没æé£ä¸ªæ件æç®å½'])80 # if index == 1:81 # print('path error'),cmd82 # return None83 del_cmd = 'rm -f HHTICK_{}.dat'.format(dte)84 handler.sendline(del_cmd)85 print handler.readline()86 time.sleep(1)87 except Exception,e:88 print 'del_data {} failed, reason:{}'.format(dte, e)89 #try:90 # cmd = 'cd /datas/production/HHTICK_C/data/day/SZ/'91 # handler.sendline(cmd)92 # print handler.readline()93 # # index = handler.expect(['SZ', '没æé£ä¸ªæ件æç®å½'])94 # # if index == 1:95 # # print('path error:'),cmd96 # # return None97 # del_cmd = 'rm -f HHTICK_{}.dat'.format(dte)98 # handler.sendline(del_cmd)99 # print handler.readline()100 # time.sleep(2)101 #except Exception,e:102 # print 'del_data {} failed, reason:{}'.format(dte, e)103def down_one(dte):104 handler = ssh_login(host, user, passwd)105 if handler:106 print dte107 do_download(handler, dte, 'SH')108 do_download(handler, dte, 'SZ')109 do_download(handler, dte, 'CF')110 handler.close(force=True)111 down_load_l2(dte, 'SH')112 down_load_l2(dte, 'SZ')113 down_load_l2(dte, 'CF')114 del_data(dte, 'SH')115 del_data(dte, 'SZ')116 del_data(dte, 'CF')117def down_cal():118 td = int(datetime.datetime.today().strftime("%Y%m%d"))119 down_one(td)120 downlist = [x.strip('\n') for x in open('../config/downlist.txt','r').readlines()]121 tme = int(datetime.datetime.now().strftime("%H%M%S"))122 while tme < 200000:123 print downlist[-1]124 down_one(downlist[-1])125 downlist = downlist[0:-1]126 time.sleep(2)127 tme = int(datetime.datetime.now().strftime("%H%M%S"))128 129 #dtes = downlist[-2:] + [td]130 #downlist = downlist[:-2] 131 #handler = ssh_login(host, user, passwd)132 #if handler:133 # for dte in dtes:134 # print dte135 # do_download(handler, dte)136 # handler.close(force=True)137 # for dte in dtes:138 # down_load_l2(dte)139 # del_data(dte)140 #open('../config/downlist.txt','w').write('\n'.join(downlist))141 #142 #time.sleep(3)143 dtes = downlist[-3:]144 downlist = downlist[:-3]145 handler = ssh_login(host, user, passwd)146 if handler:147 for dte in dtes:148 print dte149 do_download(handler, dte, 'SH')150 do_download(handler, dte, 'SZ')151 handler.close(force=True)152 for dte in dtes:153 down_load_l2(dte, 'SH')154 down_load_l2(dte, 'SZ')155 del_data(dte, 'SH')156 del_data(dte, 'SZ')157 open('../config/downlist.txt','w').write('\n'.join(downlist))158if __name__ == '__main__':159 #down_cal()160 dte = int(datetime.datetime.today().strftime("%Y%m%d"))161 if len(sys.argv) > 1:162 dte = int(sys.argv[1])...
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 localstack 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?
