How to use is_private method in Nose

Best Python code snippet using nose

DataLayerOperator.py

Source:DataLayerOperator.py

1#   Copyright 2020 The KNIX Authors2#3#   Licensed under the Apache License, Version 2.0 (the "License");4#   you may not use this file except in compliance with the License.5#   You may obtain a copy of the License at6#7#       http://www.apache.org/licenses/LICENSE-2.08#9#   Unless required by applicable law or agreed to in writing, software10#   distributed under the License is distributed on an "AS IS" BASIS,11#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12#   See the License for the specific language governing permissions and13#   limitations under the License.14from DataLayerClient import DataLayerClient15class DataLayerOperator:16    def __init__(self, suid, sid, wid, datalayer):17        self._storage_userid = suid18        self._sandboxid = sid19        self._workflowid = wid20        self._datalayer = datalayer21        # global data layer clients for either workflow-private data or user storage22        self._data_layer_client = None23        self._data_layer_client_private = None24        # TODO (?): use the local data layer for operations regarding KV, maps, sets and counters instead of in-memory data structures (e.g., transient_data_output)25        # and store the operations/data for is_queued = True operations,26        # so that we can synchronize it with the global data layer27        # (key, value) store28        self.transient_data_output = {}29        self.transient_data_output_private = {}30        self.data_to_be_deleted = {}31        self.data_to_be_deleted_private = {}32        self.map_output = {}33        self.set_output = {}34        self.counter_output = {}35        self.map_output_delete = {}36        self.set_output_delete = {}37        self.counter_output_delete = {}38    # TODO: update to use local data layer for (key, value) operations39    def put(self, key, value, is_private=False, is_queued=False, table=None):40        if is_queued:41            if is_private:42                self.transient_data_output_private[key] = value43                if key in self.data_to_be_deleted_private:44                    self.data_to_be_deleted_private.pop(key, None)45            else:46                self.transient_data_output[key] = value47                if key in self.data_to_be_deleted:48                    self.data_to_be_deleted.pop(key, None)49        else:50            data_layer_client = self._get_data_layer_client(is_private)51            data_layer_client.put(key, value, tableName=table)52    def get(self, key, is_private=False, table=None):53        # check first transient_output54        # if not there, return the actual (global) data layer data item55        # if not there either, return empty string (as defined in the DataLayerClient)56        value = None57        # if the put() or delete() were called with is_queued=False (default),58        # then the below checks will still result in 'value is None'59        # if not, then value will be obtained from the transient output60        if is_private:61            if key in self.data_to_be_deleted_private:62                return ""63            value = self.transient_data_output_private.get(key)64        else:65            if key in self.data_to_be_deleted:66                return ""67            value = self.transient_data_output.get(key)68        if value is None:69            data_layer_client = self._get_data_layer_client(is_private)70            value = data_layer_client.get(key, tableName=table)71        return value72    def delete(self, key, is_private=False, is_queued=False, table=None):73        if is_queued:74            if is_private:75                self.transient_data_output_private.pop(key, None)76                self.data_to_be_deleted_private[key] = True77            else:78                self.transient_data_output.pop(key, None)79                self.data_to_be_deleted[key] = True80        else:81            data_layer_client = self._get_data_layer_client(is_private)82            data_layer_client.delete(key, tableName=table)83    def getKeys(self, start_index, end_index, is_private=False):84        keys = set()85        # XXX: should follow "read your writes"86        # the final result should include:87        # 1. all created locally88        # 2. all existing globally minus the ones deleted locally89        # TODO: 1. check local data layer first: get locally created and deleted90        # 2. retrieve all existing globally91        dlc = self._get_data_layer_client(is_private)92        m2 = dlc.listKeys(start_index, end_index)93        if m2 is not None:94            # TODO: 3. remove the ones deleted locally95            keys = keys.union(m2)96        return list(keys)97    # map operations98    def createMap(self, mapname, is_private=False, is_queued=False):99        if is_queued:100            # TODO: use transient data structure in memory when the operation is queued101            pass102        else:103            dlc = self._get_data_layer_client(is_private)104            dlc.createMap(mapname)105    def putMapEntry(self, mapname, key, value, is_private=False, is_queued=False):106        if is_queued:107            # TODO: use transient data structure in memory when the operation is queued108            pass109        else:110            dlc = self._get_data_layer_client(is_private)111            dlc.putMapEntry(mapname, key, value)112    def getMapEntry(self, mapname, key, is_private=False):113        value = None114        # TODO: check transient data structure first115        if value is None:116            dlc = self._get_data_layer_client(is_private)117            value = dlc.getMapEntry(mapname, key)118        return value119    def deleteMapEntry(self, mapname, key, is_private=False, is_queued=False):120        if is_queued:121            # TODO: use transient data structure in memory when the operation is queued122            pass123        else:124            dlc = self._get_data_layer_client(is_private)125            dlc.deleteMapEntry(mapname, key)126    def containsMapKey(self, mapname, key, is_private=False):127        ret = False128        # TODO: check transient data structure first129        if not ret:130            dlc = self._get_data_layer_client(is_private)131            ret = dlc.containsMapKey(mapname, key)132        return ret133    def retrieveMap(self, mapname, is_private=False):134        retmap = {}135        # XXX: should follow "read your writes"136        # the final result should include:137        # 1. all created locally138        # 2. all existing globally minus the ones deleted locally139        # TODO: 1. check local data layer first: get locally created and deleted140        # 2. retrieve all existing globally141        dlc = self._get_data_layer_client(is_private)142        retmap2 = dlc.retrieveMap(mapname)143        if retmap2 is not None:144            for k in retmap2:145                retmap[k] = retmap2[k]146            # TODO: 3. remove the ones deleted locally147        return retmap148    def getMapKeys(self, mapname, is_private=False):149        keys = set()150        # XXX: should follow "read your writes"151        # the final result should include:152        # 1. all created locally153        # 2. all existing globally minus the ones deleted locally154        # TODO: 1. check local data layer first: get locally created and deleted155        # 2. retrieve all existing globally156        dlc = self._get_data_layer_client(is_private)157        k2 = dlc.getMapKeys(mapname)158        if k2 is not None:159            # TODO: 3. remove the ones deleted locally160            keys = keys.union(k2)161        return keys162    def clearMap(self, mapname, is_private=False, is_queued=False):163        if is_queued:164            # TODO: use transient data structure in memory when the operation is queued165            pass166        else:167            dlc = self._get_data_layer_client(is_private)168            dlc.clearMap(mapname)169    def deleteMap(self, mapname, is_private=False, is_queued=False):170        if is_queued:171            # TODO: use transient data structure in memory when the operation is queued172            pass173        else:174            dlc = self._get_data_layer_client(is_private)175            dlc.deleteMap(mapname)176    def getMapNames(self, start_index=0, end_index=2147483647, is_private=False):177        maps = set()178        # XXX: should follow "read your writes"179        # the final result should include:180        # 1. all created locally181        # 2. all existing globally minus the ones deleted locally182        # TODO: 1. check local data layer first: get locally created and deleted183        # 2. retrieve all existing globally184        dlc = self._get_data_layer_client(is_private)185        m2 = dlc.getMapNames(start_index, end_index)186        if m2 is not None:187            # TODO: 3. remove the ones deleted locally188            maps = maps.union(m2)189        return list(maps)190    # set operations191    def createSet(self, setname, is_private=False, is_queued=False):192        if is_queued:193            # TODO: use transient data structure in memory when the operation is queued194            pass195        else:196            dlc = self._get_data_layer_client(is_private)197            dlc.createSet(setname)198    def addSetEntry(self, setname, item, is_private=False, is_queued=False):199        if is_queued:200            # TODO: use transient data structure in memory when the operation is queued201            pass202        else:203            dlc = self._get_data_layer_client(is_private)204            dlc.addSetEntry(setname, item)205    def removeSetEntry(self, setname, item, is_private=False, is_queued=False):206        if is_queued:207            # TODO: use transient data structure in memory when the operation is queued208            pass209        else:210            dlc = self._get_data_layer_client(is_private)211            dlc.removeSetEntry(setname, item)212    def containsSetItem(self, setname, item, is_private=False):213        ret = False214        # TODO: check transient data structure first215        if not ret:216            dlc = self._get_data_layer_client(is_private)217            ret = dlc.containsSetItem(setname, item)218        return ret219    def retrieveSet(self, setname, is_private=False):220        items = set()221        # XXX: should follow "read your writes"222        # the final result should include:223        # 1. all created locally224        # 2. all existing globally minus the ones deleted locally225        # TODO: 1. check local data layer first: get locally created and deleted226        # 2. retrieve all existing globally227        dlc = self._get_data_layer_client(is_private)228        i2 = dlc.retrieveSet(setname)229        if i2 is not None:230            # TODO: 3. remove the ones deleted locally231            items = items.union(i2)232        return items233    def clearSet(self, setname, is_private=False, is_queued=False):234        if is_queued:235            # TODO: use transient data structure in memory when the operation is queued236            pass237        else:238            dlc = self._get_data_layer_client(is_private)239            dlc.clearSet(setname)240    def deleteSet(self, setname, is_private=False, is_queued=False):241        if is_queued:242            # TODO: use transient data structure in memory when the operation is queued243            pass244        else:245            dlc = self._get_data_layer_client(is_private)246            dlc.deleteSet(setname)247    def getSetNames(self, start_index=0, end_index=2147483647, is_private=False):248        sets = set()249        # XXX: should follow "read your writes"250        # the final result should include:251        # 1. all created locally252        # 2. all existing globally minus the ones deleted locally253        # TODO: 1. check local data layer first: get locally created and deleted254        # 2. retrieve all existing globally255        dlc = self._get_data_layer_client(is_private)256        s2 = dlc.getSetNames(start_index, end_index)257        if s2 is not None:258            # TODO: 3. remove the ones deleted locally259            sets = sets.union(s2)260        return list(sets)261    # counter operations262    def createCounter(self, countername, count, is_private=False, is_queued=False):263        if is_queued:264            # TODO: use transient data structure in memory when the operation is queued265            pass266        else:267            dlc = self._get_data_layer_client(is_private)268            dlc.createCounter(countername, count)269    def getCounterValue(self, countername, is_private=False):270        value = 0271        # TODO: check transient data structure first and apply any changes to the global value272        dlc = self._get_data_layer_client(is_private)273        value = dlc.getCounter(countername)274        return value275    def incrementCounter(self, countername, increment, is_private=False, is_queued=False):276        if is_queued:277            # TODO: use transient data structure in memory when the operation is queued278            pass279        else:280            dlc = self._get_data_layer_client(is_private)281            dlc.incrementCounter(countername, increment)282    def decrementCounter(self, countername, decrement, is_private=False, is_queued=False):283        if is_queued:284            # TODO: use transient data structure in memory when the operation is queued285            pass286        else:287            dlc = self._get_data_layer_client(is_private)288            dlc.decrementCounter(countername, decrement)289    def deleteCounter(self, countername, is_private=False, is_queued=False):290        if is_queued:291            # TODO: use transient data structure in memory when the operation is queued292            pass293        else:294            dlc = self._get_data_layer_client(is_private)295            dlc.deleteCounter(countername)296    def getCounterNames(self, start_index=0, end_index=2147483647, is_private=False):297        counters = set()298        # XXX: should follow "read your writes"299        # the final result should include:300        # 1. all created locally301        # 2. all existing globally minus the ones deleted locally302        # TODO: 1. check local data layer first: get locally created and deleted303        # 2. retrieve all existing globally304        dlc = self._get_data_layer_client(is_private)305        c2 = dlc.getCounterNames(start_index, end_index)306        if c2 is not None:307            # TODO: 3. remove the ones deleted locally308            counters = counters.union(c2)309        return list(counters)310    def get_transient_data_output(self, is_private=False):311        '''312        Return the transient data, so that it can be committed to the data layer313        when the function instance finishes.314        '''315        if is_private:316            return self.transient_data_output_private317        return self.transient_data_output318    def get_data_to_be_deleted(self, is_private=False):319        '''320        Return the list of deleted data items, so that they can be committed to the data layer321        when the function instance finishes.322        '''323        if is_private:324            return self.data_to_be_deleted_private325        return self.data_to_be_deleted326    def _get_data_layer_client(self, is_private=False):327        '''328        Return the data layer client, so that it can be used to commit to the data layer329        when the function instance finishes.330        If it is not initialized yet, it will be initialized here.331        '''332        # TODO: need also the locality information333        if is_private:334            if self._data_layer_client_private is None:335                self._data_layer_client_private = DataLayerClient(locality=1, sid=self._sandboxid, wid=self._workflowid, is_wf_private=True, connect=self._datalayer)336            return self._data_layer_client_private337        if self._data_layer_client is None:338            self._data_layer_client = DataLayerClient(locality=1, suid=self._storage_userid, is_wf_private=False, connect=self._datalayer)339        return self._data_layer_client340    def _shutdown_data_layer_client(self):341        '''342        Shut down the data layer client if it has been initialized343        after the function instance finishes committing changes344        to the data layer.345        '''346        if self._data_layer_client_private is not None:347            self._data_layer_client_private.shutdown()348            self._data_layer_client_private = None349        if self._data_layer_client is not None:350            self._data_layer_client.shutdown()...

file_manager.py

Source:file_manager.py

1# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors2# MIT License. See license.txt3from __future__ import unicode_literals4import frappe5import os, base64, re6import hashlib7import mimetypes8import io9from frappe.utils import get_hook_method, get_files_path, random_string, encode, cstr, call_hook_method, cint10from frappe import _11from frappe import conf12from copy import copy13from six.moves.urllib.parse import unquote14from six import text_type, PY215class MaxFileSizeReachedError(frappe.ValidationError):16	pass17def get_file_url(file_data_name):18	data = frappe.db.get_value("File", file_data_name, ["file_name", "file_url"], as_dict=True)19	return data.file_url or data.file_name20def upload():21	# get record details22	dt = frappe.form_dict.doctype23	dn = frappe.form_dict.docname24	df = frappe.form_dict.docfield25	file_url = frappe.form_dict.file_url26	filename = frappe.form_dict.filename27	frappe.form_dict.is_private = cint(frappe.form_dict.is_private)28	if not filename and not file_url:29		frappe.msgprint(_("Please select a file or url"),30			raise_exception=True)31	file_doc = get_file_doc()32	comment = {}33	if dt and dn:34		comment = frappe.get_doc(dt, dn).add_comment("Attachment",35			_("added {0}").format("<a href='{file_url}' target='_blank'>{file_name}</a>{icon}".format(**{36				"icon": ' <i class="fa fa-lock text-warning"></i>' \37					if file_doc.is_private else "",38				"file_url": file_doc.file_url.replace("#", "%23") \39					if file_doc.file_name else file_doc.file_url,40				"file_name": file_doc.file_name or file_doc.file_url41			})))42	return {43		"name": file_doc.name,44		"file_name": file_doc.file_name,45		"file_url": file_doc.file_url,46		"is_private": file_doc.is_private,47		"comment": comment.as_dict() if comment else {}48	}49def get_file_doc(dt=None, dn=None, folder=None, is_private=None, df=None):50	'''returns File object (Document) from given parameters or form_dict'''51	r = frappe.form_dict52	if dt is None: dt = r.doctype53	if dn is None: dn = r.docname54	if df is None: df = r.docfield55	if folder is None: folder = r.folder56	if is_private is None: is_private = r.is_private57	if r.filedata:58		file_doc = save_uploaded(dt, dn, folder, is_private, df)59	elif r.file_url:60		file_doc = save_url(r.file_url, r.filename, dt, dn, folder, is_private, df)61	return file_doc62def save_uploaded(dt, dn, folder, is_private, df=None):63	fname, content = get_uploaded_content()64	if content:65		return save_file(fname, content, dt, dn, folder, is_private=is_private, df=df);66	else:67		raise Exception68def save_url(file_url, filename, dt, dn, folder, is_private, df=None):69	# if not (file_url.startswith("http://") or file_url.startswith("https://")):70	# 	frappe.msgprint("URL must start with 'http://' or 'https://'")71	# 	return None, None72	file_url = unquote(file_url)73	file_size = frappe.form_dict.file_size74	f = frappe.get_doc({75		"doctype": "File",76		"file_url": file_url,77		"file_name": filename,78		"attached_to_doctype": dt,79		"attached_to_name": dn,80		"attached_to_field": df,81		"folder": folder,82		"file_size": file_size,83		"is_private": is_private84	})85	f.flags.ignore_permissions = True86	try:87		f.insert()88	except frappe.DuplicateEntryError:89		return frappe.get_doc("File", f.duplicate_entry)90	return f91def get_uploaded_content():92	# should not be unicode when reading a file, hence using frappe.form93	if 'filedata' in frappe.form_dict:94		if "," in frappe.form_dict.filedata:95			frappe.form_dict.filedata = frappe.form_dict.filedata.rsplit(",", 1)[1]96		frappe.uploaded_content = base64.b64decode(frappe.form_dict.filedata)97		frappe.uploaded_filename = frappe.form_dict.filename98		return frappe.uploaded_filename, frappe.uploaded_content99	else:100		frappe.msgprint(_('No file attached'))101		return None, None102def save_file(fname, content, dt, dn, folder=None, decode=False, is_private=0, df=None):103	if decode:104		if isinstance(content, text_type):105			content = content.encode("utf-8")106		if b"," in content:107			content = content.split(",")[1]108		content = base64.b64decode(content)109	file_size = check_max_file_size(content)110	content_hash = get_content_hash(content)111	content_type = mimetypes.guess_type(fname)[0]112	fname = get_file_name(fname, content_hash[-6:])113	file_data = get_file_data_from_hash(content_hash, is_private=is_private)114	if not file_data:115		call_hook_method("before_write_file", file_size=file_size)116		write_file_method = get_hook_method('write_file', fallback=save_file_on_filesystem)117		file_data = write_file_method(fname, content, content_type=content_type, is_private=is_private)118		file_data = copy(file_data)119	file_data.update({120		"doctype": "File",121		"attached_to_doctype": dt,122		"attached_to_name": dn,123		"attached_to_field": df,124		"folder": folder,125		"file_size": file_size,126		"content_hash": content_hash,127		"is_private": is_private128	})129	f = frappe.get_doc(file_data)130	f.flags.ignore_permissions = True131	try:132		f.insert()133	except frappe.DuplicateEntryError:134		return frappe.get_doc("File", f.duplicate_entry)135	return f136def get_file_data_from_hash(content_hash, is_private=0):137	for name in frappe.db.sql_list("select name from `tabFile` where content_hash=%s and is_private=%s", (content_hash, is_private)):138		b = frappe.get_doc('File', name)139		return {k: b.get(k) for k in frappe.get_hooks()['write_file_keys']}140	return False141def save_file_on_filesystem(fname, content, content_type=None, is_private=0):142	fpath = write_file(content, fname, is_private)143	if is_private:144		file_url = "/private/files/{0}".format(fname)145	else:146		file_url = "/files/{0}".format(fname)147	return {148		'file_name': os.path.basename(fpath),149		'file_url': file_url150	}151def get_max_file_size():152	return conf.get('max_file_size') or 10485760153def check_max_file_size(content):154	max_file_size = get_max_file_size()155	file_size = len(content)156	if file_size > max_file_size:157		frappe.msgprint(_("File size exceeded the maximum allowed size of {0} MB").format(158			max_file_size / 1048576),159			raise_exception=MaxFileSizeReachedError)160	return file_size161def write_file(content, fname, is_private=0):162	"""write file to disk with a random name (to compare)"""163	file_path = get_files_path(is_private=is_private)164	# create directory (if not exists)165	frappe.create_folder(file_path)166	# write the file167	if isinstance(content, text_type):168		content = content.encode()169	with open(os.path.join(file_path.encode('utf-8'), fname.encode('utf-8')), 'wb+') as f:170		f.write(content)171	return get_files_path(fname, is_private=is_private)172def remove_all(dt, dn, from_delete=False):173	"""remove all files in a transaction"""174	try:175		for fid in frappe.db.sql_list("""select name from `tabFile` where176			attached_to_doctype=%s and attached_to_name=%s""", (dt, dn)):177			remove_file(fid, dt, dn, from_delete)178	except Exception as e:179		if e.args[0]!=1054: raise # (temp till for patched)180def remove_file_by_url(file_url, doctype=None, name=None):181	if doctype and name:182		fid = frappe.db.get_value("File", {"file_url": file_url,183			"attached_to_doctype": doctype, "attached_to_name": name})184	else:185		fid = frappe.db.get_value("File", {"file_url": file_url})186	if fid:187		return remove_file(fid)188def remove_file(fid, attached_to_doctype=None, attached_to_name=None, from_delete=False):189	"""Remove file and File entry"""190	file_name = None191	if not (attached_to_doctype and attached_to_name):192		attached = frappe.db.get_value("File", fid,193			["attached_to_doctype", "attached_to_name", "file_name"])194		if attached:195			attached_to_doctype, attached_to_name, file_name = attached196	ignore_permissions, comment = False, None197	if attached_to_doctype and attached_to_name and not from_delete:198		doc = frappe.get_doc(attached_to_doctype, attached_to_name)199		ignore_permissions = doc.has_permission("write") or False200		if frappe.flags.in_web_form:201			ignore_permissions = True202		if not file_name:203			file_name = frappe.db.get_value("File", fid, "file_name")204		comment = doc.add_comment("Attachment Removed", _("Removed {0}").format(file_name))205	frappe.delete_doc("File", fid, ignore_permissions=ignore_permissions)206	return comment207def delete_file_data_content(doc, only_thumbnail=False):208	method = get_hook_method('delete_file_data_content', fallback=delete_file_from_filesystem)209	method(doc, only_thumbnail=only_thumbnail)210def delete_file_from_filesystem(doc, only_thumbnail=False):211	"""Delete file, thumbnail from File document"""212	if only_thumbnail:213		delete_file(doc.thumbnail_url)214	else:215		delete_file(doc.file_url)216		delete_file(doc.thumbnail_url)217def delete_file(path):218	"""Delete file from `public folder`"""219	if path:220		if ".." in path.split("/"):221			frappe.msgprint(_("It is risky to delete this file: {0}. Please contact your System Manager.").format(path))222		parts = os.path.split(path.strip("/"))223		if parts[0]=="files":224			path = frappe.utils.get_site_path("public", "files", parts[-1])225		else:226			path = frappe.utils.get_site_path("private", "files", parts[-1])227		path = encode(path)228		if os.path.exists(path):229			os.remove(path)230def get_file(fname):231	"""Returns [`file_name`, `content`] for given file name `fname`"""232	file_path = get_file_path(fname)233	# read the file234	if PY2:235		with open(encode(file_path)) as f:236			content = f.read()237	else:238		with io.open(encode(file_path), mode='rb') as f:239			try:240				# for plain text files241				content = f.read().decode()242			except UnicodeDecodeError:243				# for .png, .jpg, etc244				content = f.read()245	return [file_path.rsplit("/", 1)[-1], content]246def get_file_path(file_name):247	"""Returns file path from given file name"""248	f = frappe.db.sql("""select file_url from `tabFile`249		where name=%s or file_name=%s""", (file_name, file_name))250	if f:251		file_name = f[0][0]252	file_path = file_name253	if "/" not in file_path:254		file_path = "/files/" + file_path255	if file_path.startswith("/private/files/"):256		file_path = get_files_path(*file_path.split("/private/files/", 1)[1].split("/"), is_private=1)257	elif file_path.startswith("/files/"):258		file_path = get_files_path(*file_path.split("/files/", 1)[1].split("/"))259	else:260		frappe.throw(_("There is some problem with the file url: {0}").format(file_path))261	return file_path262def get_content_hash(content):263	if isinstance(content, text_type):264		content = content.encode()265	return hashlib.md5(content).hexdigest()266def get_file_name(fname, optional_suffix):267	# convert to unicode268	fname = cstr(fname)269	n_records = frappe.db.sql("select name from `tabFile` where file_name=%s", fname)270	if len(n_records) > 0 or os.path.exists(encode(get_files_path(fname))):271		f = fname.rsplit('.', 1)272		if len(f) == 1:273			partial, extn = f[0], ""274		else:275			partial, extn = f[0], "." + f[1]276		return '{partial}{suffix}{extn}'.format(partial=partial, extn=extn, suffix=optional_suffix)277	return fname278@frappe.whitelist()279def download_file(file_url):280	"""281	Download file using token and REST API. Valid session or282	token is required to download private files.283	Method : GET284	Endpoint : frappe.utils.file_manager.download_file285	URL Params : file_name = /path/to/file relative to site path286	"""287	file_doc = frappe.get_doc("File", {"file_url":file_url})288	file_doc.check_permission("read")289	with open(getattr(frappe.local, "site_path", None) + file_url, "rb") as fileobj:290		filedata = fileobj.read()291	frappe.local.response.filename = file_url[file_url.rfind("/")+1:]292	frappe.local.response.filecontent = filedata293	frappe.local.response.type = "download"294def extract_images_from_doc(doc, fieldname):295	content = doc.get(fieldname)296	content = extract_images_from_html(doc, content)297	if frappe.flags.has_dataurl:298		doc.set(fieldname, content)299def extract_images_from_html(doc, content):300	frappe.flags.has_dataurl = False301	def _save_file(match):302		data = match.group(1)303		data = data.split("data:")[1]304		headers, content = data.split(",")305		if "filename=" in headers:306			filename = headers.split("filename=")[-1]307			# decode filename308			if not isinstance(filename, text_type):309				filename = text_type(filename, 'utf-8')310		else:311			mtype = headers.split(";")[0]312			filename = get_random_filename(content_type=mtype)313		doctype = doc.parenttype if doc.parent else doc.doctype314		name = doc.parent or doc.name315		# TODO fix this316		file_url = save_file(filename, content, doctype, name, decode=True).get("file_url")317		if not frappe.flags.has_dataurl:318			frappe.flags.has_dataurl = True319		return '<img src="{file_url}"'.format(file_url=file_url)320	if content:321		content = re.sub('<img[^>]*src\s*=\s*["\'](?=data:)(.*?)["\']', _save_file, content)322	return content323def get_random_filename(extn=None, content_type=None):324	if extn:325		if not extn.startswith("."):326			extn = "." + extn327	elif content_type:328		extn = mimetypes.guess_extension(content_type)329	return random_string(7) + (extn or "")330@frappe.whitelist()331def validate_filename(filename):332	from frappe.utils import now_datetime333	timestamp = now_datetime().strftime(" %Y-%m-%d %H:%M:%S")334	fname = get_file_name(filename, timestamp)...

_NodeLocalQueuesMethods.py

Source:_NodeLocalQueuesMethods.py

1import sys2import os3sys.path.append(os.path.dirname(os.path.realpath(__file__)))4from _utils import eprint5from _NodeInternalClasses import CloseableQueue6def put(self, value, timeout=None, block=True):7	if self._server == None:8		self._queue.put(value)9	else:10		session_id = self._get_session_id()11		self._request(session_id, value=value, timeout=timeout, block=block)12		response = self._recv_response(session_id)13		if not response["success"]:14			eprint(response["traceback"])15			raise response["exception"]16def get(self, timeout=None, block=True):17	if self._server == None:18		return self._queue.get()19	else:20		session_id = self._get_session_id()21		self._request(session_id, timeout=timeout, block=block)22		response = self._recv_response(session_id)23		if response["success"]:24			return response["data"]["value"]25		else:26			eprint(response["traceback"])27			raise response["exception"]28def qsize(self):29	if self._server == None:30		return self._queue.qsize()31	else:32		session_id = self._get_session_id()33		self._request(session_id)34		response = self._recv_response(session_id)35		if response["success"]:36			return response["data"]["qsize"]37		else:38			eprint(response["traceback"])39			raise response["exception"]40def _locals_queue_put(self, name, value, timeout, block, is_private):41	if self._server == None:42		if not is_private:43			if name not in self._actual_queues:44				self._actual_queues[name] = CloseableQueue()45			self._actual_queues[name].put(value, timeout=timeout, block=block)46		else:47			if name not in self._actual_result_queues_for_future:48				self._actual_result_queues_for_future[name] = CloseableQueue()49			self._actual_result_queues_for_future[name].put(value, timeout=timeout, block=block)50	else:51		session_id = self._get_session_id()52		self._request(session_id, name=name, value=value, timeout=timeout, block=block, is_private=is_private)53		response = self._recv_response(session_id)54		if not response["success"]:55			eprint(response["traceback"])56			raise response["exception"]57def _locals_queue_get(self, name, timeout, block, is_private):58	if self._server == None:59		if not is_private:60			if name not in self._actual_queues:61				self._actual_queues[name] = CloseableQueue()62			return self._actual_queues[name].get(timeout=timeout, block=block)63		else:64			if name not in self._actual_result_queues_for_future:65				self._actual_result_queues_for_future[name] = CloseableQueue()66			return self._actual_result_queues_for_future[name].get(timeout=timeout, block=block)67	else:68		session_id = self._get_session_id()69		self._request(session_id, name=name, timeout=timeout, block=block, is_private=is_private)70		response = self._recv_response(session_id)71		if response["success"]:72			return response["data"]["value"]73		else:74			eprint(response["traceback"])75			raise response["exception"]76def _locals_queue_close(self, name, is_private):77	if self._server == None:78		if not is_private:79			if name not in self._actual_queues:80				return81			self._actual_queues[name].close()82		else:83			if name not in self._actual_result_queues_for_future:84				return85			self._actual_result_queues_for_future[name].close()86	else:87		session_id = self._get_session_id()88		self._request(session_id, name=name, is_private=is_private)89		response = self._recv_response(session_id)90		if not response["success"]:91			eprint(response["traceback"])92			raise response["exception"]93def _locals_queue_len(self, name, is_private):94	if self._server == None:95		if not is_private:96			if name not in self._actual_queues:97				return 098			else:99				return self._actual_queues[name].qsize()100		else:101			if name not in self._actual_result_queues_for_future:102				return 0103			else:104				return self._actual_result_queues_for_future[name].qsize()105	else:106		session_id = self._get_session_id()107		self._request(session_id, name=name, is_private=is_private)108		response = self._recv_response(session_id)109		if response["success"]:110			return response["data"]["len"]111		else:112			eprint(response["traceback"])113			raise response["exception"]114def _locals_queues_delitem(self, name, is_private):115	if self._server == None:116		if not is_private:117			self._actual_queues[name].close()118			del self._actual_queues[name]119		else:120			self._actual_result_queues_for_future[name].close()121			del self._actual_result_queues_for_future[name]122	else:123		session_id = self._get_session_id()124		self._request(session_id, name=name, is_private=is_private)125		response = self._recv_response(session_id)126		if not response["success"]:127			eprint(response["traceback"])128			raise response["exception"]129def _locals_queues_len(self, is_private):130	if self._server == None:131		if not is_private:132			return self._actual_queues.__len__()133		else:134			return self._actual_result_queues_for_future.__len__()135	else:136		session_id = self._get_session_id()137		self._request(session_id, is_private=is_private)138		response = self._recv_response(session_id)139		if response["success"]:140			return response["data"]["len"]141		else:142			eprint(response["traceback"])143			raise response["exception"]144def _locals_queues_contains(self, name, is_private):145	if self._server == None:146		if not is_private:147			return self._actual_queues.__contains__(name)148		else:149			return self._actual_result_queues_for_future.__contains__(name)150	else:151		session_id = self._get_session_id()152		self._request(session_id, name=name, is_private=is_private)153		response = self._recv_response(session_id)154		if response["success"]:155			return response["data"]["contains"]156		else:157			eprint(response["traceback"])158			raise response["exception"]159def _locals_queues_keys(self, is_private):160	if self._server == None:161		if not is_private:162			return self._actual_queues.keys()163		else:164			return self._actual_result_queues_for_future.keys()165	else:166		session_id = self._get_session_id()167		self._request(session_id, is_private=is_private)168		response = self._recv_response(session_id)169		if response["success"]:170			return response["data"]["keys"]171		else:172			eprint(response["traceback"])173			raise response["exception"]174def _locals_queues_iter(self, is_private):175	if self._server == None:176		if not is_private:177			return self._actual_queues.__iter__()178		else:179			return self._actual_result_queues_for_future.__iter__()180	else:181		session_id = self._get_session_id()182		self._request(session_id, is_private=is_private)183		response = self._recv_response(session_id)184		if response["success"]:185			return response["data"]["iter"]186		else:187			eprint(response["traceback"])188			raise response["exception"]189def _locals_queues_clear(self, is_private):190	if self._server == None:191		if not is_private:192			for name in self._actual_queues:193				self._actual_queues[name].close()194			self._actual_queues.clear()195		else:196			for name in self._actual_result_queues_for_future:197				self._actual_result_queues_for_future[name].close()198			self._actual_result_queues_for_future.clear()199	else:200		session_id = self._get_session_id()201		self._request(session_id, is_private=is_private)202		response = self._recv_response(session_id)203		if not response["success"]:204			eprint(response["traceback"])205			raise response["exception"]206def _process_put(self, request):207	try:208		self._queue.put(request["data"]["value"], timeout=request["data"]["timeout"], block=request["block"])209		self._respond_ok(request["session_id"])210	except BaseException as e:211		self._respond_exception(request["session_id"], e)212def _process_get(self, request):213	try:214		value = self._queue.get(timeout=request["data"]["timeout"], block=request["block"])215		self._respond_ok(request["session_id"], value=value)216	except BaseException as e:217		self._respond_exception(request["session_id"], e)218def _process_qsize(self, request):219	try:220		qsize = self._queue.qsize()221		self._respond_ok(request["session_id"], qsize=qsize)222	except BaseException as e:223		self._respond_exception(request["session_id"], e)224def _process__locals_queue_put(self, request):225	try:226		name = request["data"]["name"]227		if not request["data"]["is_private"]:228			if name not in self._actual_queues:229				self._actual_queues[name] = CloseableQueue()230			self._actual_queues[name].put(request["data"]["value"], timeout=request["data"]["timeout"], block=request["block"])231		else:232			if name not in self._actual_result_queues_for_future:233				self._actual_result_queues_for_future[name] = CloseableQueue()234			self._actual_result_queues_for_future[name].put(request["data"]["value"], timeout=request["data"]["timeout"], block=request["block"])235		self._respond_ok(request["session_id"])236	except BaseException as e:237		self._respond_exception(request["session_id"], e)238def _process__locals_queue_get(self, request):239	try:240		name = request["data"]["name"]241		if not request["data"]["is_private"]:242			if name not in self._actual_queues:243				self._actual_queues[name] = CloseableQueue()244			value = self._actual_queues[request["data"]["name"]].get(timeout=request["data"]["timeout"], block=request["block"])245		else:246			if name not in self._actual_result_queues_for_future:247				self._actual_result_queues_for_future[name] = CloseableQueue()248			value = self._actual_result_queues_for_future[request["data"]["name"]].get(timeout=request["data"]["timeout"], block=request["block"])249		self._respond_ok(request["session_id"], value=value)250	except BaseException as e:251		self._respond_exception(request["session_id"], e)252def _process__locals_queue_len(self, request):253	try:254		if not request["data"]["is_private"]:255			if request["data"]["name"] not in self._actual_queues:256				self._respond_ok(request["session_id"], len=0)257			else:258				self._respond_ok(request["session_id"], len=self._actual_queues[request["data"]["name"]].qsize())259		else:260			if request["data"]["name"] not in self._actual_result_queues_for_future:261				self._respond_ok(request["session_id"], len=0)262			else:263				self._respond_ok(request["session_id"], len=self._actual_result_queues_for_future[request["data"]["name"]].qsize())264	except BaseException as e:265		self._respond_exception(request["session_id"], e)266def _process__locals_queue_close(self, request):267	try:268		name = request["data"]["name"]269		if not request["data"]["is_private"]:270			try:271				self._actual_queues[name].close()272			except:273				pass274		else:275			try:276				self._actual_result_queues_for_future[name].close()277			except:278				pass279		self._respond_ok(request["session_id"])280	except BaseException as e:281		self._respond_exception(request["session_id"], e)282def _process__locals_queues_delitem(self, request):283	try:284		name = request["data"]["name"]285		if not request["data"]["is_private"]:286			try:287				self._actual_queues[name].close()288			except:289				pass290			del self._actual_queues[name]291		else:292			try:293				self._actual_result_queues_for_future[name].close()294			except:295				pass296			del self._actual_result_queues_for_future[name]297		self._respond_ok(request["session_id"])298	except BaseException as e:299		self._respond_exception(request["session_id"], e)300def _process__locals_queues_len(self, request):301	try:302		if not request["data"]["is_private"]:303			self._respond_ok(request["session_id"], len=self._actual_queues.__len__())304		else:305			self._respond_ok(request["session_id"], len=self._actual_result_queues_for_future.__len__())306	except BaseException as e:307		self._respond_exception(request["session_id"], e)308def _process__locals_queues_contains(self, request):309	try:310		if not request["data"]["is_private"]:311			self._respond_ok(request["session_id"], contains=self._actual_queues.__contains__(request["data"]["name"]))312		else:313			self._respond_ok(request["session_id"], contains=self._actual_result_queues_for_future.__contains__(request["data"]["name"]))314	except BaseException as e:315		self._respond_exception(request["session_id"], e)316def _process__locals_queues_keys(self, request):317	try:318		if not request["data"]["is_private"]:319			self._respond_ok(request["session_id"], keys=self._actual_queues.keys())320		else:321			self._respond_ok(request["session_id"], keys=self._actual_result_queues_for_future.keys())322	except BaseException as e:323		self._respond_exception(request["session_id"], e)324def _process__locals_queues_iter(self, request):325	try:326		if not request["data"]["is_private"]:327			self._respond_ok(request["session_id"], iter=self._actual_queues.__iter__())328		else:329			self._respond_ok(request["session_id"], iter=self._actual_result_queues_for_future.__iter__())330	except BaseException as e:331		self._respond_exception(request["session_id"], e)332def _process__locals_queues_clear(self, request):333	try:334		if not request["data"]["is_private"]:335			for name in self._actual_queues:336				try:337					self._actual_queues[name].close()338				except:339					pass340			self._actual_queues.clear()341		else:342			for name in self._actual_result_queues_for_future:343				try:344					self._actual_result_queues_for_future[name].close()345				except:346					pass347			self._actual_result_queues_for_future.clear()348		self._respond_ok(request["session_id"])349	except BaseException as e:...

test_bmarkmgr.py

Source:test_bmarkmgr.py

1"""Test the basics including the bmark and tags"""2from random import randint3from bookie.models import (4    Bmark,5    BmarkMgr,6    DBSession,7    TagMgr,8)9from bookie.models.auth import User10from bookie.models.stats import (11    StatBookmark,12    StatBookmarkMgr,13)14from bookie.tests import gen_random_word15from bookie.tests import TestDBBase16class TestBmarkMgrStats(TestDBBase):17    """Handle some bmarkmgr stats checks"""18    def test_total_ct(self):19        """Verify that our total count method is working"""20        ct = 521        user = User()22        user.username = gen_random_word(10)23        DBSession.add(user)24        for i in range(ct):25            b = Bmark(26                url=gen_random_word(12),27                username=user.username,28                is_private=False29            )30            b.hash_id = gen_random_word(3)31            DBSession.add(b)32        ct = BmarkMgr.count()33        self.assertEqual(5, ct,34                         'We should have 5 public bookmarks: ' + str(ct))35    def test_total_ct_accounts_for_privacy(self):36        """Verify that our total count method is working"""37        ct = 538        user = User()39        user.username = gen_random_word(10)40        DBSession.add(user)41        for i in range(ct):42            b = Bmark(43                url=gen_random_word(12),44                username=user.username,45                is_private=True46            )47            b.hash_id = gen_random_word(3)48            DBSession.add(b)49        ct = BmarkMgr.count(is_private=True)50        self.assertEqual(5, ct,51                         'We should have 5 private bookmarks: ' + str(ct))52    def test_unique_ct(self):53        """Verify that our unique count method is working"""54        ct = 555        common = u'testing.com'56        users = []57        for i in range(ct):58            user = User()59            user.username = gen_random_word(10)60            DBSession.add(user)61            users.append(user)62        for i in range(ct - 2):63            b = Bmark(64                url=gen_random_word(12),65                username=users[i].username,66                is_private=False67            )68            DBSession.add(b)69        # Add in our dupes70        c = Bmark(71            url=common,72            username=users[3].username,73            is_private=False74        )75        DBSession.add(c)76        DBSession.flush()77        d = Bmark(78            url=common,79            username=users[4].username,80            is_private=False81        )82        DBSession.add(d)83        DBSession.flush()84        ct = BmarkMgr.count(distinct=True)85        self.assertEqual(4, ct,86                         'We should have 4 public bookmarks: ' + str(ct))87    def test_unique_ct_accounts_for_privacy(self):88        """Verify that our unique count method is working"""89        ct = 590        common = u'testing.com'91        users = []92        for i in range(ct):93            user = User()94            user.username = gen_random_word(10)95            DBSession.add(user)96            users.append(user)97        for i in range(ct - 2):98            b = Bmark(99                url=gen_random_word(12),100                username=users[i].username,101                is_private=True102            )103            DBSession.add(b)104        # Add in our dupes105        c = Bmark(106            url=common,107            username=users[3].username,108            is_private=True109        )110        DBSession.add(c)111        DBSession.flush()112        d = Bmark(113            url=common,114            username=users[4].username,115            is_private=True,116        )117        DBSession.add(d)118        DBSession.flush()119        ct = BmarkMgr.count(distinct=True, is_private=True)120        self.assertEqual(4, ct,121                         'We should have 4 private bookmarks: ' + str(ct))122    def test_per_user(self):123        """We should only get a pair of results for this single user"""124        ct = 5125        common = u'testing.com'126        user = User()127        user.username = gen_random_word(10)128        DBSession.add(user)129        usercommon = User()130        usercommon.username = common131        DBSession.add(usercommon)132        for i in range(ct - 2):133            b = Bmark(134                url=gen_random_word(12),135                username=user.username,136                is_private=False137            )138            DBSession.add(b)139        # add in our dupes140        c = Bmark(141            url=gen_random_word(10),142            username=usercommon.username,143            is_private=False144        )145        DBSession.add(c)146        DBSession.flush()147        d = Bmark(148            url=gen_random_word(10),149            username=usercommon.username,150            is_private=False151        )152        DBSession.add(d)153        DBSession.flush()154        ct = BmarkMgr.count(username=usercommon.username)155        self.assertEqual(2, ct,156                         'We should have 2 public bookmarks: ' + str(ct))157    def test_per_user_accounts_for_privacy(self):158        """We should only get a pair of results for this single user"""159        ct = 5160        common = u'testing.com'161        user = User()162        user.username = gen_random_word(10)163        DBSession.add(user)164        usercommon = User()165        usercommon.username = common166        DBSession.add(usercommon)167        for i in range(ct - 2):168            b = Bmark(169                url=gen_random_word(12),170                username=user.username,171                is_private=True,172            )173            DBSession.add(b)174        c = Bmark(175            url=gen_random_word(10),176            username=usercommon.username,177            is_private=True,178        )179        DBSession.add(c)180        DBSession.flush()181        d = Bmark(182            url=gen_random_word(10),183            username=usercommon.username,184            is_private=True,185        )186        DBSession.add(d)187        DBSession.flush()188        ct = BmarkMgr.count(username=usercommon.username, is_private=True)189        self.assertEqual(2, ct,190                         'We should only have 2 private bookmarks: ' + str(ct))191    def test_count_user_bookmarks(self):192        """We should get a total count of both private and public bookmarks"""193        public_bmark_count = 5194        private_bmark_count = 5195        total_bmark_count = public_bmark_count + private_bmark_count196        user = User()197        user.username = gen_random_word(10)198        DBSession.add(user)199        stat_username = "user_bookmarks_{0}".format(user.username)200        # Add the public bookmarks.201        for i in range(public_bmark_count):202            b = Bmark(203                url=gen_random_word(12),204                username=user.username,205                is_private=False,206            )207            DBSession.add(b)208        # Add the private bookmarks.209        for i in range(private_bmark_count):210            b = Bmark(211                url=gen_random_word(12),212                username=user.username,213                is_private=True,214            )215            DBSession.add(b)216        StatBookmarkMgr.count_user_bookmarks(username=user.username)217        res = DBSession.query(StatBookmark).\218            filter(StatBookmark.attrib == stat_username).first()219        self.assertEqual(220            total_bmark_count, res.data,221            'We should have {0} bookmarks: '.format(total_bmark_count) +222            str(res.data))223    def test_delete_all_bookmarks(self):224        """Testing working of delete all bookmarks225                Case 1: No bookmark present226                Case 2: One bookmark present227                Case 3: Multiple bookmarks present"""228        bmark_counts = [0, 1]229        for i in range(10):230            bmark_counts.append(randint(10, 100))231        users = []232        for i in range(len(bmark_counts)):233            user = User()234            user.username = gen_random_word(10)235            users.append(user)236            DBSession.add(user)237            for j in range(i):238                b = Bmark(239                    url=gen_random_word(12),240                    username=user.username,241                    tags=gen_random_word(4),242                )243                b.hash_id = gen_random_word(3)244                DBSession.add(b)245        DBSession.flush()246        for user in users:247            BmarkMgr.delete_all_bookmarks(user.username)248            ct = BmarkMgr.count(user.username)249            self.assertEqual(ct, 0, 'All the bookmarks should be deleted')250            tags = TagMgr.find(username=user.username)251            self.assertEqual(252                len(tags),253                0,254                'There should be no tags left: ' + str(len(tags))255            )256            DBSession.flush()257    def test_find_bookmarks_same_user(self):258        """A user requesting their bookmarks includes private ones"""259        bookmark_count_private = 5260        bookmark_count_public = 5261        user = User()262        user.username = gen_random_word(19)263        DBSession.add(user)264        for i in range(bookmark_count_private):265            b = Bmark(266                url=gen_random_word(12),267                username=user.username,268            )269            DBSession.add(b)270        for i in range(bookmark_count_public):271            b = Bmark(272                url=gen_random_word(12),273                username=user.username,274                is_private=False,275            )276            DBSession.add(b)277        DBSession.flush()278        res = BmarkMgr.find(username=user.username, requested_by=user.username)279        self.assertEqual(280            bookmark_count_private + bookmark_count_public,281            len(res),282            'There should be ' + str(bookmark_count_private +283                                     bookmark_count_public) +284            ' bookmarks present: ' + str(len(res))285        )286    def test_find_bookmarks_diff_user(self):287        """A user requesting another user's bookmarks get public only"""288        bookmark_count_private = 5289        bookmark_count_public = 5290        user = User()291        user.username = gen_random_word(19)292        DBSession.add(user)293        for i in range(bookmark_count_private):294            b = Bmark(295                url=gen_random_word(12),296                username=user.username,297                is_private=True,298            )299            DBSession.add(b)300        for i in range(bookmark_count_public):301            b = Bmark(302                url=gen_random_word(12),303                username=user.username,304                is_private=False,305            )306            DBSession.add(b)307        DBSession.flush()308        res = BmarkMgr.find(username=user.username,309                            requested_by=gen_random_word(19))310        self.assertEqual(311            bookmark_count_public,312            len(res),313            'There should be ' + str(bookmark_count_public) +314            ' bookmarks present: ' + str(len(res))315        )316        # Also check if requested_by is None.317        res = BmarkMgr.find(username=user.username, requested_by=None)318        self.assertEqual(319            bookmark_count_public,320            len(res),321            'There should be ' + str(bookmark_count_public) +322            ' bookmarks present: ' + str(len(res))...

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.