def test_app():
# app's config should be loaded from instance/config.py
assert app.config['STORAGE_TYPE'] == 'file'
assert app.config['THEME'] == 'test'
with app.test_request_context('/'):
app.send_static_file('no-use.css')
app.send_static_file('no-use-2.css')
with raises(NotFound):
app.send_static_file('non-exists.css')
origin_mode = app.config['MODE']
app.config['MODE'] = 'api-only'
with app.test_request_context('/'):
with raises(NotFound):
app.send_static_file('no-use.css')
app.config['MODE'] = origin_mode
python类NotFound()的实例源码
def send_static_file(self, filename):
"""
Send static files from the static folder in the current selected theme prior to the global static folder.
:param filename: static filename
:return: response object
"""
if self.config['MODE'] == 'api-only':
abort(404) # if 'api-only' mode is set, we should not send static files
theme_static_folder = getattr(self, 'theme_static_folder', None)
if theme_static_folder:
try:
return send_from_directory(theme_static_folder, filename)
except NotFound:
pass
return super(CustomFlask, self).send_static_file(filename)
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or \
filename == '..' or \
filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or \
filename == '..' or \
filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def entity(entity_id):
entity = load_entity(entity_id, request.auth)
if entity is None:
raise NotFound()
# load possible duplicates via fingerprint expansion
# fingerprints = expand_fingerprints(entity.fingerprints, request.auth)
query = Query(request.args, prefix='duplicates_', path=request.path,
limit=5)
# query.add_facet('countries', 'Countries', country_label)
duplicates = search_duplicates(entity.id, entity.fingerprints, query,
request.auth)
# load links
query = Query(request.args, prefix='links_', path=request.path,
limit=10)
query.add_facet('schemata', 'Types', link_schema_label)
query.add_facet('remote.countries', 'Countries', country)
links = search_links(entity, query, request.auth)
return render_template("entity.html", entity=entity, links=links,
duplicates=duplicates)
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or \
filename == '..' or \
filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or \
filename == '..' or \
filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or \
filename == '..' or \
filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or \
filename == '..' or \
filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def safe_join(directory, filename):
"""Safely join `directory` and `filename`.
Example usage::
@app.route('/wiki/<path:filename>')
def wiki_page(filename):
filename = safe_join(app.config['WIKI_FOLDER'], filename)
with open(filename, 'rb') as fd:
content = fd.read() # Read and process the file content...
:param directory: the base directory.
:param filename: the untrusted filename relative to that directory.
:raises: :class:`~werkzeug.exceptions.NotFound` if the resulting path
would fall out of `directory`.
"""
filename = posixpath.normpath(filename)
for sep in _os_alt_seps:
if sep in filename:
raise NotFound()
if os.path.isabs(filename) or \
filename == '..' or \
filename.startswith('../'):
raise NotFound()
return os.path.join(directory, filename)
def test_get(self):
"""Test get method"""
app = self._test_application()
mockDBObject.return_value = app
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestApplication(restapi=self.restapi, server=self.server)
# Fail to find a device: raises 404 NotFound
with patch.object(Application, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.get(app.appeui), e.NotFound)
# Find a device success returns a dict of field values
with patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)):
resource.getProperties = MagicMock()
result = yield resource.get(app.appeui)
self.assertEqual(app.appeui, result['appeui'])
def test_put(self):
"""Test put method"""
app = self._test_application()
mockDBObject.return_value = app
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestApplication(restapi=self.restapi, server=self.server)
# Fail to find a device: raises 404 NotFound
with patch.object(Application, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.put(app.appeui), e.NotFound)
# Find a device, device fails validity check: raises 400 BadRequest
with patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)):
app.valid = MagicMock(return_value=(False, {}))
yield self.assertFailure(resource.put(app.appeui), e.BadRequest)
# Pass validity check, returns 200
expected = ({}, 200)
app.valid = MagicMock(return_value=(True, {}))
app.update = MagicMock()
result = yield resource.put(app.appeui)
self.assertEqual(expected, result)
def test_delete(self):
"""Test delete method"""
app = self._test_application()
mockDBObject.return_value = app
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestApplication(restapi=self.restapi, server=self.server)
# Device exists with AppEUI: raises 400 error
with patch.object(Device, 'find', classmethod(mockDBObject.findSuccess)):
yield self.assertFailure(resource.delete(app.appeui), e.BadRequest)
# Fail to find the application: raises 404 NotFound
with patch.object(Device, 'find', classmethod(mockDBObject.findFail)), \
patch.object(Application, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.delete(app.appeui), e.NotFound)
# Find and delete, returns 200
with patch.object(Device, 'find', classmethod(mockDBObject.findFail)), \
patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)):
expected = ({}, 200)
app.delete = MagicMock()
result = yield resource.delete(app.appeui)
self.assertEqual(expected, result)
def test_get(self):
"""Test get method"""
device = self._test_device()
mockDBObject.return_value = device
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestDevice(restapi=self.restapi, server=self.server)
# Fail to find a device: raises 404 NotFound
with patch.object(Device, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.get(device.deveui), e.NotFound)
# Find a device success returns a dict of field values
with patch.object(Device, 'find', classmethod(mockDBObject.findSuccess)):
result = yield resource.get(device.deveui)
self.assertEqual(device.deveui, result['deveui'])
def test_put(self):
"""Test put method"""
device = self._test_device()
mockDBObject.return_value = device
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestDevice(restapi=self.restapi, server=self.server)
# Fail to find a device: raises 404 NotFound
with patch.object(Device, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.put(device.deveui), e.NotFound)
# Find a device, device fails validity check: raises 400 BadRequest
with patch.object(Device, 'find', classmethod(mockDBObject.findSuccess)):
device.valid = MagicMock(return_value=(False, {}))
yield self.assertFailure(resource.put(device.deveui), e.BadRequest)
# Pass validity check, returns 200
expected = ({}, 200)
device.valid = MagicMock(return_value=(True, {}))
device.update = MagicMock()
result = yield resource.put(device.deveui)
self.assertEqual(expected, result)
def test_delete(self):
"""Test delete method"""
app = self._test_application()
prop = self._test_appproperty()
args = {'port': 11}
with patch.object(reqparse.RequestParser, 'parse_args',
MagicMock(return_value=args)):
resource = RestAppProperty(restapi=self.restapi, server=self.server)
mockDBObject.return_value = prop
# Fail to find the application: raises 404 NotFound
with patch.object(Application, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.delete(app.appeui), e.NotFound)
# Find and delete, returns 200
with patch.object(Application, 'find', classmethod(mockDBObject.findSuccess)), \
patch.object(AppProperty, 'find', classmethod(mockDBObject.findSuccess)):
expected = ({}, 200)
prop.delete = MagicMock()
result = yield resource.delete(app.appeui)
self.assertEqual(expected, result)
def test_get(self):
"""Test get method"""
gateway = self._test_gateway()
mockDBObject.return_value = gateway
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestGateway(restapi=self.restapi, server=self.server)
# Fail to find a gateway: raises 404 NotFound
with patch.object(Gateway, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.get(gateway.host), e.NotFound)
# Find a device success returns a dict of field values
with patch.object(Gateway, 'find', classmethod(mockDBObject.findSuccess)):
result = yield resource.get(gateway.host)
self.assertEqual(gateway.host, result['host'])
def test_put(self):
"""Test put method"""
gateway = self._test_gateway()
mockDBObject.return_value = gateway
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestGateway(restapi=self.restapi, server=self.server)
# Fail to find the gateway: raises 404 NotFound
with patch.object(Gateway, 'find', classmethod(mockDBObject.findFail)):
yield self.assertFailure(resource.put(gateway.host), e.NotFound)
# Find a device, device fails validity check: raises 400 BadRequest
with patch.object(Gateway, 'find', classmethod(mockDBObject.findSuccess)):
gateway.valid = MagicMock(return_value=(False, {}))
yield self.assertFailure(resource.put(gateway.host), e.BadRequest)
# Pass validity check, returns 200
expected = ({}, 200)
gateway.valid = MagicMock(return_value=(True, {}))
gateway.update = MagicMock()
result = yield resource.put(gateway.host)
self.assertEqual(expected, result)
def test_get(self):
"""Test get method"""
interface = self._test_azureiothttps()
appif = self._test_appinterface()
interface.appinterface = appif
interfaceManager.interfaces = [interface]
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestAppInterface(restapi=self.restapi, server=self.server)
# Fail to find the app interface: raises 404 NotFound
interfaceManager.getInterface = MagicMock(return_value=None)
yield self.assertFailure(resource.get(1), e.NotFound)
# Success finding the interface returns a dict of field values
interfaceManager.getInterface = MagicMock(return_value=interface)
result = yield resource.get(appif.id)
self.assertEqual(interface.name, result['name'])
def test_delete(self):
"""Test delete method"""
interface = self._test_azureiothttps()
appif = self._test_appinterface()
interface.appinterface = appif
interfaceManager.interfaces = [interface]
with patch.object(reqparse.RequestParser, 'parse_args'):
resource = RestAppInterface(restapi=self.restapi, server=self.server)
# Fail to find the interface: raises 404 NotFound
interfaceManager.getInterface = MagicMock(return_value=None)
yield self.assertFailure(resource.delete(1), e.NotFound)
# Find and delete, returns 200
with patch.object(AzureIotHttps, 'exists', MagicMock(return_value=True)), \
patch.object(HasMany, 'get', MagicMock(return_value=[appif])), \
patch.object(AzureIotHttps, 'delete'), \
patch.object(AppInterface, 'delete'):
interfaceManager.getInterface = MagicMock(return_value=interface)
expected = ({}, 200)
result = yield resource.delete(1)
self.assertEqual(expected, result)
