Best Python code snippet using autotest_python
webdav.py
Source:webdav.py
...102 protocol = 'http'103 self.baseuri = '%s://%s:%s/' % (protocol, interface or104 socket.gethostname(), port)105 self.verbose = False106 def _log_exception(self, exception):107 if CONFIG['verbose'] and not isinstance(exception, (108 NotLogged, ConcurrencyException, UserError, UserWarning,109 DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden)):110 tb_s = ''.join(traceback.format_exception(*sys.exc_info()))111 logger = logging.getLogger('webdav')112 logger.error('Exception:\n' + tb_s)113 @staticmethod114 def get_dburi(uri):115 uri = urlparse.urlsplit(uri)[2]116 if uri and uri[0] == '/':117 uri = uri[1:]118 dbname, uri = (uri.split('/', 1) + [None])[0:2]119 if dbname:120 dbname = urllib.unquote_plus(dbname)121 if uri:122 uri = urllib.unquote_plus(uri)123 return dbname, uri124 def _get_dburi(self, uri):125 return TrytonDAVInterface.get_dburi(uri)126 def get_childs(self, uri, filter=None):127 res = []128 dbname, dburi = self._get_dburi(uri)129 if not dbname:130 database = Database().connect()131 cursor = database.cursor()132 try:133 lists = database.list(cursor)134 except Exception:135 lists = []136 finally:137 cursor.close()138 for dbname in lists:139 res.append(urlparse.urljoin(uri, dbname))140 return res141 pool = Pool(Transaction().cursor.database_name)142 try:143 Collection = pool.get('webdav.collection')144 scheme, netloc, path, params, query, fragment = \145 urlparse.urlparse(uri)146 if path[-1:] != '/':147 path += '/'148 for child in Collection.get_childs(dburi, filter=filter,149 cache=CACHE):150 res.append(urlparse.urlunparse((scheme, netloc,151 path + child.encode('utf-8'), params, query,152 fragment)))153 except KeyError:154 return res155 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:156 self._log_exception(exception)157 raise158 except Exception, exception:159 self._log_exception(exception)160 raise DAV_Error(500)161 return res162 def get_data(self, uri, range=None):163 dbname, dburi = self._get_dburi(uri)164 if not dbname or (self.exists(uri) and self.is_collection(uri)):165 res = ('<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.0 '166 'Transitional//EN">')167 res += '<html>'168 res += '<head>'169 res += ('<meta http-equiv="Content-Type" content="text/html; '170 'charset=utf-8">')171 res += ('<title>%s - WebDAV - %s</title>'172 % (PACKAGE, dbname or 'root'))173 res += '</head>'174 res += '<body>'175 res += '<h2>Collection: %s</h2>' % (get_urifilename(uri) or '/')176 res += '<ul>'177 if dbname:178 scheme, netloc, path, params, query, fragment = \179 urlparse.urlparse(uri)180 if path[-1:] != '/':181 path += '/'182 res += ('<li><a href="%s">..</a></li>'183 % urlparse.urlunparse((scheme, netloc, path + '..',184 params, query, fragment)))185 childs = self.get_childs(uri)186 childs.sort()187 for child in childs:188 res += ('<li><a href="%s">%s</a></li>'189 % (quote_uri(child), get_urifilename(child)))190 res += '</ul>'191 res += '<hr noshade>'192 res += ('<em>Powered by <a href="%s">%s</a> version %s</em>'193 % (quote_uri(WEBSITE), PACKAGE, VERSION))194 res += '</body>'195 res += '</html>'196 return res197 pool = Pool(Transaction().cursor.database_name)198 Collection = pool.get('webdav.collection')199 try:200 res = Collection.get_data(dburi, cache=CACHE)201 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:202 self._log_exception(exception)203 raise204 except Exception, exception:205 self._log_exception(exception)206 raise DAV_Error(500)207 if range is None:208 return res209 size = len(res)210 if range[1] == '':211 range[1] = size212 else:213 range[1] = int(range[1])214 if range[1] > size:215 range[1] = size216 if range[0] == '':217 range[0] = size - range[1]218 else:219 range[0] = int(range[0])220 if range[0] > size:221 raise DAV_Requested_Range_Not_Satisfiable222 return res[range[0]:range[1]]223 def put(self, uri, data, content_type=''):224 dbname, dburi = self._get_dburi(uri)225 if not dbname or not dburi:226 raise DAV_Forbidden227 pool = Pool(Transaction().cursor.database_name)228 Collection = pool.get('webdav.collection')229 try:230 res = Collection.put(dburi, data, content_type, cache=CACHE)231 Transaction().cursor.commit()232 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:233 self._log_exception(exception)234 Transaction().cursor.rollback()235 raise236 except Exception, exception:237 self._log_exception(exception)238 Transaction().cursor.rollback()239 raise DAV_Error(500)240 if res:241 uparts = list(urlparse.urlsplit(uri))242 uparts[2] = res243 res = urlparse.urlunsplit(uparts)244 return res245 def mkcol(self, uri):246 dbname, dburi = self._get_dburi(uri)247 if not dbname or not dburi:248 raise DAV_Forbidden249 pool = Pool(Transaction().cursor.database_name)250 Collection = pool.get('webdav.collection')251 try:252 res = Collection.mkcol(dburi, cache=CACHE)253 Transaction().cursor.commit()254 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:255 self._log_exception(exception)256 Transaction().cursor.rollback()257 raise258 except Exception, exception:259 self._log_exception(exception)260 Transaction().cursor.rollback()261 raise DAV_Error(500)262 return res263 def _get_dav_resourcetype(self, uri):264 dbname, dburi = self._get_dburi(uri)265 if not dbname or not dburi:266 return COLLECTION267 pool = Pool(Transaction().cursor.database_name)268 Collection = pool.get('webdav.collection')269 try:270 res = Collection.get_resourcetype(dburi, cache=CACHE)271 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:272 self._log_exception(exception)273 raise274 except Exception, exception:275 self._log_exception(exception)276 raise DAV_Error(500)277 return res278 def _get_dav_displayname(self, uri):279 dbname, dburi = self._get_dburi(uri)280 if not dbname or not dburi:281 return uri.split('/')[-1]282 pool = Pool(Transaction().cursor.database_name)283 try:284 Collection = pool.get('webdav.collection')285 res = Collection.get_displayname(dburi, cache=CACHE)286 except KeyError:287 raise DAV_NotFound288 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:289 self._log_exception(exception)290 raise291 except Exception, exception:292 self._log_exception(exception)293 raise DAV_Error(500)294 return res295 def _get_dav_getcontentlength(self, uri):296 dbname, dburi = self._get_dburi(uri)297 if not dbname or not dburi:298 return '0'299 pool = Pool(Transaction().cursor.database_name)300 Collection = pool.get('webdav.collection')301 try:302 res = Collection.get_contentlength(dburi, cache=CACHE)303 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:304 self._log_exception(exception)305 raise306 except Exception, exception:307 self._log_exception(exception)308 raise DAV_Error(500)309 return res310 def _get_dav_getcontenttype(self, uri):311 dbname, dburi = self._get_dburi(uri)312 if not dbname or self.is_collection(uri):313 return "text/html"314 pool = Pool(Transaction().cursor.database_name)315 Collection = pool.get('webdav.collection')316 try:317 res = Collection.get_contenttype(dburi, cache=CACHE)318 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:319 self._log_exception(exception)320 raise321 except Exception, exception:322 self._log_exception(exception)323 raise DAV_Error(500)324 return res325 def _get_dav_getetag(self, uri):326 return '"' + str(self.get_lastmodified(uri)) + '"'327 def get_creationdate(self, uri):328 dbname, dburi = self._get_dburi(uri)329 if not dbname or not dburi:330 return time.time()331 pool = Pool(Transaction().cursor.database_name)332 Collection = pool.get('webdav.collection')333 try:334 res = Collection.get_creationdate(dburi, cache=CACHE)335 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:336 self._log_exception(exception)337 raise338 except Exception, exception:339 self._log_exception(exception)340 raise DAV_Error(500)341 return res342 def get_lastmodified(self, uri):343 dbname, dburi = self._get_dburi(uri)344 if not dbname or not dburi:345 return time.time()346 pool = Pool(Transaction().cursor.database_name)347 Collection = pool.get('webdav.collection')348 try:349 res = Collection.get_lastmodified(dburi, cache=CACHE)350 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:351 self._log_exception(exception)352 raise353 except Exception, exception:354 self._log_exception(exception)355 raise DAV_Error(500)356 return res357 def rmcol(self, uri):358 dbname, dburi = self._get_dburi(uri)359 if not dbname or not dburi:360 return 403361 pool = Pool(Transaction().cursor.database_name)362 Collection = pool.get('webdav.collection')363 try:364 res = Collection.rmcol(dburi, cache=CACHE)365 Transaction().cursor.commit()366 except Exception, exception:367 self._log_exception(exception)368 Transaction().cursor.rollback()369 return 500370 return res371 def rm(self, uri):372 dbname, dburi = self._get_dburi(uri)373 if not dbname or not dburi:374 return 403375 pool = Pool(Transaction().cursor.database_name)376 Collection = pool.get('webdav.collection')377 try:378 res = Collection.rm(dburi, cache=CACHE)379 Transaction().cursor.commit()380 except Exception, exception:381 self._log_exception(exception)382 Transaction().cursor.rollback()383 return 500384 return res385 def exists(self, uri):386 dbname, dburi = self._get_dburi(uri)387 if not dbname or not dburi:388 return 1389 pool = Pool(Transaction().cursor.database_name)390 Collection = pool.get('webdav.collection')391 try:392 res = Collection.exists(dburi, cache=CACHE)393 except (DAV_Error, DAV_NotFound, DAV_Secret, DAV_Forbidden), exception:394 self._log_exception(exception)395 raise396 except Exception, exception:397 self._log_exception(exception)398 raise DAV_Error(500)399 return res400 def is_collection(self, uri):401 if self._get_dav_resourcetype(uri) == COLLECTION:402 return 1403 return 0404 def copyone(self, src, dst, overwrite):405 return copyone(self, src, dst, overwrite)406 def copytree(self, src, dst, overwrite):407 return copytree(self, src, dst, overwrite)408 def moveone(self, src, dst, overwrite):409 return moveone(self, src, dst, overwrite)410 def movetree(self, src, dst, overwrite):411 return movetree(self, src, dst, overwrite)412 def delone(self, uri):413 return delone(self, uri)414 def deltree(self, uri):415 return deltree(self, uri)416 def copy(self, src, dst):417 content = self._get_dav_getcontenttype(src)418 data = self.get_data(src)419 self.put(dst, data, content)420 return 201421 def copycol(self, src, dst):422 return self.mkcol(dst)423 def _get_dav_current_user_privilege_set(self, uri):424 dbname, dburi = self._get_dburi(uri)425 privileges = []426 if not dbname or not dburi:427 privileges = ['create', 'read', 'write', 'delete']428 else:429 pool = Pool(Transaction().cursor.database_name)430 try:431 Collection = pool.get('webdav.collection')432 privileges = Collection.current_user_privilege_set(dburi,433 cache=CACHE)434 except KeyError:435 pass436 except Exception, exception:437 self._log_exception(exception)438 pass439 doc = domimpl.createDocument(None, 'privilege', None)440 privilege = doc.documentElement441 privilege.tagName = 'D:privilege'442 if 'create' in privileges:443 bind = doc.createElement('D:bind')444 privilege.appendChild(bind)445 if 'read' in privileges:446 read = doc.createElement('D:read')447 privilege.appendChild(read)448 read_acl = doc.createElement('D:read-acl')449 privilege.appendChild(read_acl)450 if 'write' in privileges:451 write = doc.createElement('D:write')...
caldav.py
Source:caldav.py
...58 res = Collection.get_calendar_description(dburi, cache=CACHE)59 except AttributeError:60 raise DAV_NotFound61 except DAV_Error, exception:62 self._log_exception(exception)63 raise64 except Exception, exception:65 self._log_exception(exception)66 raise DAV_Error(500)67 return res68TrytonDAVInterface._get_caldav_calendar_description = \69 _get_caldav_calendar_description70def _get_caldav_calendar_data(self, uri):71 dbname, dburi = self._get_dburi(uri)72 if not dbname:73 raise DAV_NotFound74 pool = Pool(Transaction().cursor.database_name)75 try:76 Collection = pool.get('webdav.collection')77 except KeyError:78 raise DAV_NotFound79 try:80 res = Collection.get_calendar_data(dburi, cache=CACHE)81 except AttributeError:82 raise DAV_NotFound83 except DAV_Error, exception:84 self._log_exception(exception)85 raise86 except Exception, exception:87 self._log_exception(exception)88 raise DAV_Error(500)89 return res90TrytonDAVInterface._get_caldav_calendar_data = _get_caldav_calendar_data91def _get_caldav_calendar_home_set(self, uri):92 dbname, dburi = self._get_dburi(uri)93 if not dbname:94 raise DAV_NotFound95 pool = Pool(Transaction().cursor.database_name)96 try:97 Collection = pool.get('webdav.collection')98 except KeyError:99 raise DAV_NotFound100 try:101 res = Collection.get_calendar_home_set(dburi, cache=CACHE)102 except AttributeError:103 raise DAV_NotFound104 except DAV_Error, exception:105 self._log_exception(exception)106 raise107 except Exception, exception:108 self._log_exception(exception)109 raise DAV_Error(500)110 uparts = list(urlparse.urlsplit(uri))111 uparts[2] = urllib.quote(dbname + res)112 doc = domimpl.createDocument(None, 'href', None)113 href = doc.documentElement114 href.tagName = 'D:href'115 #iPhone doesn't handle "http" in href116 #huri = doc.createTextNode(urlparse.urlunsplit(uparts))117 huri = doc.createTextNode(urllib.quote('/' + dbname + res))118 href.appendChild(huri)119 return href120TrytonDAVInterface._get_caldav_calendar_home_set = \121 _get_caldav_calendar_home_set122def _get_caldav_calendar_user_address_set(self, uri):123 dbname, dburi = self._get_dburi(uri)124 if not dbname:125 raise DAV_NotFound126 pool = Pool(Transaction().cursor.database_name)127 try:128 Collection = pool.get('webdav.collection')129 except KeyError:130 raise DAV_NotFound131 try:132 res = Collection.get_calendar_user_address_set(dburi, cache=CACHE)133 except AttributeError:134 raise DAV_NotFound135 except DAV_Error, exception:136 self._log_exception(exception)137 raise138 except Exception, exception:139 self._log_exception(exception)140 raise DAV_Error(500)141 doc = domimpl.createDocument(None, 'href', None)142 href = doc.documentElement143 href.tagName = 'D:href'144 huri = doc.createTextNode('MAILTO:' + res)145 href.appendChild(huri)146 return href147TrytonDAVInterface._get_caldav_calendar_user_address_set = \148 _get_caldav_calendar_user_address_set149def _get_caldav_schedule_inbox_URL(self, uri):150 dbname, dburi = self._get_dburi(uri)151 if not dbname:152 raise DAV_NotFound153 pool = Pool(Transaction().cursor.database_name)154 try:155 Collection = pool.get('webdav.collection')156 except KeyError:157 raise DAV_NotFound158 try:159 res = Collection.get_schedule_inbox_URL(dburi, cache=CACHE)160 except AttributeError:161 raise DAV_NotFound162 except DAV_Error, exception:163 self._log_exception(exception)164 raise165 except Exception, exception:166 self._log_exception(exception)167 raise DAV_Error(500)168 uparts = list(urlparse.urlsplit(uri))169 uparts[2] = urllib.quote(dbname + res)170 doc = domimpl.createDocument(None, 'href', None)171 href = doc.documentElement172 href.tagName = 'D:href'173 huri = doc.createTextNode(urlparse.urlunsplit(uparts))174 href.appendChild(huri)175 return href176TrytonDAVInterface._get_caldav_schedule_inbox_URL = \177 _get_caldav_schedule_inbox_URL178def _get_caldav_schedule_outbox_URL(self, uri):179 dbname, dburi = self._get_dburi(uri)180 if not dbname:181 raise DAV_NotFound182 pool = Pool(Transaction().cursor.database_name)183 try:184 Collection = pool.get('webdav.collection')185 except KeyError:186 raise DAV_NotFound187 try:188 res = Collection.get_schedule_outbox_URL(dburi, cache=CACHE)189 except AttributeError:190 raise DAV_NotFound191 except DAV_Error, exception:192 self._log_exception(exception)193 raise194 except Exception, exception:195 self._log_exception(exception)196 raise DAV_Error(500)197 uparts = list(urlparse.urlsplit(uri))198 uparts[2] = urllib.quote(dbname + res)199 doc = domimpl.createDocument(None, 'href', None)200 href = doc.documentElement201 href.tagName = 'D:href'202 huri = doc.createTextNode(urlparse.urlunsplit(uparts))203 href.appendChild(huri)204 return href205TrytonDAVInterface._get_caldav_schedule_outbox_URL = \206 _get_caldav_schedule_outbox_URL207_prev_get_dav_principal_collection_set = hasattr(TrytonDAVInterface,208 '_get_dav_principal_collection_set') and \209 TrytonDAVInterface._get_dav_principal_collection_set or None210def _get_dav_principal_collection_set(self, uri):211 dbname, dburi = self._get_dburi(uri)212 if dburi.startswith('Calendars'):213 uparts = list(urlparse.urlsplit(uri))214 uparts[2] = urllib.quote(dbname + '/Calendars/')215 doc = domimpl.createDocument(None, 'href', None)216 href = doc.documentElement217 href.tagName = 'D:href'218 huri = doc.createTextNode(urlparse.urlunsplit(uparts))219 href.appendChild(huri)220 return href221 if _prev_get_dav_principal_collection_set:222 return _prev_get_dav_principal_collection_set(self, uri)223 raise DAV_NotFound224TrytonDAVInterface._get_dav_principal_collection_set = \225 _get_dav_principal_collection_set226def _get_caldav_post(self, uri, body, contenttype=''):227 dbname, dburi = self._get_dburi(uri)228 if not dbname:229 raise DAV_Forbidden230 pool = Pool(Transaction().cursor.database_name)231 Calendar = pool.get('calendar.calendar')232 try:233 res = Calendar.post(dburi, body)234 except AttributeError:235 raise DAV_NotFound236 except DAV_Error, exception:237 self._log_exception(exception)238 raise239 except Exception, exception:240 self._log_exception(exception)241 raise DAV_Error(500)242 return res243TrytonDAVInterface._get_caldav_post = _get_caldav_post244_prev_do_POST = WebDAVAuthRequestHandler.do_POST245def do_POST(self):246 dc = self.IFACE_CLASS247 uri = urlparse.urljoin(self.get_baseuri(dc), self.path)248 uri = urllib.unquote(uri)249 dbname, dburi = TrytonDAVInterface.get_dburi(uri)250 if dburi.startswith('Calendars'):251 # read the body252 body = None253 if 'Content-Length' in self.headers:254 l = self.headers['Content-Length']...
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!!