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

# init logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# the max retry time
RETRY_TIME = 20

BASE_URL = "http://127.0.0.1:8300/"

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 assert_status_code(resp, expected_code, url):
    """
    Assert response status code matches expected code with detailed error message

    Args:
        resp: Response object
        expected_code: Expected status code
        url: Request URL
    """
    try:
        if resp is None:
            body = "No response"
            return
        else:
            body = resp.json() if resp.content else "Empty response"
    except ValueError:
        body = resp.text

    assert resp.status_code == expected_code, f"""
    Expected status code: {expected_code}
    Actual status code: {resp.status_code}
    Response body: {body}
    URL: {url}
    """


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:
            logging.info(
                f"request fails {retry + 1}/{max_retries} time retry...")
            time.sleep(delay_seconds)
    return None

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


def create_changefeed(sink_uri):
    url = BASE_URL1_V2+"/changefeeds?keyspace=keyspace1"
    # 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 for the first changefeed
        if i == 1 and sink_uri != "":
            data["sink_uri"] = sink_uri

        data = json.dumps(data)
        headers = {"Content-Type": "application/json"}
        logging.info(f"Start to post request, URL: {url}")
        logging.info(f"Request body: {data}")
        resp = rq.post(url, data=data, headers=headers)
        logging.info(f"Response status code: {resp.status_code}")
        logging.info(f"Response body: {resp.json()}")
        assert_status_code(resp, rq.codes.ok, url)

    # 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)
    logging.info(f"response status code: {resp.status_code}")
    logging.info(f"response body: {resp.json()}")
    assert_status_code(resp, rq.codes.bad_request, url)

    # create changefeed fail because dispatcher is invalid
    url = BASE_URL1_V2+"/changefeeds?keyspace=keyspace1"
    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}"

    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 glue schema config is invalid
    url = BASE_URL1_V2+"/changefeeds?keyspace=keyspace1"
    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_V2+"/changefeeds?state=all&keyspace=keyspace1"

    # Add retry logic to wait for changefeeds
    # We need to retry because the coordinator need some time to sync the changefeed infos from etcd
    for _ in range(RETRY_TIME):
        resp = rq.get(url)
        assert_status_code(resp, rq.codes.ok, url)
        changefeeds = resp.json()["items"]
        if len(changefeeds) > 0:
            break
        logging.info("No changefeeds found, retrying...")
        time.sleep(1)

    assert len(changefeeds) > 0, "No changefeeds found after retries"

    # test state: normal
    url = BASE_URL0_V2+"/changefeeds?state=normal&keyspace=keyspace1"
    resp = rq.get(url)
    assert_status_code(resp, rq.codes.ok, url)
    data = resp.json()
    changefeeds = data["items"]
    for cf in changefeeds:
        assert cf["state"] == "normal"

    # test state: stopped
    url = BASE_URL0_V2+"/changefeeds?state=stopped&keyspace=keyspace1"
    resp = rq.get(url)
    assert_status_code(resp, rq.codes.ok, url)
    data = resp.json()
    changefeeds = data["items"]
    for cf in changefeeds:
        assert cf["state"] == "stopped"


def get_changefeed():
    # test get changefeed success
    url = BASE_URL0_V2+"/changefeeds/changefeed-test1?keyspace=keyspace1"
    resp = rq.get(url)
    assert_status_code(resp, rq.codes.ok, url)

    url = BASE_URL0_V2+"/changefeeds/changefeed-test2?keyspace=keyspace1"
    resp = rq.get(url)
    assert_status_code(resp, rq.codes.ok, url)

    # test get changefeed failed
    url = BASE_URL0_V2+"/changefeeds/changefeed-not-exists?keyspace=keyspace1"
    resp = rq.get(url)
    assert_status_code(resp, rq.codes.bad_request, url)
    data = resp.json()
    assert data["error_code"] == "CDC:ErrChangeFeedNotExists"


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


def update_changefeed():
    # update fail
    # can only update a stopped changefeed
    url = BASE_URL0_V2+"/changefeeds/changefeed-test1?keyspace=keyspace1"
    data = json.dumps({
        "replica_config": {
            "mounter": {
                "worker_num": 32
            }
        }
    })
    headers = {"Content-Type": "application/json"}
    resp = rq.put(url, data=data, headers=headers)
    assert_status_code(resp, rq.codes.bad_request, url)

    # update success
    url = BASE_URL0_V2+"/changefeeds/changefeed-test2?keyspace=keyspace1"
    data = json.dumps({
        "replica_config": {
            "mounter": {
                "worker_num": 32
            }
        }
    })
    headers = {"Content-Type": "application/json"}
    resp = rq.put(url, data=data, headers=headers)
    assert_status_code(resp, rq.codes.ok, url)

    # update fail
    # can't update start_ts
    url = BASE_URL0_V2+"/changefeeds/changefeed-test2?keyspace=keyspace1"
    data = json.dumps({"start_ts": 1})
    headers = {"Content-Type": "application/json"}
    resp = rq.put(url, data=data, headers=headers)
    assert_status_code(resp, rq.codes.bad_request, url)

    # can't update dispatchers
    url = BASE_URL0_V2+"/changefeeds/changefeed-test2?keyspace=keyspace1"
    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_V2+"/changefeeds/changefeed-test2/resume?keyspace=keyspace1"
    data = json.dumps({})
    headers = {"Content-Type": "application/json"}
    resp = rq.post(url, data=data, headers=headers)
    assert_status_code(resp, rq.codes.ok, url)

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

    # test resume changefeed failed
    url = BASE_URL0_V2+"/changefeeds/changefeed-not-exists/resume?keyspace=keyspace1"
    resp = rq.post(url)
    assert_status_code(resp, rq.codes.bad_request, url)
    data = resp.json()
    assert data["error_code"] == "CDC:ErrChangeFeedNotExists"


def remove_changefeed(cfID="changefeed-test3"):
    # remove changefeed
    url = BASE_URL0_V2+"/changefeeds/" + cfID + "?keyspace=keyspace1"
    resp = rq.delete(url)
    assert_status_code(resp, rq.codes.ok, url)

    # test remove non-exists changefeed, it should return 200 and do nothing
    url = BASE_URL0_V2+"/changefeeds/changefeed-not-exists?keyspace=keyspace1"
    resp = rq.delete(url)
    assert_status_code(resp, rq.codes.ok, url)


# FIXME: Enable this test case after we fully support move table API
def move_table(cfID="changefeed-test1"):
    return
    # sleep 5 seconds to make sure all tables is scheduled
    time.sleep(5)

    # find the node id
    url = BASE_URL0_V2 + "/captures"
    resp = requests_get_with_retry(url)
    data = resp.json()
    capture_id = data["items"][0]["id"]
    logging.info(f"Find target capture_id: {capture_id}")

    # find the table id
    url = BASE_URL0_V2 + "/changefeeds/" + cfID + "/tables?keyspace=keyspace1"
    resp = requests_get_with_retry(url)
    data = resp.json()
    table_ids = data[0]["table_ids"]
    table_id = 0
    for id in table_ids:
        if id > 0:
            table_id = id
            break

    logging.info(f"Find target table_id: {table_id}")

    # move table
    url = BASE_URL0_V2 + "/changefeeds/" + cfID + \
        "/move_table?targetNodeID=" + capture_id + \
        "&tableID=" + str(table_id) + "&keyspace=keyspace1"
    resp = rq.post(url)
    assert_status_code(resp, rq.codes.ok, url)
    logging.info(f"Move table success")
    # move table fail
    # The target node is not found
    url = BASE_URL0_V2 + "/changefeeds/" + cfID + \
        "/move_table?targetNodeID=&tableID=" + \
        str(table_id) + "&keyspace=keyspace1"
    resp = rq.post(url)
    assert_status_code(resp, rq.codes.internal_server_error, url)


def resign_owner():
    url = BASE_URL1_V2 + "/owner/resign"
    resp = rq.post(url)
    assert_status_code(resp, rq.codes.ok, url)


def list_capture():
    url = BASE_URL0_V2 + "/captures"
    resp = requests_get_with_retry(url)
    assert_status_code(resp, rq.codes.ok, url)


def check_health():
    url = BASE_URL0_V2 + "/health"
    resp = requests_get_with_retry(url)
    assert_status_code(resp, rq.codes.ok, url)


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


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

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


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_status_code(resp, rq.codes.ok, url)

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

    # wrong pd address
    # FIXME: When we support multiple upstream, we need to support this case
    # 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_status_code(resp, rq.codes.bad_request, url)


# 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,
        "move_table": move_table,
        "set_log_level": set_log_level,
        "remove_changefeed": remove_changefeed,
        "resign_owner": resign_owner,
        # api v2
        "get_tso": get_tso
    }

    # check the test case name
    if len(sys.argv) < 2:
        logging.error("Please provide a test case name")
        sys.exit(1)

    # get the test case name
    test_case_name = sys.argv[1]
    arg = sys.argv[2:]
    # check if the test case name is in the FUNC_MAP
    if test_case_name not in FUNC_MAP:
        logging.error(f"Test case {test_case_name} not found")
        sys.exit(1)

    # get the test case function
    test_case_func = FUNC_MAP[test_case_name]
    # run the test case
    logging.info(f"Start to run test case: {test_case_name}")
    test_case_func(*arg)
    logging.info(f"Test case {test_case_name} finished")