Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use verify_connectivity method in autotest

Best Python code snippet using autotest_python

app.py

Source:app.py

...23def index():24    connectivity = True if conn.connectivity else False25    if request.method == 'POST':26        if 'check_db' in request.form:27            conn.verify_connectivity()28            connectivity = True if conn.connectivity else False29    return render_template('index.html', connectivity=connectivity)30@app.route('/query1', methods=('GET', 'POST'))31def query1():32    connectivity = True if conn.connectivity else False33    response = None34    if request.method == 'POST':35        if 'check_db' in request.form:36            conn.verify_connectivity()37            connectivity = True if conn.connectivity else False38        elif 'run_query' in request.form:39            response = conn.run_query(query_1)40    return render_template('query1.html',41                           connectivity=connectivity,42                           response=response43                           )44@app.route('/query2', methods=('GET', 'POST'))45def query2():46    connectivity = True if conn.connectivity else False47    response = None48    if request.method == 'POST':49        if 'check_db' in request.form:50            conn.verify_connectivity()51            connectivity = True if conn.connectivity else False52        elif 'run_query' in request.form:53            response = conn.run_query(query_2)54    return render_template('query2.html',55                           connectivity=connectivity,56                           response=response57                           )58@app.route('/query3', methods=('GET', 'POST'))59def query3():60    connectivity = True if conn.connectivity else False61    response = None62    if request.method == 'POST':63        if 'check_db' in request.form:64            conn.verify_connectivity()65            connectivity = True if conn.connectivity else False66        elif 'run_query' in request.form:67            title = request.form['title']68            parameters = {'title': title}69            response = conn.run_query(query_3, parameters)70    return render_template('query3.html',71                           connectivity=connectivity,72                           response=response73                           )74@app.route('/query4', methods=('GET', 'POST'))75def query4():76    connectivity = True if conn.connectivity else False77    response = None78    if request.method == 'POST':79        if 'check_db' in request.form:80            conn.verify_connectivity()81            connectivity = True if conn.connectivity else False82        elif 'run_query' in request.form:83            genre = request.form['genre']84            min_episodes = float(request.form['min_episodes'])85            max_episodes = float(request.form['max_episodes'])86            max_results = int(request.form['max_results'])87            parameters = {'min_episodes': min_episodes,88                          'max_episodes': max_episodes,89                          'max_results': max_results90                          }91            response = conn.run_query(query_4.format(genre=genre),92                                      parameters)93    return render_template('query4.html',94                           connectivity=connectivity,95                           response=response96                           )97@app.route('/query5', methods=('GET', 'POST'))98def query5():99    connectivity = True if conn.connectivity else False100    response = None101    if request.method == 'POST':102        if 'check_db' in request.form:103            conn.verify_connectivity()104            connectivity = True if conn.connectivity else False105        elif 'run_query' in request.form:106            max_results = int(request.form['max_results'])107            parameters = {'max_results': max_results}108            response = conn.run_query(query_5, parameters)109    return render_template('query5.html',110                           connectivity=connectivity,111                           response=response112                           )113@app.route('/query6', methods=('GET', 'POST'))114def query6():115    connectivity = True if conn.connectivity else False116    response = None117    if request.method == 'POST':118        if 'check_db' in request.form:119            conn.verify_connectivity()120            connectivity = True if conn.connectivity else False121        elif 'run_query' in request.form:122            gender = request.form['gender']123            title = request.form['title']124            max_results = int(request.form['max_results'])125            parameters = {'gender': gender,126                          'title': title,127                          'max_results': max_results128                          }129            response = conn.run_query(query_6, parameters)130            a_nodes = [title]131            p_nodes = [x['user'] for x in response]132            edges = [(x, title) for x in p_nodes]133            create_network(a_nodes, p_nodes, edges)134    return render_template('query6.html',135                           connectivity=connectivity,136                           response=response137                           )138@app.route('/query7', methods=('GET', 'POST'))139def query7():140    connectivity = True if conn.connectivity else False141    response = None142    if request.method == 'POST':143        if 'check_db' in request.form:144            conn.verify_connectivity()145            connectivity = True if conn.connectivity else False146        elif 'run_query' in request.form:147            name = request.form['name']148            parameters = {'name': name}149            response = conn.run_query(query_7, parameters)150            a_nodes = [x['title'] for x in response]151            p_nodes = [name]152            edges = [(name, x) for x in a_nodes]153            create_network(a_nodes, p_nodes, edges)154    return render_template('query7.html',155                           connectivity=connectivity,156                           response=response157                           )158@app.route('/query8', methods=('GET', 'POST'))159def query8():160    connectivity = True if conn.connectivity else False161    response = None162    if request.method == 'POST':163        if 'check_db' in request.form:164            conn.verify_connectivity()165            connectivity = True if conn.connectivity else False166        elif 'run_query' in request.form:167            title = request.form['title']168            parameters = {'title': title}169            response = conn.run_query(query_8, parameters)170    return render_template('query8.html',171                           connectivity=connectivity,172                           response=response173                           )174@app.route('/query9', methods=('GET', 'POST'))175def query9():176    connectivity = True if conn.connectivity else False177    response = None178    if request.method == 'POST':179        if 'check_db' in request.form:180            conn.verify_connectivity()181            connectivity = True if conn.connectivity else False182        elif 'run_query' in request.form:183            genre = request.form['genre']184            gender = request.form['gender']185            min_score = float(request.form['min_score'])186            max_score = float(request.form['max_score'])187            max_results = int(request.form['max_results'])188            parameters = {'gender': gender,189                          'min_score': min_score,190                          'max_score': max_score,191                          'max_results': max_results192                          }193            response = conn.run_query(query_9.format(genre=genre), parameters)194    return render_template('query9.html',195                           connectivity=connectivity,196                           response=response197                           )198@app.route('/query10', methods=('GET', 'POST'))199def query10():200    connectivity = True if conn.connectivity else False201    response = None202    if request.method == 'POST':203        if 'check_db' in request.form:204            conn.verify_connectivity()205            connectivity = True if conn.connectivity else False206        elif 'run_query' in request.form:207            max_results = int(request.form['max_results'])208            parameters = {'max_results': max_results}209            response = conn.run_query(query_10, parameters)210    return render_template('query10.html',211                           connectivity=connectivity,212                           response=response213                           )214@app.route('/query11', methods=('GET', 'POST'))215def query11():216    connectivity = True if conn.connectivity else False217    response = None218    if request.method == 'POST':219        if 'check_db' in request.form:220            conn.verify_connectivity()221            connectivity = True if conn.connectivity else False222        elif 'run_query' in request.form:223            title_1 = request.form['title_1']224            title_2 = request.form['title_2']225            title_3 = request.form['title_3']226            parameters = {'title_1': title_1,227                          'title_2': title_2,228                          'title_3': title_3229                          }230            response = conn.run_query(query_11, parameters)231            a_nodes = [title_1, title_2, title_3]232            p_nodes = [x['user'] for x in response]233            edges = [(x, y) for x in p_nodes for y in a_nodes]234            create_network(a_nodes, p_nodes, edges)235    return render_template('query11.html',236                           connectivity=connectivity,237                           response=response238                           )239@app.route('/query12', methods=('GET', 'POST'))240def query12():241    connectivity = True if conn.connectivity else False242    response = None243    if request.method == 'POST':244        if 'check_db' in request.form:245            conn.verify_connectivity()246            connectivity = True if conn.connectivity else False247        elif 'run_query' in request.form:248            title = request.form['title']249            gender = request.form['gender']250            max_results = int(request.form['max_results'])251            parameters = {'title': title,252                          'gender': gender,253                          'max_results': max_results}254            response = conn.run_query(query_12, parameters)255    return render_template('query12.html',256                           connectivity=connectivity,257                           response=response258                           )259@app.route('/query13', methods=('GET', 'POST'))260def query13():261    connectivity = True if conn.connectivity else False262    response = None263    if request.method == 'POST':264        if 'check_db' in request.form:265            conn.verify_connectivity()266            connectivity = True if conn.connectivity else False267        elif 'run_query' in request.form:268            title = request.form['title']269            max_results = int(request.form['max_results'])270            parameters = {'title': title,271                          'max_results': max_results272                          }273            response = conn.run_query(query_13, parameters)274    return render_template('query13.html',275                           connectivity=connectivity,276                           response=response277                           )278@app.route('/query14', methods=('GET', 'POST'))279def query14():280    connectivity = True if conn.connectivity else False281    response = None282    if request.method == 'POST':283        if 'check_db' in request.form:284            conn.verify_connectivity()285            connectivity = True if conn.connectivity else False286        elif 'run_query' in request.form:287            title_1 = request.form['title_1']288            title_2 = request.form['title_2']289            gender = request.form['gender']290            max_results = int(request.form['max_results'])291            parameters = {'title_1': title_1,292                          'title_2': title_2,293                          'gender': gender,294                          'max_results': max_results}295            response = conn.run_query(query_14, parameters)296    return render_template('query14.html',297                           connectivity=connectivity,298                           response=response299                           )300@app.route('/query15', methods=('GET', 'POST'))301def query15():302    connectivity = True if conn.connectivity else False303    response = None304    if request.method == 'POST':305        if 'check_db' in request.form:306            conn.verify_connectivity()307            connectivity = True if conn.connectivity else False308        elif 'run_query' in request.form:309            genre = request.form['genre']310            gender = request.form['gender']311            min_score = float(request.form['min_score'])312            max_results = int(request.form['max_results'])313            parameters = {'gender': gender,314                          'min_score': min_score,315                          'max_results': max_results316                          }317            response = conn.run_query(query_15.format(genre=genre), parameters)318    return render_template('query15.html',319                           connectivity=connectivity,320                           response=response...

test_check_deployment.py

Source:test_check_deployment.py

1#!/usr/bin/env python2# Copyright (c) 2017 Ericsson and others.3#4# All rights reserved. This program and the accompanying materials5# are made available under the terms of the Apache License, Version 2.06# which accompanies this distribution, and is available at7# http://www.apache.org/licenses/LICENSE-2.08import logging9import mock10import unittest11from functest.ci import check_deployment12__author__ = "Jose Lausuch <jose.lausuch@ericsson.com>"13class CheckDeploymentTesting(unittest.TestCase):14    """The super class which testing classes could inherit."""15    # pylint: disable=missing-docstring16    logging.disable(logging.CRITICAL)17    def setUp(self):18        self.client_test = mock.Mock()19        self.deployment = check_deployment.CheckDeployment()20        self.service_test = 'compute'21        self.rc_file = self.deployment.rc_file22        self.endpoint_test = 'http://192.168.0.6:5000/v3'23        creds_attr = {'auth_url': self.endpoint_test,24                      'proxy_settings': ''}25        proxy_attr = {'host': '192.168.0.1', 'port': '5000'}26        proxy_settings = mock.Mock()27        proxy_settings.configure_mock(**proxy_attr)28        self.os_creds = mock.Mock()29        self.os_creds.configure_mock(**creds_attr)30        self.os_creds.proxy_settings = proxy_settings31        self.deployment.os_creds = self.os_creds32    def test_check_rc(self):33        with mock.patch('functest.ci.check_deployment.os.path.isfile',34                        returns=True) as m, \35                mock.patch('__builtin__.open',36                           mock.mock_open(read_data='OS_AUTH_URL')):37            self.deployment.check_rc()38            self.assertTrue(m.called)39    def test_check_rc_missing_file(self):40        with mock.patch('functest.ci.check_deployment.os.path.isfile',41                        return_value=False), \42                self.assertRaises(Exception) as context:43            msg = 'RC file {} does not exist!'.format(self.rc_file)44            self.deployment.check_rc(self.rc_file)45            self.assertTrue(msg in context)46    def test_check_rc_missing_os_auth(self):47        with mock.patch('__builtin__.open',48                        mock.mock_open(read_data='test')), \49                self.assertRaises(Exception) as context:50            msg = 'OS_AUTH_URL not defined in {}.'.format(self.rc_file)51            self.assertTrue(msg in context)52    def test_check_auth_endpoint(self):53        with mock.patch('functest.ci.check_deployment.verify_connectivity',54                        return_value=True) as m:55            self.deployment.check_auth_endpoint()56            self.assertTrue(m.called)57    def test_check_auth_endpoint_not_reachable(self):58        with mock.patch('functest.ci.check_deployment.verify_connectivity',59                        return_value=False) as m, \60                self.assertRaises(Exception) as context:61            endpoint = self.os_creds.auth_url62            self.deployment.check_auth_endpoint()63            msg = "OS_AUTH_URL {} is not reachable.".format(endpoint)64            self.assertTrue(m.called)65            self.assertTrue(msg in context)66    def test_check_public_endpoint(self):67        with mock.patch('functest.ci.check_deployment.verify_connectivity',68                        return_value=True) as m, \69                mock.patch('functest.ci.check_deployment.keystone_utils.'70                           'get_endpoint') as n:71            self.deployment.check_public_endpoint()72            self.assertTrue(m.called)73            self.assertTrue(n.called)74    def test_check_public_endpoint_not_reachable(self):75        with mock.patch('functest.ci.check_deployment.verify_connectivity',76                        return_value=False) as m, \77                mock.patch('functest.ci.check_deployment.keystone_utils.'78                           'get_endpoint',79                           return_value=self.endpoint_test) as n, \80                self.assertRaises(Exception) as context:81            self.deployment.check_public_endpoint()82            msg = ("Public endpoint {} is not reachable."83                   .format(self.mock_endpoint))84            self.assertTrue(m.called)85            self.assertTrue(n.called)86            self.assertTrue(msg in context)87    def test_check_service_endpoint(self):88        with mock.patch('functest.ci.check_deployment.verify_connectivity',89                        return_value=True) as m, \90                mock.patch('functest.ci.check_deployment.keystone_utils.'91                           'get_endpoint') as n:92            self.deployment.check_service_endpoint(self.service_test)93            self.assertTrue(m.called)94            self.assertTrue(n.called)95    def test_check_service_endpoint_not_reachable(self):96        with mock.patch('functest.ci.check_deployment.verify_connectivity',97                        return_value=False) as m, \98                mock.patch('functest.ci.check_deployment.keystone_utils.'99                           'get_endpoint',100                           return_value=self.endpoint_test) as n, \101                self.assertRaises(Exception) as context:102            self.deployment.check_service_endpoint(self.service_test)103            msg = "{} endpoint {} is not reachable.".format(self.service_test,104                                                            self.endpoint_test)105            self.assertTrue(m.called)106            self.assertTrue(n.called)107            self.assertTrue(msg in context)108    def test_check_nova(self):109        with mock.patch('functest.ci.check_deployment.nova_utils.nova_client',110                        return_value=self.client_test) as m:111            self.deployment.check_nova()112            self.assertTrue(m.called)113    def test_check_nova_fail(self):114        with mock.patch('functest.ci.check_deployment.nova_utils.nova_client',115                        return_value=self.client_test) as m, \116                mock.patch.object(self.client_test, 'servers.list',117                                  side_effect=Exception):118            self.deployment.check_nova()119            self.assertTrue(m.called)120            self.assertRaises(Exception)121    def test_check_neutron(self):122        with mock.patch('functest.ci.check_deployment.neutron_utils.'123                        'neutron_client', return_value=self.client_test) as m:124            self.deployment.check_neutron()125            self.assertTrue(m.called)126    def test_check_neutron_fail(self):127        with mock.patch('functest.ci.check_deployment.neutron_utils.'128                        'neutron_client',129                        return_value=self.client_test) as m, \130                mock.patch.object(self.client_test, 'list_networks',131                                  side_effect=Exception), \132                self.assertRaises(Exception):133            self.deployment.check_neutron()134            self.assertRaises(Exception)135            self.assertTrue(m.called)136    def test_check_glance(self):137        with mock.patch('functest.ci.check_deployment.glance_utils.'138                        'glance_client', return_value=self.client_test) as m:139            self.deployment.check_glance()140            self.assertTrue(m.called)141    def test_check_glance_fail(self):142        with mock.patch('functest.ci.check_deployment.glance_utils.'143                        'glance_client', return_value=self.client_test) as m, \144                mock.patch.object(self.client_test, 'images.list',145                                  side_effect=Exception):146            self.deployment.check_glance()147            self.assertRaises(Exception)148            self.assertTrue(m.called)149if __name__ == "__main__":150    logging.disable(logging.CRITICAL)...

neo4jconn.py

Source:neo4jconn.py

...5        self.connectivity = None6        self.driver = None7        try:8            self.driver = GraphDatabase.driver(uri, auth=(user, password))9            self.driver.verify_connectivity()10            self.connectivity = True11        except Exception as e:12            print("Cannot establish a connection to the database")13            print(e)14    def close(self):15        if self.driver:16            self.driver.close()17    def verify_connectivity(self):18        self.connectivity = None19        if self.driver:20            try:21                self.driver.verify_connectivity()22                self.connectivity = True23            except Exception as e:24                print("Failed to connect")25                print(e)26    def run_query(self, query, parameters=None):27        response = []28        if self.connectivity:29            try:30                with self.driver.session(database=self.db) as session:31                    response = list(session.run(query, parameters))32            except Exception as e:33                print("Query failed")34                print(e)35        return response

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.