最近有個(gè)需求捍掺，在web端創(chuàng)建多種類型的數(shù)據(jù)源臣樱，會(huì)給出連接鏈接和連接賬號(hào)密碼靶擦。然后需要測(cè)試數(shù)據(jù)源是否可連接，給出的賬號(hào)能否進(jìn)行增刪改查等操作雇毫。目前支持的數(shù)據(jù)源有mongodb玄捕、redis、es棚放、ck枚粘、sqlserver、postgresql飘蚯。

#!/usr/bin/env python
# _*_ coding: utf-8 _*_

from redis import StrictRedis
import pymongo
from elasticsearch6 import Elasticsearch
import pymssql
import psycopg2
from clickhouse_driver import Client


# 連接信息
redis_info = ['10.10.XXX', XXX]
mongo_info = ['10.10.XXX', XXX]
es_info = ['10.10.XXX', XXX]
sqlserver_info = ['10.10.XXX:XXX', 'XXX', 'XXX', 'XXX']
db2_info = ['10.10.XXX', XXX, 'XXX', 'XXX']
postgre_info = ['XXX', 'XXX', 'XXX', '10.10.XXX', XXX]
ck_info = ['10.10.XXX', XXX]


def test_ck():
    print('------------ck 測(cè)試開始------------')
    client = Client(host=ck_info[0], port=ck_info[1])
    try:
        client.execute('CREATE DATABASE IF NOT EXISTS ddd_test')
        client.execute('USE ddd_test')
        client.execute('DROP TABLE IF EXISTS test')
        client.execute('CREATE TABLE test (x Int32) ENGINE = Memory')
        client.execute(
            'INSERT INTO test (x) VALUES',
            [{'x': 1}, {'x': 2}, {'x': 3}, {'x': 100}]
        )
        client.execute('INSERT INTO test (x) VALUES', [[200]])
        print(client.execute('SELECT sum(x) FROM test'))
        print('------------ck 測(cè)試成功------------')
    except Exception as ex:
        print(ex)
        print('------------ck 測(cè)試失敗------------')


def test_postgresql():
    print('------------postgresql 測(cè)試開始------------')
    conn = psycopg2.connect(database=postgre_info[0], user=postgre_info[1], password=postgre_info[2],
                            host=postgre_info[3], port=postgre_info[4])
    print("Opened database successfully")
    try:
        cur = conn.cursor()
        cur.execute('drop table COMPANY;')
        cur.execute('''CREATE TABLE COMPANY
               (ID INT PRIMARY KEY     NOT NULL,
               NAME           TEXT    NOT NULL,
               AGE            INT     NOT NULL,
               ADDRESS        CHAR(50),
               SALARY         REAL);''')
        print("Table created successfully")
        cur.execute("INSERT INTO COMPANY (ID,NAME,AGE,ADDRESS,SALARY) VALUES (4, 'Mark', 25, 'Rich-Mond ', 65000.00 )");
        cur.execute("SELECT id, name, address, salary  from COMPANY")
        rows = cur.fetchall()
        for row in rows:
            print("ID = ", row[0])
            print("NAME = ", row[1])
            print("ADDRESS = ", row[2])
            print("SALARY = ", row[3], "\n")

        print("Operation done successfully");

        conn.commit()
        conn.close()
        print('------------postgresql 測(cè)試成功------------')
    except Exception as ex:
        print(ex)
        print('------------postgresql 操作失敗------------')


def test_sqlserver():
    """
    測(cè)試sqlserver
    """
    print('------------sql server測(cè)試開始------------')
    conn = pymssql.connect(host=sqlserver_info[0], user=sqlserver_info[1], password=sqlserver_info[2],
                           database=sqlserver_info[3],
                           charset='utf8')
    cur = conn.cursor()
    if not cur:
        print('連接失敗')
        return

    sql_createtable = 'create table classify ( id int primary key identity(1,1), name varchar(20) not null )'
    sql_insert = "insert into classify (name) values('abc')"
    sql_update = "update classify set name='bbc' where id = 1"
    sql_select = 'select * from classify where id = 1'
    sql_delete = 'delete from classify where id = 1'
    try:
        cur.execute(sql_createtable)
        cur.execute(sql_insert)
        cur.execute(sql_update)
        cur.execute(sql_select)
        result = cur.fetchall()
        # cur.execute(sql_delete)
        conn.commit()
        conn.close()
        print(str(result))
        print("------------sql server測(cè)試成功------------")
    except Exception as ex:
        print(ex)
        print('------------sql server測(cè)試失敗------------')


def test_redis():
    try:
        print("------------redis測(cè)試開始------------")
        redis = StrictRedis(host=redis_info[0], port=redis_info[1], db=0, password='admin')
        redis.set('name', 'Bob')
        redis.set('name', 'mary')
        print(redis.get('name'))
        redis.delete('name')
        print("------------redis測(cè)試成功------------")
    except Exception as ex:
        print(ex)
        print("------------redis測(cè)試失敗------------")


def test_mongodb():
    print("------------mongdb測(cè)試開始------------")
    try:
        client = pymongo.MongoClient(host=mongo_info[0], port=mongo_info[1])
        mongo_auth = client.admin
        mongo_auth.authenticate('root', 'koala321')

        mongo_db = client['test_db']
        mongo_coll = mongo_db['test_coll']
        param = {"name": "jack", "age": "12"}
        # mongo_coll.insert_one(param)
        result = mongo_coll.find_one(param)
        print(result)

        print("------------mongdb測(cè)試成功------------")
    except Exception as ex:
        print(ex)
        print("------------mongodb測(cè)試失敗------------")


def test_es():
    print('------------es測(cè)試開始------------')
    try:
        es = Elasticsearch([{'host': es_info[0], 'port': es_info[1]}], timeout=3600)
        doc = {
            'article_id': 1,
            'user_id': 1,
            'title': 1
        }
        es.index(index='articles', doc_type='article', body=doc, id=1)
        query = {
            'query': {
                'bool': {
                    'must': [
                        {'match': {'_all': 'python web'}}
                    ],
                    'filter': [
                        {'term': {'status': 2}}
                    ]
                }
            }
        }
        ret = es.search(index='articles', doc_type='article', body=query)
        print(ret)

        print('------------es測(cè)試成功------------')
    except Exception as ex:
        print(ex)
        print('------------es測(cè)試失敗------------')


if __name__ == '__main__':
    test_ck()