Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_missing_api_key():
'''verify exception raised on missing API key'''
client = api.ClientV1(api_key=None)
try:
client._get('whatevs').get_body()
except api.exceptions.InvalidAPIKey as ex:
assert str(ex) == 'No API key provided'
else:
assert False
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from _common import read_fixture
import json
import os
from planet import api
from requests_mock import Mocker
client = api.Client(api_key='xyz')
def test_list_mosaics():
with Mocker() as m:
text = read_fixture('list-mosaics.json')
uri = os.path.join(client.base_url, 'mosaics/')
m.get(uri, text=text, status_code=200)
r = client.list_mosaics()
assert r.response.status_code == 200
assert r.get() == json.loads(text)
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
from mock import MagicMock
from mock import DEFAULT
from planet import api
from planet.api.models import Scenes
from planet.api.sync import _SyncTool
from _common import read_fixture
client = MagicMock(name='client', spec=api.Client)
def test_sync_tool(tmpdir):
# test non-existing destination
try:
_SyncTool(client, 'should-not-exist', None, None, None)
except ValueError as ve:
assert str(ve) == 'destination must exist and be a directory'
# test existing destination, no aoi.geojson
td = tmpdir.mkdir('sync-dest')
try:
_SyncTool(client, td.strpath, None, None, None)
except ValueError as ve:
assert str(ve) == 'no aoi provided and no aoi.geojson file'
def get_and_set_Planet_key(user_account):
keyfile = HOME+'/.planetkey'
with open(keyfile) as f_obj:
lines = f_obj.readlines()
for line in lines:
if user_account in line:
key_str = line.split(':')[1]
key_str = key_str.strip() # remove '\n'
os.environ["PL_API_KEY"] = key_str
# set Planet API client
global client
client = api.ClientV1(api_key = key_str)
return True
raise ValueError('account: %s cannot find in %s'%(user_account,keyfile))
time.sleep(3)
output_stream = sys.stdout
def download_progress(start=None,wrote=None,total=None, finish=None): #result,skip=None
# print(start,wrote,total,finish)
# if total:
# # print('received: %.2f K'%(float(total)/1024.0))
# output_stream.write('received: %.2f K'%(float(total)/1024.0))
# output_stream.flush()
# if total:
# if finish is None:
# print('received: %.2f K'%(float(total)/1024.0), end='\r')
# else:
# print('received: %.2f K' % (float(total) / 1024.0))
pass
callback = api.write_to_file(directory=save_dir + '/', callback=download_progress) # save_dir + '/' #
body = client.download(assets[asset_key], callback=callback)
# if body._body.name == '':
# basic.outputlogMessage('Warning, the body name is missed, set as the asset key and id: item id: %s, asset %s'%(item['id'],asset_key))
# body._body.name = item['id']+'_'+asset_key # AttributeError: can't set attribute
body.await()
return True
def get_and_set_Planet_key(user_account):
keyfile = HOME+'/.planetkey'
with open(keyfile) as f_obj:
lines = f_obj.readlines()
for line in lines:
if user_account in line:
key_str = line.split(':')[1]
key_str = key_str.strip() # remove '\n'
os.environ["PL_API_KEY"] = key_str
# set Planet API client
global client
client = api.ClientV1(api_key = key_str)
return True
raise ValueError('account: %s cannot find in %s'%(user_account,keyfile))
def click_exception(ex):
if type(ex) is api.exceptions.APIException:
raise click.ClickException('Unexpected response: %s' % str(ex))
msg = "%s: %s" % (type(ex).__name__, str(ex))
raise click.ClickException(msg)
help='Valid API key - or via ENV variable %s' % api.auth.ENV_KEY)
@click.option('-u', '--base-url', envvar='PL_API_BASE_URL',
help='Change the base Planet API URL or ENV PL_API_BASE_URL'
' - Default https://api.planet.com/')
@click.option('-au', '--analytics-base-url',
envvar='PL_ANALYTICS_API_BASE_URL',
help=('Change the base Planet API URL or ENV '
'PL_ANALYTICS_API_BASE_URL'
' - Default https://api.planet.com/analytics'))
@click.version_option(version=__version__, message='%(version)s')
def cli(context, verbose, api_key, base_url, analytics_base_url, workers):
'''Planet API Client'''
configure_logging(verbose)
client_params.clear()
client_params['api_key'] = api_key
def call_and_wrap(func, *args, **kw):
'''call the provided function and wrap any API exception with a click
exception. this means no stack trace is visible to the user but instead
a (hopefully) nice message is provided.
note: could be a decorator but didn't play well with click
'''
try:
return func(*args, **kw)
except api.exceptions.APIException as ex:
click_exception(ex)