import sys
import os
import requests as rq
from requests.exceptions import RequestException
import time
import json

# the max retry time
RETRY_TIME = 10

BASE_URL0 = "http://127.0.0.1:8300/api/v1"
BASE_URL1 = "http://127.0.0.1:8301/api/v1"


BASE_URL0_V2 = "http://127.0.0.1:8300/api/v2"
BASE_URL1_V2 = "http://127.0.0.1:8301/api/v2"

PD_ADDR = "http://127.0.0.1:2379"
# This value is generated by:
# echo -n '123456' | base64
# MTIzNDU2
# Use this value here to test redo apply function works well
# when use base64 encoded password
ENPASSWORD = "MTIzNDU2"
SINK_URI = "mysql://normal:%s@127.0.0.1:3306/" % ENPASSWORD

physicalShiftBits = 18


def requests_get_with_retry(url, max_retries=RETRY_TIME, delay_seconds=1):
    """
    requests get with retry

    :param url: request url
    :param max_retries: max retry times
    :param delay_seconds: retry delay seconds
    :return: when success, return response, else return None
    """
    for retry in range(max_retries):
        try:
            response = rq.get(url)
            if response.status_code == 200 or response.status_code == 202:
                return response
        except RequestException as e:
            print(f"request fails {retry + 1}/{max_retries} time retry...")
            time.sleep(delay_seconds)
    return None

# we should write some SQLs in the run.sh after call create_changefeed


def create_changefeed(sink_uri):
    url = BASE_URL1+"/changefeeds"
    # create changefeed
    for i in range(1, 4):
        data = {
            "changefeed_id": "changefeed-test"+str(i),
            "sink_uri": "blackhole://",
            "ignore_ineligible_table": True
        }
        # set sink_uri
        if i == 1 and sink_uri != "":
            data["sink_uri"] = sink_uri

        data = json.dumps(data)
        headers = {"Content-Type": "application/json"}
        resp = rq.post(url, data=data, headers=headers)
        assert resp.status_code == rq.codes.accepted

    # create changefeed fail because sink_uri is invalid
    data = json.dumps({
        "changefeed_id": "changefeed-test",
        "sink_uri": "mysql://127.0.0.1:1111",
        "ignore_ineligible_table": True
    })
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.bad_request

    # create changefeed fail because dispatcher is invalid
    url = BASE_URL1_V2+"/changefeeds?keyspace=keyspace1"
    data = json.dumps({
        "changefeed_id": "changefeed-test-v2",
        "sink_uri": "pulsar://127.0.0.1:6650/http_api_?protocol=canal-json",
        "replica_config": {
            "sink": {
                "dispatchers": [
                    {
                        "matcher": ["*.*"],
                        "partition": "columns",
                        "columns": ["a.b"],
                        "topic": "persistent://public/default/http_api_{schema}_{table}",
                    }
                ]
            }
        }
    })
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert "ErrDispatcherFailed" in resp.text, f"{resp.text}"

    # create changefeed fail because dispatcher is invalid
    url = BASE_URL1_V2+"/changefeeds"
    data = json.dumps({
        "changefeed_id": "changefeed-test-v2",
        "sink_uri": "kafka://127.0.0.1:9092/http_api?protocol=simple",
        "replica_config": {
            "sink": {
                "dispatchers": [
                    {
                        "matcher": ["*.*"],
                        "partition": "columns",
                        "columns": ["a.b"]
                    }
                ]
            }
        }
    })
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert "ErrDispatcherFailed" in resp.text, f"{resp.text}"

    # create changefeed fail because glue schema config is invalid
    url = BASE_URL1_V2+"/changefeeds"
    data = json.dumps({
        "changefeed_id": "changefeed-test-v2",
        "sink_uri": "kafka://127.0.0.1:9092/http_api?protocol=avro",
        "replica_config": {
            "sink": {
                "kafka_config": {
                    "glue_schema_registry_config": {
                        "region": "us-west-1",
                        "registry_name": "ticdc-test"
                    }
                }
            }
        }
    })
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert "CDC:ErrKafkaNewProducer" in resp.text, f"{resp.text}"
    assert "not found, ResolveEndpointV2" not in resp.text, f"{resp.text}"
    print("pass test: create changefeed")


def list_changefeed():
    # test state: all
    url = BASE_URL0+"/changefeeds?state=all"
    resp = rq.get(url)
    assert resp.status_code == rq.codes.ok

    # test state: normal
    url = BASE_URL0+"/changefeeds?state=normal"
    resp = rq.get(url)
    assert resp.status_code == rq.codes.ok
    data = resp.json()
    for changefeed in data:
        assert changefeed["state"] == "normal"

    # test state: stopped
    url = BASE_URL0+"/changefeeds?state=stopped"
    resp = rq.get(url)
    assert resp.status_code == rq.codes.ok
    data = resp.json()
    for changefeed in data:
        assert changefeed["state"] == "stopped"

    print("pass test: list changefeed")


def get_changefeed():
    # test get changefeed success
    url = BASE_URL0+"/changefeeds/changefeed-test1"
    resp = rq.get(url)
    assert resp.status_code == rq.codes.ok

    url = BASE_URL0+"/changefeeds/changefeed-test2"
    resp = rq.get(url)
    assert resp.status_code == rq.codes.ok

    # test get changefeed failed
    url = BASE_URL0+"/changefeeds/changefeed-not-exists"
    resp = rq.get(url)
    assert resp.status_code == rq.codes.bad_request
    data = resp.json()
    assert data["error_code"] == "CDC:ErrChangeFeedNotExists"

    print("pass test: get changefeed")


def pause_changefeed():
    # pause changefeed
    url = BASE_URL0+"/changefeeds/changefeed-test2/pause"
    for i in range(RETRY_TIME):
        resp = rq.post(url)
        if resp.status_code == rq.codes.accepted:
            break
        time.sleep(1)
    assert resp.status_code == rq.codes.accepted
    # check if pause changefeed success
    url = BASE_URL0+"/changefeeds/changefeed-test2"
    for i in range(RETRY_TIME):
        resp = rq.get(url)
        assert resp.status_code == rq.codes.ok
        data = resp.json()
        if data["state"] == "stopped":
            break
        time.sleep(1)
    assert data["state"] == "stopped"
    # test pause changefeed failed
    url = BASE_URL0+"/changefeeds/changefeed-not-exists/pause"
    resp = rq.post(url)
    assert resp.status_code == rq.codes.bad_request
    data = resp.json()
    assert data["error_code"] == "CDC:ErrChangeFeedNotExists"

    print("pass test: pause changefeed")


def update_changefeed():
    # update fail
    # can only update a stopped changefeed
    url = BASE_URL0+"/changefeeds/changefeed-test1"
    data = json.dumps({"mounter_worker_num": 32})
    headers = {"Content-Type": "application/json"}
    resp = rq.put(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.bad_request

    # update success
    url = BASE_URL0+"/changefeeds/changefeed-test2"
    data = json.dumps({"mounter_worker_num": 32})
    headers = {"Content-Type": "application/json"}
    resp = rq.put(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.accepted

    # update fail
    # can't update start_ts
    url = BASE_URL0+"/changefeeds/changefeed-test2"
    data = json.dumps({"start_ts": 0})
    headers = {"Content-Type": "application/json"}
    resp = rq.put(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.bad_request

    # can't update dispatchers
    url = BASE_URL0_V2+"/changefeeds/changefeed-test2"
    data = json.dumps({
        "sink_uri": "kafka://127.0.0.1:9092/http_api?protocol=simple",
        "replica_config": {
            "sink": {
                "dispatchers": [
                    {
                        "matcher": ["*.*"],
                        "partition": "columns",
                        "columns": ["a.b"]
                    }
                ]
            }
        }
    })
    headers = {"Content-Type": "application/json"}
    resp = rq.put(url, data=data, headers=headers)
    assert "ErrDispatcherFailed" in resp.text, f"{resp.text}"

    print("pass test: update changefeed")


def resume_changefeed():
    # resume changefeed
    url = BASE_URL1+"/changefeeds/changefeed-test2/resume"
    resp = rq.post(url)
    assert resp.status_code == rq.codes.accepted

    # check if resume changefeed success
    url = BASE_URL1+"/changefeeds/changefeed-test2"
    for i in range(RETRY_TIME):
        resp = rq.get(url)
        assert resp.status_code == rq.codes.ok
        data = resp.json()
        if data["state"] == "normal":
            break
        time.sleep(1)
    assert data["state"] == "normal"

    # test resume changefeed failed
    url = BASE_URL0+"/changefeeds/changefeed-not-exists/resume"
    resp = rq.post(url)
    assert resp.status_code == rq.codes.bad_request
    data = resp.json()
    assert data["error_code"] == "CDC:ErrChangeFeedNotExists"

    print("pass test: resume changefeed")


def remove_changefeed():
    # remove changefeed
    url = BASE_URL0+"/changefeeds/changefeed-test3"
    resp = rq.delete(url)
    assert resp.status_code == rq.codes.accepted

    # check if remove changefeed success
    url = BASE_URL0+"/changefeeds/changefeed-test3"
    for i in range(RETRY_TIME):
        resp = rq.get(url)
        if resp.status_code == rq.codes.bad_request:
            break
        time.sleep(1)
    assert resp.status_code == rq.codes.bad_request
    assert resp.json()["error_code"] == "CDC:ErrChangeFeedNotExists"

    # test remove changefeed failed
    url = BASE_URL0+"/changefeeds/changefeed-not-exists"
    resp = rq.delete(url)
    assert (resp.status_code == rq.codes.bad_request or resp.status_code ==
            rq.codes.internal_server_error)
    data = resp.json()
    assert data["error_code"] == "CDC:ErrChangeFeedNotExists"

    print("pass test: remove changefeed")


def rebalance_table():
    # rebalance_table
    url = BASE_URL0 + "/changefeeds/changefeed-test1/tables/rebalance_table"
    resp = rq.post(url)
    assert resp.status_code == rq.codes.accepted

    print("pass test: rebalance table")


def move_table():
    # move table
    url = BASE_URL0 + "/changefeeds/changefeed-test1/tables/move_table"
    data = json.dumps({"capture_id": "test-aaa-aa", "table_id": 11})
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.accepted

    # move table fail
    # not allow empty capture_id
    data = json.dumps({"capture_id": "", "table_id": 11})
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.bad_request

    print("pass test: move table")


def resign_owner():
    url = BASE_URL1 + "/owner/resign"
    resp = rq.post(url)
    assert resp.status_code == rq.codes.accepted

    print("pass test: resign owner")


def list_capture():
    url = BASE_URL0 + "/captures"
    resp = requests_get_with_retry(url)
    assert resp.status_code == rq.codes.ok

    print("pass test: list captures")


def list_processor():
    url = BASE_URL0 + "/processors"
    resp = requests_get_with_retry(url)
    assert resp.status_code == rq.codes.ok

    print("pass test: list processors")


def get_processor():
    # list processor to get changefeed_id and capture_id
    base_url = BASE_URL0 + "/processors"
    resp = requests_get_with_retry(base_url)
    assert resp.status_code == rq.codes.ok
    data = resp.json()[0]
    time.sleep(2)
    url = base_url + "/changefeed-test1/" + data["capture_id"]
    resp = requests_get_with_retry(url)
    # print error message for debug
    if (resp.status_code != rq.codes.ok):
        print("request url", url)
        print("response status code:", resp.status_code)
    assert resp.status_code == rq.codes.ok

    # test capture_id error and cdc server no panic
    url = base_url + "/" + data["changefeed_id"] + "/" + "non-exist-capture-id"
    resp = rq.get(url)
    assert resp.status_code == rq.codes.bad_request

    print("pass test: get processors")


def check_health():
    url = BASE_URL0 + "/health"
    for i in range(RETRY_TIME):
        resp = rq.get(url)
        if resp.status_code == rq.codes.ok:
            break
        time.sleep(1)
    assert resp.status_code == rq.codes.ok

    print("pass test: check health")


def get_status():
    url = BASE_URL0 + "/status"
    resp = requests_get_with_retry(url)
    assert resp.status_code == rq.codes.ok
    assert resp.json()["is_owner"]

    print("pass test: get status")


def set_log_level():
    url = BASE_URL0 + "/log"
    data = json.dumps({"log_level": "debug"})
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.ok

    data = json.dumps({"log_level": "info"})
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.ok

    print("pass test: set log level")


def get_tso():
    # test state: all
    url = BASE_URL0_V2+"/tso"
    data = json.dumps({})
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.ok

    data = json.dumps({"pd_addrs": [PD_ADDR]})
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code == rq.codes.ok

    # wrong pd address
    data = json.dumps({"pd_addrs": ["http://127.0.0.1:2233"]})
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert resp.status_code != rq.codes.ok

    print("pass test: get tso")

# util functions define belows

# compose physical time and logical time into tso


def compose_tso(ps, ls):
    return (ps << physicalShiftBits) + ls


# arg1: test case name
# arg2: certificates dir
# arg3: sink uri
if __name__ == "__main__":

    # test all the case as the order list in this map
    FUNC_MAP = {
        # api v1
        "check_health": check_health,
        "get_status": get_status,
        "create_changefeed": create_changefeed,
        "list_changefeed": list_changefeed,
        "get_changefeed": get_changefeed,
        "pause_changefeed": pause_changefeed,
        "update_changefeed": update_changefeed,
        "resume_changefeed": resume_changefeed,
        "rebalance_table": rebalance_table,
        "move_table": move_table,
        "get_processor": get_processor,
        "list_processor": list_processor,
        "set_log_level": set_log_level,
        "remove_changefeed": remove_changefeed,
        "resign_owner": resign_owner,
        # api v2
        "get_tso": get_tso
    }

    func = FUNC_MAP[sys.argv[1]]
    if len(sys.argv) >= 2:
        func(*sys.argv[2:])
    else:
        func()