python-base--/project-test/bigtree-aircp/flink.py

import unittest

import requests

from util import req, reqGet, reqForm


class MyTest(unittest.TestCase):
    def setUp(self) -> None:
        self.host = "http://127.0.0.1:8080"  # local

    def test623(self):
        """【数据源】新增成功后，获取列表数据返回错误"""
        res = reqGet("https://edsp.rxdptest.k5.bigtree.tech/api/rxdp-xdata-web/dataSource/pageList", {})
        print(len(res['data']))

        res = req("https://edsp.rxdptest.k5.bigtree.tech/api/rxdp-xdata-web/dataSource",
                  {"sourceType": "mysql", "sourceCode": "test23324", "sourceName": "test2134234", "sourceDesc": "",
                   "driverName": "com.mysql.cj.jdbc.Driver",
                   "jdbcUrl": "jdbc:mysql://mysql.rxdptest.svc.k5.bigtree.zone:3306/rxdp_auth", "username": "cs_user",
                   "password": "ceshimima.1",
                   "sourceConfig": "{\"driverName\":\"com.mysql.cj.jdbc.Driver\",\"jdbcUrl\":\"jdbc:mysql://mysql.rxdptest.svc.k5.bigtree.zone:3306/rxdp_auth\",\"username\":\"cs_user\",\"password\":\"ceshimima.1\"}"})
        print(res)

        res = reqGet("https://edsp.rxdptest.k5.bigtree.tech/api/rxdp-xdata-web/dataSource/pageList", {})
        print(len(res['data']))

    def testhttp(self):
        req("https://httpbin.org/post",
            {"username": "刘学良", "password": "lxl123456", "type": "account"})

    # 原子标签
    def testComputer(self):
        reqForm("http://localhost:8080/test/dealWithTagCompute",
                {"computeType": 3, "dataSourceCode": "tn_markting_db", "dataSourceType": "oracle",
                 "dimensionNo": "10610", "dimensionsName": "GENDER", "entityName": "tn营销基本信息",
                 "entityNo": "EN892456279547714785", "primaryKey": "CST_ID", "primaryTableName": "DC_CST_INFO",
                 "tagName": "性别0-未知的性别1-男性2-女性9-未说明的性别", "tagNo": "TAG16693487690724795",
                 "tenantId": "TENA883826260025957899"})

    # 衍生标签
    def testComputer2(self):
        reqForm("http://localhost:8080/test/dealWithTagCompute",
                {"computeType": 1, "dataSourceCode": "mysqlTestHu123", "dataSourceType": "mysql",
                 "entityName": "mysqlTestShopping33333(mysql数据源)", "entityNo": "EN875254841037837824",
                 "packageId": "rxdp/mySqlShoppingTest112", "primaryKey": "projectcode", "primaryTableName": "shopping",
                 "statisticsCycle": "year", "statisticsCycleTime": "createtime",
                 "tagName": "衍生标签测试多数据源mysql33333", "tagNo": "TAG16674623867383022", "timeStyle": "TS2",
                 "tenantId": "test-test-test2"})

    # sql标签
    def testComputer3(self):
        reqForm("http://localhost:8080/test/dealWithTagCompute",
                {"computeType": 2, "dataSourceCode": "mysqlTestHu123", "dataSourceType": "mysql",
                 "entityName": "mysql实体HU123", "entityNo": "EN879274694203750264", "primaryKey": "business_code",
                 "primaryTableName": "shopping",
                 "tagCompuleSql": "SELECT projectcode AS businessCode,amount AS tagValue FROM shopping LIMIT ${startNum},${endNum}",
                 "tagName": "SQL标签HU1112", "tagNo": "TAG16684123244372739", "tenantId": "test-test-test3"})