test_data_maintain.py

# -*- coding:utf-8 -*-
import unittest

from flask_app import global_app
from app.utils.db_tools import db_driver

delete_conncetion_sql = 'delete from project_info'

test_insert_connection_sql = '''insert into project_info
(id,project_no,db_ip,db_port,db_user
,db_password,mes_db_name,tiip_db_name,host_ip)
values
('1','jkq_test','47.97.206.38',9610,'huansi'
,'huansi','HSGmtMes','HSTIIP','192.168.4.116')
'''

delete_remote_conncetion_sql = 'delete from remote_server_info'

test_insert_remote_connection_sql = '''insert into remote_server_info
(id,project_no,server_ip,server_ssh_port,server_user
,server_password)
values
(1,'jkq_test','192.168.4.116','1111','root'
,'huansi.net')
'''


class TestDataMaintain(unittest.TestCase):
    def setUp(self):
        self.client = global_app.test_client()

    def tearDown(self):
        pass

    def test_get_db_connection(self):
        # 删除数据
        with db_driver as session:
            session.exec_sql(delete_conncetion_sql)
            session.exec_sql(test_insert_connection_sql)

        res = self.client.get('/connection/')
        self.assertEqual(res.status_code, 200)
        self.assertIsInstance(res.json, dict)
        self.assertEqual(res.json.get('db_password'), 'huansi')
        self.assertEqual(res.json.get('mes_db_name'), 'HSGmtMes')
        self.assertEqual(res.json.get('db_ip'), '47.97.206.38')
        self.assertEqual(res.json.get('project_no'), 'jkq_test')

    def test_set_db_conncetion(self):
        with db_driver as session:
            session.exec_sql(delete_conncetion_sql)

        headers = {
            'Content-Type': "application/json",
        }
        # 正常新增
        json_data = {"project_no": "jkq", "db_ip": "47.97.206.38", "db_port": "9610", "db_user": "huansi",
                     "db_password": "huansi", "mes_db_name": "HSGmtMes", "tiip_db_name": "HSTIIP"}

        res = self.client.post('/connection/', headers=headers, json=json_data)
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json['message'], '保存成功')

        with db_driver as session:
            project_info = session.retrive_sql('select * from project_info')

        self.assertEqual(project_info.get('db_password'), 'huansi')
        self.assertEqual(project_info.get('mes_db_name'), 'HSGmtMes')
        self.assertEqual(project_info.get('db_ip'), '47.97.206.38')
        self.assertEqual(project_info.get('project_no'), 'jkq')
        id = project_info['id']

        json_data = {"id": id, "project_no": "jkq_test", "db_ip": "47.97.206.38", "db_port": "9610",
                     "db_user": "huansi",
                     "db_password": "pwd", "mes_db_name": "HSGmtMes22", "tiip_db_name": "HSTIIP"}

        res = self.client.post('/connection/', headers=headers, json=json_data)
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json['message'], '保存成功')

        with db_driver as session:
            project_info = session.retrive_sql('select * from project_info')

        self.assertEqual(project_info.get('db_password'), 'pwd')
        self.assertEqual(project_info.get('mes_db_name'), 'HSGmtMes22')
        self.assertEqual(project_info.get('db_ip'), '47.97.206.38')
        self.assertEqual(project_info.get('project_no'), 'jkq_test')
        self.assertEqual(project_info.get('id'), id)

    def test_connecton_test(self):
        res = self.client.get('/connection/jkq_test/HSGmtMes/')
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json['message'], '连接成功')

        res = self.client.get('/connection/jkq_test/HSGmtMes2/')
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json['message'], '连接失败')

        res = self.client.get('/connection/jkq1/HSGmtMes/')
        self.assertEqual(res.status_code, 501)
        self.assertEqual(res.json['error_title'], '未找到【jkq1】的信息')

    def test_port(self):
        res = self.client.get('/connection/test_port/83/')
        self.assertEqual(res.status_code, 501)
        self.assertIn('连接', res.json['error_title'])

        res = self.client.get('/connection/test_port/82/')
        self.assertEqual(res.status_code, 200)

    def test_get_remote_server_connection(self):
        # 删除数据
        with db_driver as session:
            session.exec_sql(delete_remote_conncetion_sql)
            session.exec_sql(test_insert_remote_connection_sql)

        res = self.client.get('/connection/remote_server/')
        self.assertEqual(res.status_code, 200)
        self.assertIsInstance(res.json, dict)
        self.assertEqual(res.json.get('server_ssh_port'), '1111')
        self.assertEqual(res.json.get('project_no'), 'jkq_test')

    def test_set_remote_conncetion(self):
        with db_driver as session:
            session.exec_sql(delete_remote_conncetion_sql)

        headers = {
            'Content-Type': "application/json",
        }

        # 正常新增
        json_data = {"project_no": "jkq", "server_ip": "47.97.206.38", "server_ssh_port": "9610",
                     "server_user": "huansi",
                     "server_password": "huansi"}

        res = self.client.post('/connection/remote_server/', headers=headers, json=json_data)
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json['message'], '保存成功')

        with db_driver as session:
            project_info = session.retrive_sql('select * from remote_server_info')

        self.assertEqual(project_info.get('server_password'), 'huansi')
        self.assertEqual(project_info.get('server_ssh_port'), '9610')
        self.assertEqual(project_info.get('server_ip'), '47.97.206.38')
        self.assertEqual(project_info.get('project_no'), 'jkq')
        id = project_info['id']

        json_data = {"id": id, "project_no": "jkq_test", "server_ip": "192.168.4.116", "server_ssh_port": "1111",
                     "server_user": "root",
                     "server_password": "huansi.net"}

        res = self.client.post('/connection/remote_server/', headers=headers, json=json_data)
        self.assertEqual(res.status_code, 200)
        self.assertEqual(res.json['message'], '保存成功')

        with db_driver as session:
            project_info = session.retrive_sql('select * from remote_server_info')

        self.assertEqual(project_info.get('server_password'), 'huansi.net')
        self.assertEqual(project_info.get('server_ssh_port'), '1111')
        self.assertEqual(project_info.get('server_ip'), '192.168.4.116')
        self.assertEqual(project_info.get('project_no'), 'jkq_test')
        self.assertEqual(project_info.get('id'), id)