def fetch_sources_from_repo(branch, code_directory):
if exists(code_directory):
print('Removing the following directory: %s' % code_directory)
sudo('rm -rf %s' % code_directory)
git_clone_command = 'git clone {1} {2} --branch {0} --single-branch'
sudo(git_clone_command.format(branch, REPOSITORY_URL, code_directory))
python类exists()的实例源码
def create_dhparam_if_necessary():
if exists(DHPARAM_PATH):
print('dhparam file exists, skipping this step')
return
sudo('openssl dhparam -out %s 2048' % DHPARAM_PATH)
def create_ssl_params_if_necessary():
create_dhparam_if_necessary()
if exists(SSL_PARAMS_PATH):
print('Not creating ssl-params.conf, already exists')
return
upload_template(
filename='deploy_configs/ssl_params',
destination=SSL_PARAMS_PATH,
context={'dhparam_path': DHPARAM_PATH},
use_sudo=True
)
def configure_letsencrypt_if_necessary():
create_ssl_params_if_necessary()
env.letsencrypt_folder = os.path.join('/etc/letsencrypt/live', env.domain_name)
print('Assuming letsencrypt folder is %s' % env.letsencrypt_folder)
if exists(env.letsencrypt_folder, use_sudo=True):
print('letsencrypt folder found, skipping letsencrypt setup')
return
start_letsencrypt_setup()
def exists(path, remote=True):
'''
Check if the path exists in the remote or locally,
depending upon the `remote` parameter.
'''
if remote:
return files.exists(path)
return os.path.exists(path)
def _setup_ros_packages_from_git(ros_package_name, git_url, subpackage_list):
run("echo 'Starting...'")
home_path = run("pwd")
git_path = home_path + "/gitspace"
ros_package_path = git_path + "/" + ros_package_name #"/rosbots"
ws_dir = home_path + WS_DIR
install_dir = home_path + INSTALL_DIR
_fp("Do we need to create gitspace folder?")
if not fabfiles.exists(git_path):
run("mkdir " + git_path)
_fp("Do we need to git clone the repo?")
if not fabfiles.exists(ros_package_path):
_fp("Did not find " + ros_package_name + " repo, cloning...")
with cd(git_path):
run("git clone " + git_url)
_fp("Creating symbolic link to main ros workspace")
with cd(ws_dir + "/src"):
if fabfiles.exists(ros_package_name):
run("rm " + ros_package_name)
run("ln -s " + ros_package_path)
else:
#_fp("Found the repo, just fetching top and rebasing")
#with cd(ros_package_path):
# run("git fetch origin")
# run("git rebase origin/master")
_pp("Found the repo, not doing anything - feel free to git fetch and rebase manually")
for subpackage in subpackage_list:
_fp("Compiling " + subpackage + "...")
with cd(ws_dir):
run("./src/catkin/bin/catkin_make_isolated --pkg " + subpackage + " --install -DCMAKE_BUILD_TYPE=Release --install-space " + install_dir + " -j1")
def _setup_ros_other_packages(rospkg, run_rosdep=True):
run("echo 'Starting...'")
home_path = run("pwd")
ws_dir = home_path + WS_DIR
if not fabfiles.exists(ws_dir):
_fp("ROS Workspace not found - run the main set up first")
return
with cd(ws_dir):
ts = str(time.time()).split(".")[0]
fn = "kinetic-custom_" + str(ts) + "_ros.rosinstall"
run("rosinstall_generator " + rospkg + " --rosdistro kinetic --deps --wet-only --tar > " + fn)
run("cat " + fn)
_pp("Did rosinstall generator create the install file correctly? If so, we're going to merge and update the workspace. (If there are duplicate packages, hit DELETE and REPLACE!)")
run("wstool merge -y -t src " + fn)
_pp("Did the wstool merge correctly? If so, we are going to update on the install file for the workspace.")
run("wstool update --delete-changed-uris -t src")
_pp("Did the wstool update correctly? If so, we are going to update dependencies.")
if run_rosdep:
run("rosdep install --from-paths src --ignore-src --rosdistro kinetic -y -r --os=debian:jessie")
_pp("Did the dependencies update ok? If so, let's compile the new packages.")
run("./src/catkin/bin/catkin_make_isolated --install -DCMAKE_BUILD_TYPE=Release --install-space " + home_path + INSTALL_DIR + " -j1")
def verify_reboot_performed(self, max_wait=60*30):
logger.info("waiting for system to reboot")
successful_connections = 0
timeout = time.time() + max_wait
while time.time() <= timeout:
try:
with settings(warn_only=True, abort_exception=FabricFatalException):
time.sleep(1)
if exists(self.tfile):
logger.debug("temp. file still exists, device hasn't rebooted.")
continue
else:
logger.debug("temp. file no longer exists, device has rebooted.")
successful_connections += 1
# try connecting 10 times before returning
if successful_connections <= 9:
continue
return
except (BaseException):
logger.debug("system exit was caught, this is probably because SSH connectivity is broken while the system is rebooting")
continue
if time.time() > timeout:
pytest.fail("Device never rebooted!")
def verify_reboot_not_performed(self, wait=60):
time.sleep(wait)
assert exists(self.tfile)
def load_db_dump(dump_file):
"""Given a dump on your home dir on the server, load it to the server's
database, overwriting any existing data. BE CAREFUL!"""
require('environment')
if not files.exists("%(home)s/.pgpass" % env):
abort("Please get a copy of .pgpass and put it in your home dir")
temp_file = os.path.join(env.home, '%(project)s-%(environment)s.sql' % env)
put(dump_file, temp_file)
run('psql -h %s -U %s -d %s -f %s' % (env.db_host, env.db_user, env.db, temp_file))
def make_messages():
"""Extract English text from code and templates, and update the
.po files for translators to translate"""
# Make sure gettext is installed
local("gettext --help >/dev/null 2>&1")
if os.path.exists("locale/fr/LC_MESSAGES/django.po"):
local("python manage.py makemessages -a")
else:
local("python manage.py makemessages -l fr")
def mkdir_or_backup(appname):
appfolder = applications+'/'+appname
if not exists(appfolder):
sudo('mkdir %s' % appfolder)
sudo('chown -R www-data:www-data %s' % appfolder)
backup = None
else:
dt = now.strftime('%y-%m-%d-%h-%m')
backup = '%s.%s.zip' % (appname, dt)
with cd(applications):
sudo('zip -r %s %s' % (backup, appname))
return backup
