Best Python code snippet using localstack_python
install_erpnext.py
Source:install_erpnext.py  
1# Copyright (c) 2013, Web Notes Technologies Pvt. Ltd. and Contributors2# License: GNU General Public License v3. See license.txt3#!/usr/bin/env python4from __future__ import unicode_literals5import os, sys6import argparse7import subprocess8is_redhat = is_debian = None9root_password = None10requirements = [11	"chardet", 12	"cssmin", 13	"dropbox", 14	"google-api-python-client", 15	"gunicorn", 16	"httplib2", 17	"jinja2", 18	"markdown2", 19	"markupsafe", 20	"mysql-python",21	"pygeoip", 22	"python-dateutil", 23	"python-memcached", 24	"pytz==2013d", 25	"requests", 26	"six", 27	"slugify", 28	"termcolor", 29	"werkzeug",30	"semantic_version",31	"gitpython==0.3.2.RC1"32]33def install(install_path):34	setup_folders(install_path)35	install_erpnext(install_path)36	37	post_install(install_path)38	39def install_pre_requisites():40	global is_redhat, is_debian41	is_redhat, is_debian = validate_install()42	if is_redhat:43		install_using_yum()44	elif is_debian:45		install_using_apt()46		47	install_python_modules()48	49	print "-"*8050	print "Pre-requisites Installed"51	print "-"*8052def validate_install():53	import platform54	# check os55	operating_system = platform.system()56	print "Operating System =", operating_system57	if operating_system != "Linux":58		raise Exception, "Sorry! This installer works only for Linux based Operating Systems"59	60	# check python version61	python_version = sys.version.split(" ")[0]62	print "Python Version =", python_version63	if not (python_version and int(python_version.split(".")[0])==2 and int(python_version.split(".")[1]) >= 7):64		raise Exception, "Hey! ERPNext needs Python version to be 2.7+"65	66	# check distribution67	distribution = platform.linux_distribution()[0].lower().replace('"', '')68	print "Distribution = ", distribution69	is_redhat = distribution in ("redhat", "red hat enterprise linux server", "centos", "centos linux", "fedora")70	is_debian = distribution in ("debian", "ubuntu", "elementary os", "linuxmint")71	72	if not (is_redhat or is_debian):73		raise Exception, "Sorry! This installer works only with yum or apt-get package management"74	75	return is_redhat, is_debian76		77def install_using_yum():78	packages = "gcc MySQL-python git memcached ntp vim-enhanced screen"79	80	print "-"*8081	print "Installing Packages: (This may take some time)"82	print packages83	print "-"*8084	exec_in_shell("yum install -y %s" % packages)85	86	try:87		exec_in_shell("which mysql")88	except subprocess.CalledProcessError:89		packages = "mysql mysql-server mysql-devel"90		print "Installing Packages:", packages91		exec_in_shell("yum install -y %s" % packages)92		exec_in_shell("service mysqld restart")93		94		# set a root password post install95		global root_password96		print "Please create a password for root user of MySQL"97		root_password = (get_root_password() or "erpnext").strip()98		exec_in_shell('mysqladmin -u root password "%s"' % (root_password,))99		print "Root password set as", root_password100	101	update_config_for_redhat()102	103def update_config_for_redhat():104	import re105	106	# set to autostart on startup107	for service in ("mysqld", "memcached"):108		exec_in_shell("chkconfig --level 2345 %s on" % service)109		exec_in_shell("service %s restart" % service)110	111def install_using_apt():112	exec_in_shell("apt-get update")113	packages = "python python-setuptools python-dev build-essential python-mysqldb git memcached ntp vim screen htop"114	print "-"*80115	print "Installing Packages: (This may take some time)"116	print packages117	print "-"*80118	exec_in_shell("apt-get install -y %s" % packages)119	global root_password120	if not root_password:121		root_password = get_root_password()122	exec_in_shell("echo mysql-server mysql-server/root_password password %s | sudo debconf-set-selections" % root_password)123	exec_in_shell("echo mysql-server mysql-server/root_password_again password %s | sudo debconf-set-selections" % root_password)124	125	try:126		exec_in_shell("which mysql")127	except subprocess.CalledProcessError:128		packages = "mysql-server libmysqlclient-dev"129		print "Installing Packages:", packages130		exec_in_shell("apt-get install -y %s" % packages)131	132	update_config_for_debian()133	134def update_config_for_debian():135	for service in ("mysql",):136		exec_in_shell("service %s restart" % service)137	138def install_python_modules():139	print "-"*80140	print "Installing Python Modules: (This may take some time)"141	print "-"*80142	143	try:144		exec_in_shell("which pip2.7")	145	except subprocess.CalledProcessError:146		exec_in_shell("easy_install-2.7 pip")147	148	exec_in_shell("pip2.7 install --upgrade setuptools --no-use-wheel")149	exec_in_shell("pip2.7 install --upgrade setuptools")150	exec_in_shell("pip2.7 install {}".format(' '.join(requirements)))151	152def install_erpnext(install_path):153	print154	print "-"*80155	print "Installing ERPNext"156	print "-"*80157	158	# ask for details159	global root_password160	if not root_password:161		root_password = get_root_password()162		test_root_connection(root_password)163	164	db_name = raw_input("ERPNext Database Name: ")165	if not db_name:166		raise Exception, "Sorry! You must specify ERPNext Database Name"167	168	# setup paths169	sys.path = [".", "lib", "app"] + sys.path170	import wnf171	172	# install database, run patches, update schema173	# setup_db(install_path, root_password, db_name)174	wnf.install(db_name, root_password=root_password)175	setup_cron(install_path)176	177def get_root_password():178	# ask for root mysql password179	import getpass180	root_pwd = None181	root_pwd = getpass.getpass("MySQL Root user's Password: ")182	return root_pwd183	184def test_root_connection(root_pwd):185	out = exec_in_shell("mysql -u root %s -e 'exit'" % \186		(("-p"+root_pwd) if root_pwd else "").replace('$', '\$').replace(' ', '\ '))187	if "access denied" in out.lower():188		raise Exception("Incorrect MySQL Root user's password")189		190def setup_folders(install_path):191	os.chdir(install_path)192	app = os.path.join(install_path, "app")193	if not os.path.exists(app):194		print "Cloning erpnext"195		exec_in_shell("cd %s && git clone --branch master https://github.com/nihadnagi/ERPNino.git app" % install_path)196		exec_in_shell("cd app && git config core.filemode false")197		if not os.path.exists(app):198			raise Exception, "Couldn't clone erpnext repository"199	200	lib = os.path.join(install_path, "lib")201	if not os.path.exists(lib):202		print "Cloning wnframework"203		exec_in_shell("cd %s && git clone --branch master https://github.com/webnotes/wnframework.git lib" % install_path)204		exec_in_shell("cd lib && git config core.filemode false")205		if not os.path.exists(lib):206			raise Exception, "Couldn't clone wnframework repository"207	208	public = os.path.join(install_path, "public")209	for p in [public, os.path.join(public, "files"), os.path.join(public, "backups"),210		os.path.join(install_path, "logs")]:211			if not os.path.exists(p):212				os.mkdir(p)213				214def setup_conf(install_path, db_name):215	import os, string, random, re216	# generate db password217	char_range = string.ascii_letters + string.digits218	db_password = "".join((random.choice(char_range) for n in xrange(16)))219	220	# make conf file221	with open(os.path.join(install_path, "lib", "conf", "conf.py"), "r") as template:222		conf = template.read()223	224	conf = re.sub("db_name.*", 'db_name = "%s"' % (db_name,), conf)225	conf = re.sub("db_password.*", 'db_password = "%s"' % (db_password,), conf)226	227	with open(os.path.join(install_path, "conf.py"), "w") as conf_file:228		conf_file.write(conf)229	230	return db_password231	232def post_install(install_path):233	pass234def exec_in_shell(cmd):235	# using Popen instead of os.system - as recommended by python docs236	import subprocess237	out = subprocess.check_output(cmd, shell=True)238	return out239def parse_args():240	parser = argparse.ArgumentParser()241	parser.add_argument('--create_user', default=False, action='store_true')242	parser.add_argument('--username', default='erpnext')243	parser.add_argument('--password', default='erpnext')244	parser.add_argument('--no_install_prerequisites', default=False, action='store_true')245	return parser.parse_args()246def create_user(username, password):247	import subprocess, pwd248	p = subprocess.Popen("useradd -m -d /home/{username} -s {shell} {username}".format(username=username, shell=os.environ.get('SHELL')).split())249	p.wait()250	p = subprocess.Popen("passwd {username}".format(username=username).split(), stdin=subprocess.PIPE)251	p.communicate('{password}\n{password}\n'.format(password=password))252	p.wait()253	return pwd.getpwnam(username).pw_uid254def setup_cron(install_path):255	erpnext_cron_entries = [256		"*/3 * * * * cd %s && python2.7 lib/wnf.py --run_scheduler >> erpnext-sch.log 2>&1" % install_path,257		"0 */6 * * * cd %s && python2.7 lib/wnf.py --backup >> erpnext-backup.log 2>&1" % install_path258		]259	for row in erpnext_cron_entries:260		try:261			existing_cron = exec_in_shell("crontab -l")262			if row not in existing_cron:263				exec_in_shell('{ crontab -l; echo "%s"; } | crontab' % row)264		except:265			exec_in_shell('echo "%s" | crontab' % row)266if __name__ == "__main__":267	args = parse_args()268	install_path = os.getcwd()269	if os.getuid() != 0 and args.create_user and not args.no_install_prequisites:270		raise Exception, "Please run this script as root"271	if args.create_user:272		uid = create_user(args.username, args.password)273		install_path = '/home/{username}/erpnext'.format(username=args.username)274	if not args.no_install_prerequisites:275		install_pre_requisites()276	if os.environ.get('SUDO_UID') and not args.create_user:277		os.setuid(int(os.environ.get('SUDO_UID')))278	279	if os.getuid() == 0 and args.create_user:280		os.setuid(uid)281		if install_path:282			os.mkdir(install_path)283	284	install(install_path=install_path)285	print286	print "-"*80287	print "Installation complete"288	print "To start the development server,"289	print "Login as {username} with password {password}".format(username=args.username, password=args.password)290	print "cd {}".format(install_path)291	print "./lib/wnf.py --serve"292	print "-"*80293	print "Open your browser and go to http://localhost:8000"...upload_history_gatherer.py
Source:upload_history_gatherer.py  
1# Last-Modified: <Sun Aug 17 12:13:02 2008>2# This file is part of the Ultimate Debian Database Project3from gatherer import gatherer4import aux5from glob import glob6import gzip7import psycopg28import sys9import os.path10def get_gatherer(config, connection, source):11  return upload_history_gatherer(config, connection, source)12class upload_history_gatherer(gatherer):13  def __init__(self, connection, config, source):14    self.is_ubuntu = source == 'ubuntu-upload-history'15    self.is_debian = not self.is_ubuntu16    gatherer.__init__(self, connection, config, source)17    if not 'path' in self.my_config:18      raise aux.ConfigException('path not specified for source ' + source)19  def run(self):20    path = self.my_config['path']21    if 'only-recent' in self.my_config:22      onlyrecent = self.my_config['only-recent']23    else:24      onlyrecent = True25    cursor = self.cursor()26    tables = ['source', 'version', 'date', 'changed_by', 'changed_by_name', 'changed_by_email', 'maintainer', 'maintainer_name', 'maintainer_email', 'nmu', 'signed_by', 'signed_by_name', 'signed_by_email', 'key_id', 'fingerprint', 'distribution', 'file']27    if self.is_ubuntu:28      tables = tables + ['original_maintainer', 'original_maintainer_name', 'original_maintainer_email', 'component']29    indices = ', '.join(map(lambda x: '$' + str(x), range(1,len(tables)+1)))30    tables = ', '.join(tables)31    cursor.execute("PREPARE uh_insert AS INSERT INTO %s (%s) VALUES \32      (%s)" % (self.my_config['table'], tables, indices))33    if self.is_debian:34      cursor.execute("PREPARE uh_arch_insert AS INSERT INTO %s (source, \35        version, architecture, file) VALUES \36        ($1, $2, $3, $4)" % (self.my_config['table'] + '_architecture'))37    if self.is_ubuntu:38      cursor.execute("PREPARE uh_launchpad_close_insert AS INSERT INTO %s (source, version, bug, file) \39          VALUES ($1, $2, $3, $4)" % (self.my_config['table'] + '_launchpad_closes'))40    41    cursor.execute("PREPARE uh_close_insert AS INSERT INTO %s (source, version, bug, file) \42        VALUES ($1, $2, $3, $4)" % (self.my_config['table'] + '_closes'))43    query = "EXECUTE uh_insert(%(Source)s, %(Version)s, %(Message-Date)s, \44      %(Changed-By)s, %(Changed-By_name)s, %(Changed-By_email)s, \45      %(Maintainer)s, %(Maintainer_name)s, %(Maintainer_email)s, %(NMU)s, \46      %(Signed-By)s, %(Signed-By_name)s, %(Signed-By_email)s, %(Key)s, \47      %(Fingerprint)s, %(Distribution)s, %(File)s)"48    if self.is_ubuntu:49        query = query[:-1] + ", %(Original-Maintainer)s, %(Original-Maintainer_name)s, %(Original-Maintainer_email)s, %(Component)s)"50        51    if self.is_debian:52        query_archs = "EXECUTE uh_arch_insert(%(Source)s, %(Version)s, %(arch)s, %(File)s)"53    query_closes = "EXECUTE uh_close_insert(%(Source)s, %(Version)s, %(closes)s, %(File)s)"54    if self.is_ubuntu:55        query_launchpad_closes = "EXECUTE uh_launchpad_close_insert(%(Source)s, %(Version)s, %(closes)s, %(File)s)"56    added = {}57    files = glob(path + '/*-changes*')58    files.sort()59    if onlyrecent:60      files = files[-2:]61    else:62      print "Doing full import!"63      if self.is_debian:64        cursor.execute("delete from " + self.my_config['table'] + "_architecture")65      if self.is_ubuntu:66        cursor.execute("delete from " + self.my_config['table'] + "_launchpad_closes")67      cursor.execute("delete from " + self.my_config['table'] + "_closes")68      cursor.execute("delete from " + self.my_config['table'])69    for name in files:70      bname = os.path.basename(name).replace(".gz","").replace(".out","")71      if self.is_debian:72        cursor.execute("DELETE FROM " + self.my_config['table'] + "_architecture where file='%s'" % (bname))73      if self.is_ubuntu:74        cursor.execute("DELETE FROM " + self.my_config['table'] + "_launchpad_closes where file='%s'" % (bname))75      cursor.execute("DELETE FROM " + self.my_config['table'] + "_closes where file='%s'" % (bname))76      cursor.execute("DELETE FROM " + self.my_config['table'] +  " where file='%s'" % (bname))77      f = None78      if name.endswith(".gz"):79        f = gzip.open(name)80      else:81        f = open(name)82      current = {}83      current['Fingerprint'] = 'N/A' # hack: some entries don't have fp84      current['NMU'] = False85      current['Key'] = ''86      current['File'] = bname87      last_field = None88      line_count = 089      uploads = []90      uploads_archs = []91      uploads_closes = []92      uploads_launchpad_closes = []93      for line in f:94        line_count += 195        line = line.lstrip()96        # Stupid multi-line maintainer fields *grml*97        if line == '':98          current['Changed-By_name'], current['Changed-By_email'] = aux.parse_email(current['Changed-By'])99          current['Maintainer_name'], current['Maintainer_email'] = aux.parse_email(current['Maintainer'])100          current['Signed-By_name'], current['Signed-By_email'] = aux.parse_email(current['Signed-By'])101          if current['Signed-By_name'] == '':102            current['Signed-By_name'] = current['Signed-By']103            current['Signed-By_email'] = ''104          if current.has_key('Original-Maintainer'):105            if current['Original-Maintainer'] != 'N/A':106              current['Original-Maintainer_name'], current['Original-Maintainer_email'] = aux.parse_email(current['Original-Maintainer'])107            else:108              current['Original-Maintainer_name'] = current['Original-Maintainer_email'] = 'N/A'109          if (not current.has_key('Message-Date')) and (current.has_key('Date')):110            current['Message-Date'] = current['Date']111          current['Message-Date'] = current['Message-Date'].partition('(')[0].replace('+4200','+0000').replace('+4300','+0000').replace('+4100','+0000').replace('+4400','+0000').replace('+4000','+0000')112          if (current['Source'], current['Version']) in added or \113            (current['Source'], current['Version']) == ('libapache-authznetldap-perl', '0.07-4') or \114            (current['Source'], current['Version']) == ('knj10font', '1.01-1') or \115            (current['Source'], current['Version']) == ('xmorph', '1:20010421') or \116            current['Message-Date'] == 'None':117              print "Skipping upload: "+current['Source']+" "+current['Version']+" "+current['Date']118              current = {}119              current['Fingerprint'] = 'N/A' # hack: some entries don't have fp120              current['NMU'] = False121              current['Key'] = ''122              current['File'] = bname123              last_field = None124              continue125          added[(current['Source'], current['Version'])] = True126          uploads.append(current)127          if self.is_debian:128            for arch in set(current['Architecture'].split()):129              current_arch = {'Source': current['Source'], 'Version': current['Version'], 'File': bname} 130              current_arch['arch'] = arch131              uploads_archs.append(current_arch)132          if current['Closes'] != 'N/A':133            for closes in set(current['Closes'].split()):134              current_closes = {'Source': current['Source'], 'Version': current['Version'], 'File': bname} 135              current_closes['closes'] = closes136              uploads_closes.append(current_closes)137          if current.has_key('Launchpad-Bugs-Fixed') and current['Launchpad-Bugs-Fixed'] != 'N/A':138            for closes in set(current['Launchpad-Bugs-Fixed'].split()):139              current_closes = {'Source': current['Source'], 'Version': current['Version'], 'File': bname} 140              current_closes['closes'] = closes141              uploads_launchpad_closes.append(current_closes)142          current = {}143          current['Fingerprint'] = 'N/A' # hack: some entries don't have fp144          current['NMU'] = False145          current['Key'] = ''146          current['File'] = bname147          last_field = None148          continue149        if line.find(':') == -1:150          if not last_field:151            raise Exception, "Format error on line " + line_count + "of file " + name152          current[last_field] += line153          continue154        (field, data) = line.split(':', 1)155        data = data.strip()156        current[field] = data157        158        last_field = field159      #print uploads160      #for u in uploads:161      #  print u162      #  cursor.execute(query, u)163      cursor.executemany(query, uploads)164      if self.is_debian:165        cursor.executemany(query_archs, uploads_archs)166      if self.is_ubuntu:167        cursor.executemany(query_launchpad_closes, uploads_launchpad_closes)168      cursor.executemany(query_closes, uploads_closes)169      170    cursor.execute("DEALLOCATE uh_insert")171    if self.is_debian:172      cursor.execute("ANALYZE " + self.my_config['table'] + '_architecture')173    if self.is_ubuntu:174      cursor.execute("ANALYZE " + self.my_config['table'] + '_launchpad_closes')175    cursor.execute("ANALYZE " + self.my_config['table'] + '_closes')...shared.py
Source:shared.py  
1from offregister_fab_utils.apt import apt_depends2from offregister_fab_utils.fs import cmd_avail3def install(c):4    """5    :param c: Connection6    :type c: ```fabric.connection.Connection```7    """8    if cmd_avail(c, "certbot"):9        return "certbot already installed"10    uname = c.run("uname -v").stdout.rstrip()11    dist_version = float(c.run("lsb_release -rs", hide=True).stdout.rstrip())12    is_debian = "Debian" in uname13    is_ubuntu = "Ubuntu" in uname14    if is_debian or is_ubuntu:15        apt_depends(16            c,17            "certbot",18            "python{}-certbot-nginx".format(19                "3"20                if is_ubuntu21                and dist_version >= 20.0422                or is_debian23                and dist_version >= 1024                else ""25            ),26        )27    else:28        raise NotImplementedError()...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
