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_session_basic(self, m_req):
"""
Test using basic access authentication.
"""
m_response_ok = mock.MagicMock()
type(m_response_ok).status_code = mock.PropertyMock(return_value=200)
type(m_response_ok).text = mock.PropertyMock(return_value='["animaldb"]')
m_req.return_value = m_response_ok
client = Cloudant('foo', 'bar', url=self.url, use_basic_auth=True)
client.connect()
self.assertIsInstance(client.r_session, BasicSession)
all_dbs = client.all_dbs()
m_req.assert_called_once_with(
'GET',
self.url + '/_all_dbs',
allow_redirects=True,
auth=('foo', 'bar'), # uses HTTP Basic Auth
timeout=None
)
self.assertEquals(all_dbs, ['animaldb'])
# construct Cloudant client (using basic access authentication)
self.client = Cloudant(
self.user,
self.pwd,
url=self.url,
x_cloudant_user=self.account,
connect=auto_connect,
auto_renew=auto_renew,
encoder=encoder,
timeout=timeout,
use_basic_auth=True,
)
elif self.iam_api_key:
self.use_cookie_auth = False
# construct Cloudant client (using IAM authentication)
self.client = Cloudant(
None, # username is not required
self.iam_api_key,
url=self.url,
x_cloudant_user=self.account,
connect=auto_connect,
auto_renew=auto_renew,
encoder=encoder,
timeout=timeout,
use_iam=True,
)
else:
# construct Cloudant client (using cookie authentication)
self.client = Cloudant(
self.user,
self.pwd,
url=self.url,
def test_constructor_with_creds_removed_from_url(self):
"""
Test instantiating a client object using a URL
"""
client = Cloudant(None, None, url='https://a9a9a9a9-a9a9-a9a9-a9a9-a9a9a9a9a9a9-bluemix'
':a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9'
'a9a9a9a9a9a9@d8a01891-e4d2-4102-b5f8-751fb735ce31-'
'bluemix.cloudant.com')
self.assertEqual(client.server_url, 'https://d8a01891-e4d2-4102-b5f8-751fb735ce31-'
'bluemix.cloudant.com')
self.assertEqual(client._user, 'a9a9a9a9-a9a9-a9a9-a9a9-a9a9a9a9a9a9-bluemix')
self.assertEqual(client._auth_token, 'a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a9a'
'9a9a9a9a9a9a9a9a9a9a9a9a9')
url=self.url,
connect=auto_connect,
auto_renew=auto_renew,
encoder=encoder,
timeout=timeout
)
else:
self.account = os.environ.get('CLOUDANT_ACCOUNT')
self.url = os.environ.get(
'DB_URL',
'https://{0}.cloudant.com'.format(self.account))
if os.environ.get('RUN_BASIC_AUTH_TESTS'):
self.use_cookie_auth = False
# construct Cloudant client (using basic access authentication)
self.client = Cloudant(
self.user,
self.pwd,
url=self.url,
x_cloudant_user=self.account,
connect=auto_connect,
auto_renew=auto_renew,
encoder=encoder,
timeout=timeout,
use_basic_auth=True,
)
elif self.iam_api_key:
self.use_cookie_auth = False
# construct Cloudant client (using IAM authentication)
self.client = Cloudant(
None, # username is not required
self.iam_api_key,
def test_constructor_with_account(self):
"""
Test instantiating a client object using an account name
"""
# Ensure that the client is new
del self.client
self.client = Cloudant(self.user, self.pwd, account=self.account)
self.assertEqual(
self.client.server_url,
'https://{0}.cloudant.com'.format(self.account)
)
self.use_cookie_auth = False
# construct Cloudant client (using IAM authentication)
self.client = Cloudant(
None, # username is not required
self.iam_api_key,
url=self.url,
x_cloudant_user=self.account,
connect=auto_connect,
auto_renew=auto_renew,
encoder=encoder,
timeout=timeout,
use_iam=True,
)
else:
# construct Cloudant client (using cookie authentication)
self.client = Cloudant(
self.user,
self.pwd,
url=self.url,
x_cloudant_user=self.account,
connect=auto_connect,
auto_renew=auto_renew,
encoder=encoder,
timeout=timeout
)
slack_bot_id = os.environ.get("SLACK_BOT_ID")
slack_client = SlackClient(os.environ.get('SLACK_BOT_TOKEN'))
conversation_workspace_id = os.environ.get("CONVERSATION_WORKSPACE_ID")
conversation_client = ConversationV1(
username=os.environ.get("CONVERSATION_USERNAME"),
password=os.environ.get("CONVERSATION_PASSWORD"),
version='2016-07-11'
)
recipe_client = RecipeClient(os.environ.get("SPOONACULAR_KEY"))
recipe_store_url = os.environ.get("CLOUDANT_URL")
if recipe_store_url.find('@') > 0:
prefix = recipe_store_url[0:recipe_store_url.find('://')+3]
suffix = recipe_store_url[recipe_store_url.find('@')+1:]
recipe_store_url = '{}{}'.format(prefix, suffix)
recipe_store = CloudantRecipeStore(
Cloudant(
os.environ.get("CLOUDANT_USERNAME"),
os.environ.get("CLOUDANT_PASSWORD"),
url=recipe_store_url
),
os.environ.get("CLOUDANT_DB_NAME")
)
# start the souschef bot
souschef = SousChef(slack_bot_id,
slack_client,
conversation_client,
conversation_workspace_id,
recipe_client,
recipe_store)
souschef.start()
sys.stdin.readline()
except (KeyboardInterrupt, SystemExit):
cloudant_account = cloudant_creds['username']
# Instantiate Watson Assistant client.
# - only give a url if we have one (don't override the default)
assistant_kwargs = {
'version': '2019-02-28',
'iam_apikey': assistant_iam_apikey
}
if assistant_url:
assistant_kwargs['url'] = assistant_url
assistant_client = AssistantV1(**assistant_kwargs)
# Instantiate Cloudant DB.
cloudant_online_store = CloudantOnlineStore(
Cloudant.iam(
cloudant_account,
cloudant_iam_apikey,
connect=True
),
cloudant_db_name
)
# Instantiate Watson Discovery client.
# - only give a url if we have one (don't override the default)
discovery_kwargs = {
'version': '2019-04-30',
'iam_apikey': discovery_iam_apikey
}
if discovery_url:
discovery_kwargs['url'] = discovery_url
slack_bot_id = os.environ.get("SLACK_BOT_ID")
slack_client = SlackClient(os.environ.get('SLACK_BOT_TOKEN'))
conversation_workspace_id = os.environ.get("CONVERSATION_WORKSPACE_ID")
conversation_client = ConversationV1(
username=os.environ.get("CONVERSATION_USERNAME"),
password=os.environ.get("CONVERSATION_PASSWORD"),
version='2016-07-11'
)
recipe_client = RecipeClient(os.environ.get("SPOONACULAR_KEY"))
recipe_store_url = os.environ.get("CLOUDANT_URL")
if recipe_store_url.find('@') > 0:
prefix = recipe_store_url[0:recipe_store_url.find('://')+3]
suffix = recipe_store_url[recipe_store_url.find('@')+1:]
recipe_store_url = '{}{}'.format(prefix, suffix)
recipe_store = CloudantRecipeStore(
Cloudant(
os.environ.get("CLOUDANT_USERNAME"),
os.environ.get("CLOUDANT_PASSWORD"),
url=recipe_store_url
),
os.environ.get("CLOUDANT_DB_NAME")
)
# start the souschef bot
souschef = SousChef(slack_bot_id,
slack_client,
conversation_client,
conversation_workspace_id,
recipe_client,
recipe_store)
souschef.start()
# start the http server
print("Start serving at port %i" % PORT)