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

How to use upload_pkg_file method in autotest

Best Python code snippet using autotest_python

apollo_upload_pkg.py

Source:apollo_upload_pkg.py

1#-*- encoding: utf-8 -*-2from optparse import OptionParser3import os4import re5import string6import sys7from datetime import datetime8import traceback9import hashlib10import socket11import random12class ApolloRsyncPkg(object):13    """ å®ç°æä»¶çrsyncã14    #  Attributes:15    """16    def __init__(self, user, app, svr, version, directory, reupload):17        """ åå§å18        """19        # rsync ip addr. éå®ååçipå°å20        rsync_ip = socket.gethostbyname("rsync.sre.lecloud.com")21        # æä»¶ä¸ä¼ çrsync å°å22        self.rsync_server="apollo@%s::upload" % (rsync_ip)23        # ä¸ä¼ çç¨æ·24        self.user = user25        # ä¸ä¼ pkgçäº§åå26        self.app = app27        # ä¸ä¼ pkgçæå¡å28        self.svr = svr29        # ä¸ä¼ pkgçæå¡çæ¬30        self.version = version31        # pkgæå¨çç®å½ï¼å¿é¡»æ¯ç»å¯¹ç®å½32        self.directory = directory33        # ä¸ä¼ çæ¶é´34        self.uptime = datetime.now().strftime('%Y-%m-%d %H:%M:%S')35        # pkgçmd536        self.md5 = ""37        # pkgçæºå¯éæºåç¬¦ä¸²38        self.passwd = ''.join(random.sample(string.ascii_letters + string.digits, 32)) + \39                      ''.join(random.sample(string.ascii_letters + string.digits, 32)) + \40                      ''.join(random.sample(string.ascii_letters + string.digits, 32)) + \41                      ''.join(random.sample(string.ascii_letters + string.digits, 32))42        self.pkg_file = "pkg_%s_%s_%s.tar.gz" % (app, svr, version)43        self.pkg_pathfile = "/tmp/%s" % (self.pkg_file)44        self.pkg_info_file = "pkg_%s_%s_%s.info" % (app, svr, version)45        self.pkg_info_pathfile = "/tmp/%s" % (self.pkg_info_file)46        self.reupload = reupload47        48    def rsync(self):49        """ rsync pkg50        """51        cmd = "cd %s && tar -czvf - * | openssl des3 -salt -k %s -out %s" % \52            (self.directory, self.passwd, self.pkg_pathfile)53        print "tar pkg path, cmd:%s" % (cmd)54        ret =os.system(cmd)55        if 0 <> ret:56            print "Failed to exec:%s" % (cmd)57            return False58        self.__md5()59        upload_pkg_file = "pkg_%s_%s_%s.%s.tar.gz" % (self.app, self.svr, self.version, self.md5[0:16])60        pkg_info = "[pkg]\ntype=pkg\nreupload=%s\nuser=%s\napp=%s\nsvr=%s\nversion=%s\ndirectory=%s\npkg_file=%s\ntime=%s\nmd5=%s\npasswd=%s\n" % \61            (self.reupload, self.user, self.app, self.svr, self.version, self.directory, upload_pkg_file, self.uptime, self.md5, self.passwd)62        # write pkg_info63        print "Write pkg info to:%s" % (self.pkg_info_pathfile)64        with open(self.pkg_info_pathfile, "wb") as f:65            f.write(pkg_info)66        cmd = "rsync -atz %s %s/%s" % (self.pkg_info_pathfile, self.rsync_server, self.pkg_info_file)67        print "rsync pkg info file:%s, cmd:%s" % (self.pkg_info_pathfile, cmd)68        ret =os.system(cmd)69        if 0 <> ret:70            print "Failed to exec:%s" % (cmd)71            return False72        cmd = "rsync -atz %s %s/%s" % (self.pkg_pathfile, self.rsync_server, upload_pkg_file)73        print "rsync pkg file:%s to file:%s, cmd:%s" % (self.pkg_pathfile, upload_pkg_file, cmd)74        ret =os.system(cmd)75        if 0 <> ret:76            print "Failed to exec:%s" % (cmd)77            return False78        return True79    def __md5(self):80        """  pkg file's md581        """82        md5_obj = hashlib.md5()83        with open(self.pkg_pathfile, "rb") as f:84            while True:85                d = f.read(8096)86                if not d: break87                md5_obj.update(d)88        self.md5 = md5_obj.hexdigest().lower()89def main():90    """91      Running rsync upload pkg.92    """93    opt = OptionParser()94    opt.add_option("-u","--username",action="store",type="string",dest="user",95                   help="apollo's username.")96    opt.add_option("-a","--app", action="store", type="string", dest="app",97                   help="apollo's product name.")98    opt.add_option("-s", "--svr", action="store", type="string", dest="svr",99                   help="apollo's service name.")100    opt.add_option("-v", "--version", action="store", type="string", dest="version",101                   help="service's version.")102    opt.add_option("-d", "--directory", action="store", type="string", dest="directory",103                   help="service pkg's directory, it must be absolute path.")104    opt.add_option("-r", "--reupload", action="store_true", default="False", dest="reupload",105                   help="re-upload the pkg, default is false.")106    (options,args)=opt.parse_args()107    if not options.user or not len(options.user):108        print "Must specify the apollo user by -u, exit."109        exit(1)110    if not options.app or not len(options.app):111        print "Must specify the apollo product name by -a, exit."112        exit(1)113    if not options.svr or not len(options.svr):114        print "Must specify the apollo sevice name by -s, exit."115        exit(1)116    if not options.version or not len(options.version):117        print "Must specify the service's version by -v, exit."118        exit(1)119    if not options.directory or not len(options.directory):120        print "Must specify the service pkg's directory by -d, exit."121        exit(1)122    if not os.path.isdir(options.directory):123        print "Service pkg's directory[%s] doesn't exist, exit." % (options.directory)124        exit(1)125    if not options.directory.startswith("/"):126        print "Service pkg's directory[%s] isn't absolute path, exit." % (options.directory)127        exit(1)128        129    print "Begin to rsync, user:%s, product:%s, svr:%s, version:%s, directory:%s, reupload:%s" % \130        (options.user, options.app, options.svr, options.version, options.directory, options.reupload)131    try:132        rsync = ApolloRsyncPkg(options.user, options.app, options.svr, options.version, options.directory, options.reupload)133        if not rsync.rsync():134            exit(1)135    except Exception, e:136        print "Failed to upload, exception:%s" % (e)137        traceback.print_exc()138        exit(1)139    print "Success to rysnc."140    exit(0)141if __name__ == "__main__":...

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.