本文整理汇总了Python中backend.okeanos_utils.check_credentials函数的典型用法代码示例。如果您正苦于以下问题：Python check_credentials函数的具体用法？Python check_credentials怎么用？Python check_credentials使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了check_credentials函数的7个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: setUp
def setUp(self):
self.driver = webdriver.Firefox()
self.driver.implicitly_wait(30)
self.verificationErrors = []
self.accept_next_alert = True
parser = RawConfigParser()
config_file = join(BASE_DIR, '.private/.config.txt')
self.name = 'testcluster'
parser.read(config_file)
try:
self.token = parser.get('cloud \"~okeanos\"', 'token')
self.auth_url = parser.get('cloud \"~okeanos\"', 'url')
self.base_url = parser.get('deploy', 'url')
self.project_name = parser.get('project', 'name')
auth = check_credentials(self.token)
try:
list_of_projects = auth.get_projects(state='active')
except Exception:
self.assertTrue(False,'Could not get list of projects')
for project in list_of_projects:
if project['name'] == self.project_name:
self.project_id = project['id']
except NoSectionError:
self.token = 'INVALID_TOKEN'
self.auth_url = "INVALID_AUTH_URL"
self.base_url = "INVALID_APP_URL"
self.project_name = "INVALID_PROJECT_NAME"
print 'Current authentication details are kept off source control. ' \
'\nUpdate your .config.txt file in <projectroot>/.private/'
开发者ID:themiszamani，项目名称:e-science，代码行数:29，代码来源:test_create_clusters_celery.py
示例2: setUp
def setUp(self):
self.driver = webdriver.Firefox()
self.driver.implicitly_wait(30)
self.verificationErrors = []
self.accept_next_alert = True
parser = RawConfigParser()
config_file = join(BASE_DIR, ".private/.config.txt")
self.name = "testcluster"
parser.read(config_file)
try:
self.token = parser.get('cloud "~okeanos"', "token")
self.auth_url = parser.get('cloud "~okeanos"', "url")
self.base_url = parser.get("deploy", "url")
self.project_name = parser.get("project", "name")
auth = check_credentials(self.token)
try:
list_of_projects = auth.get_projects(state="active")
except Exception:
self.assertTrue(False, "Could not get list of projects")
for project in list_of_projects:
if project["name"] == self.project_name:
self.project_id = project["id"]
except NoSectionError:
self.token = "INVALID_TOKEN"
self.auth_url = "INVALID_AUTH_URL"
self.base_url = "INVALID_APP_URL"
self.project_name = "INVALID_PROJECT_NAME"
print "Current authentication details are kept off source control. " "\nUpdate your .config.txt file in <projectroot>/.private/"
开发者ID:amathilda，项目名称:e-science，代码行数:28，代码来源:test_create_cluster_with_hadoop_image.py
示例3: test_create_clusters_celery
def test_create_clusters_celery(self):
driver = self.driver
driver.get(self.base_url + "#/homepage")
driver.find_element_by_id("id_login").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_login_route"))
)
except: self.fail("time out")
driver.find_element_by_id("token").clear()
driver.find_element_by_id("token").send_keys(self.token)
driver.find_element_by_xpath("//button[@type='login']").click()
if (self.is_element_present(By.XPATH, "//div[@id='id_alert_wrongtoken']/strong") == True):
self.assertTrue(False,'Invalid token')
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_welcome_route"))
)
except: self.fail("time out")
driver.find_element_by_id("id_services_dd").click()
driver.find_element_by_id("id_create_cluster").click()
try:
element = WebDriverWait(driver, 60).until(
EC.presence_of_element_located((By.ID, "id_title_cluster_create_route"))
)
except: self.fail("time out")
auth = check_credentials(self.token)
try:
list_of_projects = auth.get_projects(state='active')
except Exception:
self.assertTrue(False,'Could not get list of projects')
kamaki_flavors = get_flavor_id(self.token)
user_quota = check_quota(self.token, self.project_id)
list = Select(driver.find_element_by_id("project_id")).options
no_project = True
for index in range(0,len(list)):
if re.match(self.project_name, list[index].text):
Select(driver.find_element_by_id("project_id")).select_by_visible_text(list[index].text)
no_project = False
break
if no_project:
self.assertTrue(False,'No project found with given project name')
driver.find_element_by_id("cluster_name").clear()
cluster_name = 'test_cluster' + str(randint(0,9999))
driver.find_element_by_id("cluster_name").send_keys(cluster_name)
hadoop_image = 'Hadoop-2.5.2'
Select(driver.find_element_by_id("os_systems")).select_by_visible_text(hadoop_image)
Select(driver.find_element_by_id("size_of_cluster")).select_by_visible_text('2')
for role in ['master' , 'slaves']:
for flavor in ['cpus' , 'ram' , 'disk']:
button_id = role + '_' + flavor + '_' + str(kamaki_flavors[flavor][0])
driver.find_element_by_id(button_id).click()
driver.find_element_by_id("next").click()
print 'Creating cluster...'
print cluster_name
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_welcome_route"))
)
except: self.fail("time out")
driver.find_element_by_id("id_services_dd").click()
driver.find_element_by_id("id_create_cluster").click()
try:
element = WebDriverWait(driver, 60).until(
EC.presence_of_element_located((By.ID, "id_title_cluster_create_route"))
)
except: self.fail("time out")
driver.find_element_by_id("id_apply_last_cluster").click()
time.sleep(2)
driver.find_element_by_id("cluster_name").clear()
cluster_name1 = 'test_cluster' + str(randint(0,9999))
driver.find_element_by_id("cluster_name").send_keys(cluster_name1)
driver.find_element_by_id("next").click()
print 'Creating cluster...'
print cluster_name1
for i in range(1500):
# wait for cluster create to finish
try:
if "glyphicon glyphicon-play text-success" == driver.find_element_by_id('id_hadoop_status_'+cluster_name1).get_attribute("class"):
break
elif "glyphicon glyphicon-remove text-danger" == driver.find_element_by_id('id_cluster_status_'+cluster_name1).get_attribute("class"):
self.assertTrue(False,'Cluster destoryed')
break
else:
pass
except: pass
time.sleep(1)
cluster_url = str(driver.find_element_by_id("id_ip_"+cluster_name1).get_attribute("href"))
time.sleep(30)
driver.get(cluster_url)
#check that cluster url is up and page is running
try: self.assertEqual("All Applications", driver.find_element_by_css_selector("h1").text)
except AssertionError as e: self.verificationErrors.append(str(e))
开发者ID:themiszamani，项目名称:e-science，代码行数:93，代码来源:test_create_clusters_celery.py
示例4: test_create_cluster
def test_create_cluster(self):
driver = self.driver
driver.get(self.base_url + "#/homepage")
driver.find_element_by_id("id_login").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_login_route"))
)
except: self.fail("time out")
driver.find_element_by_id("token").clear()
driver.find_element_by_id("token").send_keys(self.token)
driver.find_element_by_xpath("//button[@type='login']").click()
if (self.is_element_present(By.XPATH, "//div[@id='id_alert_wrongtoken']/strong") == True):
self.assertTrue(False,'Invalid token')
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_welcome_route"))
)
except: self.fail("time out")
driver.find_element_by_id("id_services_dd").click()
driver.find_element_by_id("id_create_cluster").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_cluster_create_route"))
)
except: self.fail("time out")
auth = check_credentials(self.token)
try:
list_of_projects = auth.get_projects(state='active')
except Exception:
self.assertTrue(False,'Could not get list of projects')
kamaki_flavors = get_flavor_lists(self.token)
user_quota = check_quota(self.token, self.project_id)
list = Select(driver.find_element_by_id("project_id")).options
no_project = True
for index in range(0,len(list)):
if re.match(self.project_name, list[index].text):
Select(driver.find_element_by_id("project_id")).select_by_visible_text(list[index].text)
no_project = False
break
if no_project:
self.assertTrue(False,'No project found with given project name')
driver.find_element_by_id("cluster_name").clear()
cluster_name = 'test_cluster' + str(randint(0,9999))
driver.find_element_by_id("cluster_name").send_keys(cluster_name)
Select(driver.find_element_by_id("size_of_cluster")).select_by_visible_text('2')
for role in ['master' , 'slaves']:
for flavor in ['cpus' , 'ram' , 'disk']:
button_id = role + '_' + flavor + '_' + str(kamaki_flavors[flavor][0])
driver.find_element_by_id(button_id).click()
driver.find_element_by_id("next").click()
print 'Creating cluster...'
for i in range(1800):
# wait for cluster create to finish
try:
if "" != driver.find_element_by_id('id_output_message').text: break
except: pass
time.sleep(1)
message = driver.find_element_by_id('id_output_message').text
if message.rsplit(':', 1)[-1] == '8088/cluster':
cluster_url = message.rsplit(' ', 1)[-1]
driver.get(cluster_url)
print message
#check that cluster url is up and page is running
try: self.assertEqual("All Applications", driver.find_element_by_css_selector("h1").text)
except AssertionError as e: self.verificationErrors.append(str(e))
else:
self.assertTrue(False, message)
开发者ID:grnet，项目名称:e-science，代码行数:68，代码来源:test_create_cluster.py
示例5: test_text_respond_to_buttons
def test_text_respond_to_buttons(self):
driver = self.driver
driver.get(self.base_url + "#/homepage")
driver.find_element_by_id("id_login").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_login_route"))
)
except: self.fail("time out")
driver.find_element_by_id("token").clear()
driver.find_element_by_id("token").send_keys(self.token)
driver.find_element_by_xpath("//button[@type='login']").click()
if (self.is_element_present(By.XPATH, "//div[@id='id_alert_wrongtoken']/strong") == True):
self.assertTrue(False,'Invalid token')
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_welcome_route"))
)
except: self.fail("time out")
driver.find_element_by_id("id_services_dd").click()
driver.find_element_by_id("id_create_cluster").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_cluster_create_route"))
)
except: self.fail("time out")
auth = check_credentials(self.token)
try:
list_of_projects = auth.get_projects(state='active')
except Exception:
self.assertTrue(False,'Could not get list of projects')
kamaki_flavors = get_flavor_id(self.token)
for project in list_of_projects:
user_quota = check_quota(self.token, project['id'])
if project['name'] == 'system:' + project['id']:
project_name = 'system'
else:
project_name = project['name']
list = Select(driver.find_element_by_id("project_id")).options
no_project = True
for index in range(0,len(list)):
if re.match(project_name, list[index].text):
Select(driver.find_element_by_id("project_id")).select_by_visible_text(list[index].text)
no_project = False
break
if no_project:
self.assertTrue(False,'No project found with given project name')
driver.find_element_by_id("cluster_name").clear()
cluster_name = 'test_cluster' + str(randint(0,9999))
driver.find_element_by_id("cluster_name").send_keys(cluster_name)
hadoop_image = 'Hadoop-2.5.2'
Select(driver.find_element_by_id("os_systems")).select_by_visible_text(hadoop_image)
cluster_sizes = driver.find_element_by_id("size_of_cluster").text
try:
current_cluster_size = int(cluster_sizes.rsplit('\n', 1)[-1])
print ('Project '+ project_name +' has enough vms to run the test')
if ((user_quota['cpus']['available']-2*kamaki_flavors['cpus'][0]) >= 0):
driver.find_element_by_id("master_cpus_" + str(kamaki_flavors['cpus'][0])).click()
try: self.assertEqual("CPUs: {0}".format(str(kamaki_flavors['cpus'][0])), driver.find_element_by_id("master_cpu_summary").text)
except AssertionError as e: self.verificationErrors.append(str(e))
driver.find_element_by_id("slaves_cpus_" + str(kamaki_flavors['cpus'][0])).click()
try: self.assertEqual("CPUs each: {0}".format(str(kamaki_flavors['cpus'][0])), driver.find_element_by_id("slaves_cpu_summary").text)
except AssertionError as e: self.verificationErrors.append(str(e))
else:
self.assertTrue(False,'Not enought cpu to run the test')
if ((user_quota['ram']['available']-2*kamaki_flavors['ram'][0]) >= 0):
driver.find_element_by_id("master_ram_" + str(kamaki_flavors['ram'][0])).click()
try: self.assertEqual("RAM: {0}".format(str(kamaki_flavors['ram'][0])), driver.find_element_by_id("master_ram_summary").text)
except AssertionError as e: self.verificationErrors.append(str(e))
driver.find_element_by_id("slaves_ram_" + str(kamaki_flavors['ram'][0])).click()
try: self.assertEqual("RAM each: {0}".format(str(kamaki_flavors['ram'][0])), driver.find_element_by_id("slaves_ram_summary").text)
except AssertionError as e: self.verificationErrors.append(str(e))
else:
self.assertTrue(False,'Not enought ram to run the test')
if ((user_quota['disk']['available']-2*kamaki_flavors['disk'][0]) >= 0):
driver.find_element_by_id("master_disk_" + str(kamaki_flavors['disk'][0])).click()
try: self.assertEqual("Disk Size: {0}".format(str(kamaki_flavors['disk'][0])), driver.find_element_by_id("master_disk_summary").text)
except AssertionError as e: self.verificationErrors.append(str(e))
driver.find_element_by_id("slaves_disk_" + str(kamaki_flavors['disk'][0])).click()
try: self.assertEqual("Disk Size each: {0}".format(str(kamaki_flavors['disk'][0])), driver.find_element_by_id("slaves_disk_summary").text)
except AssertionError as e: self.verificationErrors.append(str(e))
else:
self.assertTrue(False,'Not enought disk to run the test')
except:
# self.assertTrue(False,'Not enought vms to run the test')
print ('Project '+ project_name +' has not enough vms to run the test')
开发者ID:amathilda，项目名称:e-science，代码行数:87，代码来源:test_text_respond_to_buttons.py
示例6: test_buttons_availability_respond_to_cluster_size_change
def test_buttons_availability_respond_to_cluster_size_change(self):
driver = self.driver
driver.get(self.base_url + "#/homepage")
driver.find_element_by_id("id_login").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_login_route"))
)
except: self.fail("time out")
driver.find_element_by_id("token").clear()
driver.find_element_by_id("token").send_keys(self.token)
driver.find_element_by_xpath("//button[@type='login']").click()
if (self.is_element_present(By.XPATH, "//div[@id='id_alert_wrongtoken']/strong") == True):
self.assertTrue(False,'Invalid token')
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_welcome_route"))
)
except: self.fail("time out")
driver.find_element_by_id("id_services_dd").click()
driver.find_element_by_id("id_create_cluster").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_cluster_create_route"))
)
except: self.fail("time out")
auth = check_credentials(self.token)
try:
list_of_projects = auth.get_projects(state='active')
except Exception:
self.assertTrue(False,'Could not get list of projects')
kamaki_flavors = get_flavor_id(self.token)
for project in list_of_projects:
user_quota = check_quota(self.token, project['id'])
if project['name'] == 'system:' + project['id']:
project_name = 'system'
else:
project_name = project['name']
list = Select(driver.find_element_by_id("project_id")).options
no_project = True
for index in range(0,len(list)):
if re.match(project_name, list[index].text):
Select(driver.find_element_by_id("project_id")).select_by_visible_text(list[index].text)
no_project = False
break
if no_project:
self.assertTrue(False,'No project found with given project name')
driver.find_element_by_id("cluster_name").clear()
cluster_name = 'test_cluster' + str(randint(0,9999))
driver.find_element_by_id("cluster_name").send_keys(cluster_name)
hadoop_image = 'Hadoop-2.5.2'
Select(driver.find_element_by_id("os_systems")).select_by_visible_text(hadoop_image)
flag = False
cluster_sizes = driver.find_element_by_id("size_of_cluster").text
try:
current_cluster_size = int(cluster_sizes.rsplit('\n', 1)[-1])
print ('Project '+ project_name +' has enough vms to run the test')
for cluster_size_selection in range(3,user_quota['cluster_size']['available']):
for flavor in ['cpus' , 'ram' , 'disk']:
for master in kamaki_flavors[flavor]:
for slaves in reversed(kamaki_flavors[flavor]):
if (((user_quota[flavor]['available'] - (master + slaves)) >= 0) and ((user_quota[flavor]['available'] - (master + (cluster_size_selection-1)*slaves)) < 0)):
initial_cluster_size = driver.find_element_by_id("size_of_cluster").text
button_id = 'master' + '_' + flavor + '_' + str(master)
driver.find_element_by_id(button_id).click()
button_id = 'slaves' + '_' + flavor + '_' + str(master)
driver.find_element_by_id(button_id).click()
current_cluster_sizes = driver.find_element_by_id("size_of_cluster").text
try: self.assertNotEqual(initial_cluster_size, driver.find_element_by_id("size_of_cluster").text)
except AssertionError as e: self.verificationErrors.append(str(e))
flag = True
break
if flag: break
if flag: break
if flag: break
if flag: break
if not flag:
self.assertTrue(False,'Not enought vms to see a change in cluster size')
except:
flag = True
# self.assertTrue(False,'Not enought vms to run the test')
print ('Project '+ project_name +' has not enough vms to run the test')
开发者ID:amathilda，项目名称:e-science，代码行数:82，代码来源:test_cluster_size_respond_to_buttons_clicks.py
示例7: test_buttons_availability_respond_based_on_user_quota
def test_buttons_availability_respond_based_on_user_quota(self):
driver = self.driver
driver.get(self.base_url + "#/homepage")
driver.find_element_by_id("id_login").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_login_route"))
)
except: self.fail("time out")
driver.find_element_by_id("token").clear()
driver.find_element_by_id("token").send_keys(self.token)
driver.find_element_by_xpath("//button[@type='login']").click()
if (self.is_element_present(By.XPATH, "//div[@id='id_alert_wrongtoken']/strong") == True):
self.assertTrue(False,'Invalid token')
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_user_welcome_route"))
)
except: self.fail("time out")
driver.find_element_by_id("id_services_dd").click()
driver.find_element_by_id("id_create_cluster").click()
try:
element = WebDriverWait(driver, 30).until(
EC.presence_of_element_located((By.ID, "id_title_cluster_create_route"))
)
except: self.fail("time out")
auth = check_credentials(self.token)
try:
list_of_projects = auth.get_projects(state='active')
except Exception:
logging.error(' Could not get list of projects')
kamaki_flavors = get_flavor_lists(self.token)
for project in list_of_projects:
user_quota = check_quota(self.token, project['id'])
if project['name'] == 'system:' + project['id']:
project_name = 'system'
else:
project_name = project['name']
list = Select(driver.find_element_by_id("project_id")).options
no_project = True
for index in range(0,len(list)):
if re.match(project_name, list[index].text):
Select(driver.find_element_by_id("project_id")).select_by_visible_text(list[index].text)
no_project = False
break
if no_project:
self.assertTrue(False,'No project found with given project name')
driver.find_element_by_id("cluster_name").clear()
cluster_name = 'test_cluster' + str(randint(0,9999))
driver.find_element_by_id("cluster_name").send_keys(cluster_name)
hadoop_image = 'Hadoop-2.5.2'
Select(driver.find_element_by_id("os_systems")).select_by_visible_text(hadoop_image)
for role in ["master" , "slaves"]:
for flavor in ['cpus' , 'ram' , 'disk']:
for item in kamaki_flavors[flavor]:
button_id = role + '_' + flavor + '_' + str(item)
if ((user_quota[flavor]['available']-(item + kamaki_flavors[flavor][0])) >= 0):
on = driver.find_element_by_id(button_id)
try: self.assertTrue(on.is_enabled())
except AssertionError as e: self.verificationErrors.append(str(e))
else:
off = driver.find_element_by_id(button_id)
try: self.assertFalse(off.is_enabled())
except AssertionError as e: self.verificationErrors.append(str(e))
开发者ID:grnet，项目名称:e-science，代码行数:64，代码来源:test_buttons_availability_respond_based_on_user_quota.py
注：本文中的backend.okeanos_utils.check_credentials函数示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论