KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use get_install_dir method in autotest
Best Python code snippet using autotest_python
dse_node.py
Source:dse_node.py 
...22 self._dse_config_options = {}23 if self.cluster.hasOpscenter():24 self._copy_agent()25 def get_install_cassandra_root(self):26 return os.path.join(self.get_install_dir(), 'resources', 'cassandra')27 def get_node_cassandra_root(self):28 return os.path.join(self.get_path(), 'resources', 'cassandra')29 def get_conf_dir(self):30 """31 Returns the path to the directory where Cassandra config are located32 """33 return os.path.join(self.get_path(), 'resources', 'cassandra', 'conf')34 def get_tool(self, toolname):35 return common.join_bin(os.path.join(self.get_install_dir(), 'resources', 'cassandra'), 'bin', toolname)36 def get_tool_args(self, toolname):37 return [common.join_bin(os.path.join(self.get_install_dir(), 'resources', 'cassandra'), 'bin', 'dse'), toolname]38 def get_env(self):39 return common.make_dse_env(self.get_install_dir(), self.get_path(), self.ip_addr)40 def node_setup(self, version, verbose):41 dir, v = repository.setup_dse(version, self.cluster.dse_username, self.cluster.dse_password, verbose=verbose)42 return dir43 def set_workloads(self, workloads):44 self.workloads = workloads45 self._update_config()46 if 'solr' in self.workloads:47 self.__generate_server_xml()48 if 'graph' in self.workloads:49 self.__update_gremlin_config_yaml()50 if 'dsefs' in self.workloads:51 dsefs_options = {'dsefs_options': {'enabled': True,52 'work_dir': os.path.join(self.get_path(), 'dsefs'),53 'data_directories': [{'dir': os.path.join(self.get_path(), 'dsefs', 'data')}]}}54 self.set_dse_configuration_options(dsefs_options)55 if 'spark' in self.workloads:56 dsefs_enabled = 'dsefs' in self.workloads57 dse_options = {'dsefs_options': {'enabled': dsefs_enabled,58 'work_dir': os.path.join(self.get_path(), 'dsefs'),59 'data_directories': [{'dir': os.path.join(self.get_path(), 'dsefs', 'data')}]}}60 if self.cluster.version() >= '6.0':61 # Don't overwrite aoss options62 if 'resource_manager_options' not in self._dse_config_options:63 dse_options['resource_manager_options'] = {'worker_options': {'memory_total': '1g', 'cores_total': 2}}64 self.set_dse_configuration_options(dse_options)65 self._update_spark_env()66 def enable_aoss(self, thrift_port=10000, web_ui_port=9077):67 self._dse_config_options['alwayson_sql_options'] = {'enabled': True, 'thrift_port': thrift_port, 'web_ui_port': web_ui_port}68 self._dse_config_options['resource_manager_options'] = {'worker_options':69 {'cores_total': 2,70 'memory_total': '2g',71 'workpools': [{72 'name': 'alwayson_sql',73 'cores': 0.5,74 'memory': 0.575 }]}}76 self._update_config()77 self._update_spark_env()78 self._update_yaml()79 def set_dse_configuration_options(self, values=None):80 if values is not None:81 self._dse_config_options = common.merge_configuration(self._dse_config_options, values)82 self.import_dse_config_files()83 def watch_log_for_alive(self, nodes, from_mark=None, timeout=720, filename='system.log'):84 """85 Watch the log of this node until it detects that the provided other86 nodes are marked UP. This method works similarly to watch_log_for_death.87 We want to provide a higher default timeout when this is called on DSE.88 """89 super(DseNode, self).watch_log_for_alive(nodes, from_mark=from_mark, timeout=timeout, filename=filename)90 def get_launch_bin(self):91 cdir = self.get_install_dir()92 launch_bin = common.join_bin(cdir, 'bin', 'dse')93 # Copy back the dse scripts since profiling may have modified it the previous time94 shutil.copy(launch_bin, self.get_bin_dir())95 return common.join_bin(self.get_path(), 'bin', 'dse')96 def add_custom_launch_arguments(self, args):97 args.append('cassandra')98 for workload in self.workloads:99 if 'hadoop' in workload:100 args.append('-t')101 if 'solr' in workload:102 args.append('-s')103 if 'spark' in workload:104 args.append('-k')105 if 'cfs' in workload:106 args.append('-c')107 if 'graph' in workload:108 args.append('-g')109 def start(self,110 join_ring=True,111 no_wait=False,112 verbose=False,113 update_pid=True,114 wait_other_notice=True,115 replace_token=None,116 replace_address=None,117 jvm_args=None,118 wait_for_binary_proto=False,119 profile_options=None,120 use_jna=False,121 quiet_start=False,122 allow_root=False,123 set_migration_task=True,124 jvm_version=None):125 mark = self.mark_log()126 process = super(DseNode, self).start(join_ring, no_wait, verbose, update_pid, wait_other_notice, replace_token,127 replace_address, jvm_args, wait_for_binary_proto, profile_options, use_jna,128 quiet_start, allow_root, set_migration_task, jvm_version)129 if self.cluster.hasOpscenter():130 self._start_agent()131 def _start_agent(self):132 agent_dir = os.path.join(self.get_path(), 'datastax-agent')133 if os.path.exists(agent_dir):134 self._write_agent_address_yaml(agent_dir)135 self._write_agent_log4j_properties(agent_dir)136 args = [os.path.join(agent_dir, 'bin', common.platform_binary('datastax-agent'))]137 subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)138 def stop(self, wait=True, wait_other_notice=False, signal_event=signal.SIGTERM, **kwargs):139 if self.cluster.hasOpscenter():140 self._stop_agent()141 return super(DseNode, self).stop(wait=wait, wait_other_notice=wait_other_notice, signal_event=signal_event, **kwargs)142 def _stop_agent(self):143 agent_dir = os.path.join(self.get_path(), 'datastax-agent')144 if os.path.exists(agent_dir):145 pidfile = os.path.join(agent_dir, 'datastax-agent.pid')146 if os.path.exists(pidfile):147 with open(pidfile, 'r') as f:148 pid = int(f.readline().strip())149 f.close()150 if pid is not None:151 try:152 os.kill(pid, signal.SIGKILL)153 except OSError:154 pass155 os.remove(pidfile)156 def nodetool(self, cmd, username=None, password=None, capture_output=True, wait=True):157 if password is not None:158 cmd = '-pw {} '.format(password) + cmd159 if username is not None:160 cmd = '-u {} '.format(username) + cmd161 return super(DseNode, self).nodetool(cmd)162 def dsetool(self, cmd):163 env = self.get_env()164 extension.append_to_client_env(self, env)165 node_ip, binary_port = self.network_interfaces['binary']166 dsetool = common.join_bin(self.get_install_dir(), 'bin', 'dsetool')167 args = [dsetool, '-h', node_ip, '-j', str(self.jmx_port), '-c', str(binary_port)]168 args += cmd.split()169 p = subprocess.Popen(args, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)170 return handle_external_tool_process(p, args)171 def dse(self, dse_options=None):172 if dse_options is None:173 dse_options = []174 env = self.get_env()175 extension.append_to_client_env(self, env)176 env['JMX_PORT'] = self.jmx_port177 dse = common.join_bin(self.get_install_dir(), 'bin', 'dse')178 args = [dse]179 args += dse_options180 p = subprocess.Popen(args, env=env) #Don't redirect stdout/stderr, users need to interact with new process181 return handle_external_tool_process(p, args)182 def hadoop(self, hadoop_options=None):183 if hadoop_options is None:184 hadoop_options = []185 env = self.get_env()186 env['JMX_PORT'] = self.jmx_port187 dse = common.join_bin(self.get_install_dir(), 'bin', 'dse')188 args = [dse, 'hadoop']189 args += hadoop_options190 p = subprocess.Popen(args, env=env) #Don't redirect stdout/stderr, users need to interact with new process191 return handle_external_tool_process(p, args)192 def hive(self, hive_options=None):193 if hive_options is None:194 hive_options = []195 env = self.get_env()196 env['JMX_PORT'] = self.jmx_port197 dse = common.join_bin(self.get_install_dir(), 'bin', 'dse')198 args = [dse, 'hive']199 args += hive_options200 p = subprocess.Popen(args, env=env) #Don't redirect stdout/stderr, users need to interact with new process201 return handle_external_tool_process(p, args)202 def pig(self, pig_options=None):203 if pig_options is None:204 pig_options = []205 env = self.get_env()206 env['JMX_PORT'] = self.jmx_port207 dse = common.join_bin(self.get_install_dir(), 'bin', 'dse')208 args = [dse, 'pig']209 args += pig_options210 p = subprocess.Popen(args, env=env) #Don't redirect stdout/stderr, users need to interact with new process211 return handle_external_tool_process(p, args)212 def sqoop(self, sqoop_options=None):213 if sqoop_options is None:214 sqoop_options = []215 env = self.get_env()216 env['JMX_PORT'] = self.jmx_port217 dse = common.join_bin(self.get_install_dir(), 'bin', 'dse')218 args = [dse, 'sqoop']219 args += sqoop_options220 p = subprocess.Popen(args, env=env) #Don't redirect stdout/stderr, users need to interact with new process221 return handle_external_tool_process(p, args)222 def spark(self, spark_options=None):223 if spark_options is None:224 spark_options = []225 env = self.get_env()226 env['JMX_PORT'] = self.jmx_port227 dse = common.join_bin(self.get_install_dir(), 'bin', 'dse')228 args = [dse, 'spark']229 args += spark_options230 p = subprocess.Popen(args, env=env) #Don't redirect stdout/stderr, users need to interact with new process231 return handle_external_tool_process(p, args)232 def import_dse_config_files(self):233 self._update_config()234 if not os.path.isdir(os.path.join(self.get_path(), 'resources', 'dse', 'conf')):235 os.makedirs(os.path.join(self.get_path(), 'resources', 'dse', 'conf'))236 common.copy_directory(os.path.join(self.get_install_dir(), 'resources', 'dse', 'conf'), os.path.join(self.get_path(), 'resources', 'dse', 'conf'))237 self._update_yaml()238 def copy_config_files(self):239 for product in ['dse', 'cassandra', 'hadoop', 'hadoop2-client', 'sqoop', 'hive', 'tomcat', 'spark', 'shark', 'mahout', 'pig', 'solr', 'graph']:240 src_conf = os.path.join(self.get_install_dir(), 'resources', product, 'conf')241 dst_conf = os.path.join(self.get_path(), 'resources', product, 'conf')242 if not os.path.isdir(src_conf):243 continue244 if os.path.isdir(dst_conf):245 common.rmdirs(dst_conf)246 shutil.copytree(src_conf, dst_conf)247 if product == 'solr':248 src_web = os.path.join(self.get_install_dir(), 'resources', product, 'web')249 dst_web = os.path.join(self.get_path(), 'resources', product, 'web')250 if os.path.isdir(dst_web):251 common.rmdirs(dst_web)252 shutil.copytree(src_web, dst_web)253 if product == 'tomcat':254 src_lib = os.path.join(self.get_install_dir(), 'resources', product, 'lib')255 dst_lib = os.path.join(self.get_path(), 'resources', product, 'lib')256 if os.path.isdir(dst_lib):257 common.rmdirs(dst_lib)258 if os.path.exists(src_lib):259 shutil.copytree(src_lib, dst_lib)260 src_webapps = os.path.join(self.get_install_dir(), 'resources', product, 'webapps')261 dst_webapps = os.path.join(self.get_path(), 'resources', product, 'webapps')262 if os.path.isdir(dst_webapps):263 common.rmdirs(dst_webapps)264 shutil.copytree(src_webapps, dst_webapps)265 src_lib = os.path.join(self.get_install_dir(), 'resources', product, 'gremlin-console', 'conf')266 dst_lib = os.path.join(self.get_path(), 'resources', product, 'gremlin-console', 'conf')267 if os.path.isdir(dst_lib):268 common.rmdirs(dst_lib)269 if os.path.exists(src_lib):270 shutil.copytree(src_lib, dst_lib)271 def import_bin_files(self):272 common.copy_directory(os.path.join(self.get_install_dir(), 'bin'), self.get_bin_dir())273 cassandra_bin_dir = os.path.join(self.get_path(), 'resources', 'cassandra', 'bin')274 shutil.rmtree(cassandra_bin_dir, ignore_errors=True)275 os.makedirs(cassandra_bin_dir)276 common.copy_directory(os.path.join(self.get_install_dir(), 'resources', 'cassandra', 'bin'), cassandra_bin_dir)277 if os.path.exists(os.path.join(self.get_install_dir(), 'resources', 'cassandra', 'tools')):278 cassandra_tools_dir = os.path.join(self.get_path(), 'resources', 'cassandra', 'tools')279 shutil.rmtree(cassandra_tools_dir, ignore_errors=True)280 shutil.copytree(os.path.join(self.get_install_dir(), 'resources', 'cassandra', 'tools'), cassandra_tools_dir)281 self.export_dse_home_in_dse_env_sh()282 def export_dse_home_in_dse_env_sh(self):283 '''284 Due to the way CCM lays out files, separating the repository285 from the node(s) confs, the `dse-env.sh` script of each node286 needs to have its DSE_HOME var set and exported. Since DSE287 4.5.x, the stock `dse-env.sh` file includes a commented-out288 place to do exactly this, intended for installers.289 Basically: read in the file, write it back out and add the two290 lines.291 'sstableloader' is an example of a node script that depends on292 this, when used in a CCM-built cluster.293 '''294 with open(self.get_bin_dir() + "/dse-env.sh", "r") as dse_env_sh:295 buf = dse_env_sh.readlines()296 with open(self.get_bin_dir() + "/dse-env.sh", "w") as out_file:297 for line in buf:298 out_file.write(line)299 if line == "# This is here so the installer can force set DSE_HOME\n":300 out_file.write("DSE_HOME=" + self.get_install_dir() + "\nexport DSE_HOME\n")301 def _update_log4j(self):302 super(DseNode, self)._update_log4j()303 conf_file = os.path.join(self.get_conf_dir(), common.LOG4J_CONF)304 append_pattern = 'log4j.appender.V.File='305 log_file = os.path.join(self.get_path(), 'logs', 'solrvalidation.log')306 if common.is_win():307 log_file = re.sub("\\\\", "/", log_file)308 common.replace_in_file(conf_file, append_pattern, append_pattern + log_file)309 append_pattern = 'log4j.appender.A.File='310 log_file = os.path.join(self.get_path(), 'logs', 'audit.log')311 if common.is_win():312 log_file = re.sub("\\\\", "/", log_file)313 common.replace_in_file(conf_file, append_pattern, append_pattern + log_file)314 append_pattern = 'log4j.appender.B.File='315 log_file = os.path.join(self.get_path(), 'logs', 'audit', 'dropped-events.log')316 if common.is_win():317 log_file = re.sub("\\\\", "/", log_file)318 common.replace_in_file(conf_file, append_pattern, append_pattern + log_file)319 def _update_yaml(self):320 super(DseNode, self)._update_yaml()321 conf_file = os.path.join(self.get_path(), 'resources', 'dse', 'conf', 'dse.yaml')322 with open(conf_file, 'r') as f:323 data = yaml.safe_load(f)324 data['system_key_directory'] = os.path.join(self.get_path(), 'keys')325 # Get a map of combined cluster and node configuration with the node326 # configuration taking precedence.327 full_options = common.merge_configuration(328 self.cluster._dse_config_options,329 self._dse_config_options, delete_empty=False)330 # Merge options with original yaml data.331 data = common.merge_configuration(data, full_options)332 with open(conf_file, 'w') as f:333 yaml.safe_dump(data, f, default_flow_style=False)334 def __generate_server_xml(self):335 server_xml = os.path.join(self.get_path(), 'resources', 'tomcat', 'conf', 'server.xml')336 if os.path.isfile(server_xml):337 os.remove(server_xml)338 with open(server_xml, 'w+') as f:339 f.write('<Server port="8005" shutdown="SHUTDOWN">\n')340 f.write(' <Service name="Solr">\n')341 f.write(' <Connector port="8983" address="%s" protocol="HTTP/1.1" connectionTimeout="20000" maxThreads = "200" URIEncoding="UTF-8"/>\n' % self.ip_addr)342 f.write(' <Engine name="Solr" defaultHost="localhost">\n')343 f.write(' <Host name="localhost" appBase="../solr/web"\n')344 f.write(' unpackWARs="true" autoDeploy="true"\n')345 f.write(' xmlValidation="false" xmlNamespaceAware="false">\n')346 f.write(' </Host>\n')347 f.write(' </Engine>\n')348 f.write(' </Service>\n')349 f.write('</Server>\n')350 f.close()351 def __update_gremlin_config_yaml(self):352 conf_file = os.path.join(self.get_path(), 'resources', 'graph', 'gremlin-console', 'conf', 'remote.yaml')353 with open(conf_file, 'r') as f:354 data = yaml.safe_load(f)355 data['hosts'] = [self.ip_addr]356 with open(conf_file, 'w') as f:357 yaml.safe_dump(data, f, default_flow_style=False)358 def _get_directories(self):359 dirs = []360 for i in ['data', 'commitlogs', 'saved_caches', 'logs', 'bin', 'keys', 'resources', os.path.join('data', 'hints')]:361 dirs.append(os.path.join(self.get_path(), i))362 return dirs363 def _copy_agent(self):364 agent_source = os.path.join(self.get_install_dir(), 'datastax-agent')365 agent_target = os.path.join(self.get_path(), 'datastax-agent')366 if os.path.exists(agent_source) and not os.path.exists(agent_target):367 shutil.copytree(agent_source, agent_target)368 def _write_agent_address_yaml(self, agent_dir):369 address_yaml = os.path.join(agent_dir, 'conf', 'address.yaml')370 if not os.path.exists(address_yaml):371 with open(address_yaml, 'w+') as f:372 ip = self.ip_addr373 jmx = self.jmx_port374 f.write('stomp_interface: 127.0.0.1\n')375 f.write('local_interface: %s\n' % ip)376 f.write('agent_rpc_interface: %s\n' % ip)377 f.write('agent_rpc_broadcast_address: %s\n' % ip)378 f.write('cassandra_conf: %s\n' % os.path.join(self.get_path(), 'resources', 'cassandra', 'conf', 'cassandra.yaml'))...
install.py
Source:install.py
...125 for row in rows:126 download(package, row, provides)127 progress.tick()128 failed_dependency = False129 while (os.path.exists(locations.get_install_dir()) and130 not failed_dependency):131 provides, requires = install_sources(provides, package)132 for req in requires:133 if (not local_info.installed(req) and134 not local_info.provided(req)):135 if run(req, package):136 provides.append(req)137 else:138 failed_dependency = True139 shutil.rmtree(locations.get_install_dir())140 provides = list(set(provides))141 provides.sort()142 requires = []143 for fl in list(provides):144 try:145 requires.extend(get_requirements(fl))146 except IOError:147 provides.remove(fl)148 requires = list(set(requires))149 for pack in provides:150 name = re.sub(r'.*/(.*)\.(.*)', r'\1', pack)151 if name in requires:152 requires.remove(name)153 requires.sort()154 with open(locations.get_metadata_file(), 'a') as meta:155 meta.write('=='.join([package, version, '='.join(provides),156 '='.join(requires)]))157 meta.write('\n')158 for req in requires:159 if not local_info.installed(req) and not local_info.provided(req):160 if run(req, package):161 pass162 else:163 failed_dependency = True164 if os.path.exists(locations.get_discard_dir()):165 shutil.rmtree(locations.get_discard_dir())166 directory = locations.get_path(None, package)167 if not os.path.exists(directory):168 os.makedirs(directory)169 with open(os.path.join(directory, '.log'), 'w') as txt:170 for line in folder.diff:171 txt.write(line)172 txt.write('\n')173 if failed_dependency:174 print ('Package may have installed incorrectly. '175 'Error: a dependency was not found on CTAN.')176def install_sources(provides, package):177 old_files = []178 files = []179 while True:180 old_files.append(files)181 files = os.listdir(locations.get_install_dir())182 if files in old_files:183 break184 requires = []185 for f in files:186 if not os.path.isfile(os.path.join(locations.get_install_dir(),187 f)):188 continue189 if f.endswith('.dtx'):190 for other in files:191 if other.endswith('.ins'):192 continue193 elif other.endswith('.drv'):194 continue195 try:196 print ' building', f[:-4]197 subprocess.call(['tex', '-interaction=batchmode', f],198 stdout=open(os.devnull, 'w'),199 cwd=locations.get_install_dir())200 dest_dir = locations.get_path('sty', package)201 if not os.path.exists(dest_dir):202 os.makedirs(dest_dir)203 dest_file = f[:-3] + 'sty'204 orig = os.path.join(locations.get_install_dir(), dest_file)205 dest = os.path.join(dest_dir, dest_file)206 os.rename(orig, dest)207 for t in ('dtx', 'ins', 'tex'):208 fl = orig[:-3] + t209 if os.path.isfile(fl):210 os.remove(fl)211 provides.append(dest)212 except OSError:213 tex_dep = os.path.join(locations.get_install_dir(),214 f[:-3] + 'tex')215 if os.path.isfile(tex_dep):216 requires.extend(get_requirements(tex_dep))217 elif f.endswith('.drv'):218 subprocess.call(['latex', '-interaction=batchmode', f],219 stdout=open(os.devnull, 'w'),220 cwd=locations.get_install_dir())221 os.remove(os.path.join(locations.get_install_dir(), f))222 elif f.endswith('.dvi'):223 # subprocess.call(['dvipdfm', f], stdout=open(os.devnull, 'w'),224 # cwd=locations.get_install_dir())225 os.remove(os.path.join(locations.get_install_dir(), f))226 elif f.endswith('.ins'):227 local_requires = []228 local_requires.extend(get_requirements(os.path.join(229 locations.get_install_dir(), f)))230 tex_dep = os.path.join(locations.get_install_dir(),231 f[:-3] + 'tex')232 if os.path.isfile(tex_dep):233 local_requires.extend(get_requirements(tex_dep))234 for req in list(local_requires):235 if req == package:236 local_requires.remove(req)237 continue238 elif local_info.installed(req) or local_info.provided(req):239 local_requires.remove(req)240 continue241 for pack in provides:242 name = re.sub(r'.*/(.*)\.(.*)', r'\1.\2', pack)243 if req + '.sty' == name:244 local_requires.remove(req)245 break246 elif req == name:247 local_requires.remove(req)248 break249 if local_requires:250 requires.extend(local_requires)251 else:252 print ' building', f[:-4]253 subprocess.call(['latex', '-interaction=batchmode', f],254 stdout=open(os.devnull, 'w'),255 cwd=locations.get_install_dir())256 for line in open(os.path.join(locations.get_install_dir(),257 f), 'r').readlines():258 if line.startswith(r'\generate'):259 if 'file' in line:260 sty, dtx = re.sub(261 (r'\\generate.*\\file.*?\{(.*\..*)\}.*\\'262 r'from.*?\{(.*\.[^\}]*)\}.*', r'\1\t\2'),263 line).split('\t')264 sty = sty.replace('\n', '')265 dtx = dtx.replace('\n', '')266 filetype = sty[-3:]267 if filetype == 'tex':268 filetype = 'sty'269 dest_dir = locations.get_path(filetype,270 package)271 if not os.path.exists(dest_dir):272 os.makedirs(dest_dir)273 orig = os.path.join(274 locations.get_install_dir(), sty)275 dest = os.path.join(dest_dir, sty)276 os.rename(orig, dest)277 dtx_path = os.path.join(278 locations.get_install_dir(), dtx)279 if os.path.isfile(dtx_path):280 os.remove(dtx_path)281 if 'discard' not in dest_dir:282 provides.append(dest)283 os.remove(os.path.join(locations.get_install_dir(), f))284 elif f.endswith('.pdf'):285 dest_dir = locations.get_path('pdf', package)286 if not os.path.exists(dest_dir):287 os.makedirs(dest_dir)288 orig = os.path.join(locations.get_install_dir(), f)289 dest = os.path.join(dest_dir, f)290 os.rename(orig, dest)291 fl = orig[:-3] + 'tex'292 if os.path.isfile(fl):293 os.remove(fl)294 elif f.endswith('.sty'):295 dest_dir = locations.get_path('sty', package)296 if not os.path.exists(dest_dir):297 os.makedirs(dest_dir)298 orig = os.path.join(locations.get_install_dir(), f)299 dest = os.path.join(dest_dir, f)300 os.rename(orig, dest)301 provides.append(dest)302 install_dir = locations.get_install_dir()303 if not requires:304 for f in os.listdir(install_dir):305 orig = os.path.join(install_dir, f)306 if empty(orig):307 pass308 elif f.endswith('.tex'):309 dest_dir = locations.get_path('sty', package)310 if not os.path.exists(dest_dir):311 os.makedirs(dest_dir)312 dest = os.path.join(dest_dir, f)313 os.rename(orig, dest)314 else:315 dest_dir = locations.get_path(f[-3:], package)316 if not os.path.exists(dest_dir):...
deploy.py
Source:deploy.py
...14 ../../../revision15 :return: (str | None) cadena con la versión o None16 """17 revision = None18 revpath = os.path.join(get_install_dir(), 'revision')19 if is_production():20 if os.path.exists(revpath):21 try:22 with open(revpath , 'r') as f:23 revision = f.readline()24 revision = revision[:-1] if revision.endswith('\n') else revision25 except (IndexError, IOError):26 pass27 if revision:28 if debug: print >>sys.stderr, "Detectada revision {0}".format(revision)29 else:30 if debug: print >>sys.stderr, "No se ha detectado numero de revision".format(revision)31 else:32 if debug: print >>sys.stderr, "No se usa numero de revision en sistema de desarrollo".format(revision)33 return revision34def apply_intelligenia_settings_local(debug=False):35 """36 Sobrescribe valores definidos en el módulo intelligenia_settings con todos aquellos que estén 37 definidos en el módulo intelligenia_settings_local.38 39 Si falla la carga del módulo intelligenia_settings_local no hace nada.40 41 Normalmente este método se invocará desde settings.py de la siguiente manera:42 43 from intelligenia_util import deploy44 import intelligenia_settings45 deploy.apply_intelligenia_settings_local()46 Estas lÃneas deben estar al principio del fichero, antes de acceder a cualquier miembro del 47 módulo intelligenia_settings48 """49 import importlib50 # Cargar el módulo intelligenia_settings para poder sobrescribir sus definiciones51 try:52 package = os.environ.get("DJANGO_SETTINGS_MODULE").split(".")[0]53 intelligenia_settings = importlib.import_module(package + ".intelligenia_settings")54 except ImportError:55 if debug: print >>sys.stderr, "No se ha podido cargar el modulo intelligenia_settings"56 return57 # Sobrescribir definiciones del módulo intelligenia_settings con las que haya en en módulo58 # intelligenia_settings_local.intelligenia_local59 try:60 # Si el módulo no existe, se lanza la excepción ImportError y no se hace nada61 from intelligenia_local import intelligenia_settings_local62 if debug: print >>sys.stderr, "Aplicando definiciones locales en la configuracion"63 for i in dir(intelligenia_settings_local):64 if i.startswith("__"):65 continue66 if type( getattr(intelligenia_settings_local, i) ) is ModuleType:67 continue68 if not hasattr(intelligenia_settings, i):69 if debug: print >>sys.stderr, u"AVISO: intelligenia_settings no tiene el atributo {0}".format(unicode(i))70 continue71 try:72 v = getattr(intelligenia_settings_local, i)73 setattr(intelligenia_settings, i, v)74 if debug: print >>sys.stderr, u"Aplicada configuracion local: {0}\t\t{1}".format(unicode(i), v)75 except Exception as e:76 print e77 pass78 79 except ImportError:80 if debug: print >>sys.stderr, "No se ha podido cargar el modulo intelligenia_local.intelligenia_settings_local"81 return82def get_media_root(debug=False):83 """84 Obtiene el valor para MEDIA_ROOT apropiado, teniendo en cuenta si se trata de un sistema en explotación85 o de un sistema para desarrollo. En un sistema en explotación, el directorio MEDIA_ROOT quedará fuera 86 del directorio exportado, para evitar que se pierdan datos al desplegar nuevas versiones.87 88 Normalmente este método se invocará desde settings.py de la siguiente manera:89 90 from intelligenia_util import deploy91 ...92 ...93 MEDIA_ROOT = deploy.get_media_root()94 ...95 :param debug: (Boolean) Flag para indicar si se muestra mensajes de depuración96 :return: (str) Ruta97 """98 99 media_root = get_install_dir(debug=False) + "/media"100 if debug: print >>sys.stderr, "MEDIA_ROOT = {0}".format(media_root)101 return media_root102def get_install_dir(debug=False):103 """104 Obtiene la ruta absoluta al directorio donde está instalado el proyecto, Será el directorio que105 contenga "versions/" y otros en el sistema en producción o el directorio "src/" en un sistema de desarrollo106 :param debug: (Boolean) Flag para indicar si se muestra mensajes de depuración107 :return: (str) Ruta108 """109 current_path=os.path.dirname(os.path.realpath(__file__))110 dev_path = os.path.dirname(os.path.dirname(current_path + "/.."))111 production_path = os.path.dirname(os.path.dirname(os.path.dirname(current_path)))112 if is_production(debug=False):113 install_dir = production_path114 else:115 install_dir = dev_path116 if debug: print >>sys.stderr, "INSTALL_DIR = {0}".format(install_dir)117 return install_dir118def get_deployment_dir(debug=False):119 """120 Obtiene la ruta absoluta al directorio que contiene la versión desplegada o de desarrollo121 Para sistemas en explotación:122 Si existe el fichero "INSTALL_DIR/target", es que se acaba de completar una exportación123 y el enlace "INSTALL_DIR/current" aún no apunta al sitio adecuado, y se usa la ruta almacenada124 en el fichero 'target'. En otro caso se usa la ruta a la que apunta el enlace current125 Para sistemas de desarrollo:126 La ruta coincide con la de instalación127 :param debug: (Boolean) Flag para indicar si se muestra mensajes de depuración128 :return: (str) Ruta129 """130 if is_production(debug=False):131 try:132 with open(get_install_dir() + "/target", 'r') as t:133 deployed_dir = t.readline()134 if debug: print >>sys.stderr, "DEPLOYMENT_DIR = {0} [despliegue en proceso]".format(deployed_dir)135 except IOError:136 deployed_dir = get_install_dir() + '/current'137 if debug: print >>sys.stderr, "DEPLOYMENT_DIR = {0} -> {1} [despliegue completo]".format(deployed_dir, os.path.realpath(deployed_dir))138 else:139 deployed_dir = get_install_dir()140 if debug: print >>sys.stderr, "DEPLOYMENT_DIR = {0}".format(deployed_dir)141 return deployed_dir142def is_production(debug=False):143 """144 Devuelve True si se detecta que es la instalación de producción o False en caso contrario.145 Puede utilizarse para usar una configuracion personalizada según el entorno detectado, por146 ejemplo, en settings.py:147 148 from intelligenia_util import deploy149 ...150 ...151 DEBUG = not deploy.is_production()152 :param debug: (Boolean) Flag para indicar si se muestra mensajes de depuración153 """154 155 current_path=os.path.dirname(os.path.realpath(__file__))156 #dev_path = os.path.dirname(os.path.dirname(current_path + "/.."))157 production_path = os.path.dirname(os.path.dirname(os.path.dirname(current_path)))158 if os.path.isdir(production_path + "/intelligenia_local"):159 if debug: print >>sys.stderr, "Detectado entorno de produccion"160 return True161 else:162 if debug: print >>sys.stderr, "Detectado entorno de desarrollo"163 return False164def is_development(debug=False):165 """166 Devuelve True si se detecta que es un sistema de desarrollo o False en caso contrario167 Puede utilizarse para usar una configuracion personalizada según el entorno detectado, por168 ejemplo, en settings.py:169 170 from intelligenia_util import deploy171 ...172 ...173 DEBUG = deploy.is_development()174 :param debug: (Boolean) Flag para indicar si se muestra mensajes de depuración175 """176 177 return not is_production(debug)178def get_domainname_from_path(debug=False):179 """180 Devuelve el nombre de dominio, usando el basename de la ruta de instalación. Si la cadena obtenida no se parece a un181 nombre de dominio, devuelve None. El nombre de dominio debe ser un FQDN182 :param debug: (Boolean) Flag para indicar si se muestra mensajes de depuración183 :return: (str | None) cadena con el nombre de dominio o None184 """185 if is_development():186 if debug: print >>sys.stderr, "No se autodectecta nombre de dominio en sistema de desarrollo"187 return None188 basename = os.path.basename(get_install_dir())189 allowed = re.compile("(?=^.{4,253}$)(^((?!-)[a-zA-Z0-9-]{1,63}(?<!-)\.)+[a-zA-Z]{2,63}$)", re.IGNORECASE)190 if allowed.match(basename):191 if debug: print >>sys.stderr, "Nombre de dominio: '{0}'".format(basename)192 return basename193 else:194 if debug: print >>sys.stderr, "No se ha detectado nombre de dominio"195 return None196def file_exists_in_install_dir(filename):197 """198 Devuelve True o False dependiendo de si un determinado fichero se puede encontrar a partir de la ruta de instalación199 :param filename: (str) ruta al fichero a partir del directorio de instalación200 :return: (boolean) True si el fichero existe o False en caso contrario201 """202 return os.path.exists(os.path.join(get_install_dir(), filename))203# ------------------------------------------------------------------------204# Inicialización de constantes205# ------------------------------------------------------------------------206# Estas variables almacenan el resultado de los diferentes métodos. Todos los métodos son estables para una determinada207# ejecución, por lo que los valores devueltos siempre serán los mismos y se pueden considerar constantes208SHOW_DEBUG_INFO=True209if SHOW_DEBUG_INFO:210 print "------- info instalacion (deploy.py) ----------------------------------"211# Flags212IS_DEVELOPMENT = is_development(debug=SHOW_DEBUG_INFO)213IS_PRODUCTION = is_production()214# Rutas215INSTALL_DIR = get_install_dir(debug=SHOW_DEBUG_INFO)216DEPLOYMENT_DIR = get_deployment_dir(debug=SHOW_DEBUG_INFO)217MEDIA_ROOT = get_media_root(debug=SHOW_DEBUG_INFO)218# Otras constantes219REVISION = get_revision(debug=SHOW_DEBUG_INFO)220DOMAIN_NAME = get_domainname_from_path(debug=SHOW_DEBUG_INFO)221if SHOW_DEBUG_INFO:...
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 autotest 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?
