#!/usr/bin/env python3


import pytest
import time
from helpers.cluster import ClickHouseCluster
from helpers.network import PartitionManager
from helpers.test_tools import assert_eq_with_retry
import random
import string
import json

cluster = ClickHouseCluster(__file__)
node1 = cluster.add_instance("node1", with_zookeeper=True)
node2 = cluster.add_instance("node2", with_zookeeper=True)


@pytest.fixture(scope="module")
def started_cluster():
    try:
        cluster.start()

        yield cluster

    finally:
        cluster.shutdown()


def get_random_string(length):
    return "".join(
        random.choice(string.ascii_uppercase + string.digits) for _ in range(length)
    )


def test_system_replicated_fetches(started_cluster):
    node1.query(
        "CREATE TABLE t (key UInt64, data String) ENGINE = ReplicatedMergeTree('/clickhouse/test/t', '1') ORDER BY tuple()"
    )
    node2.query(
        "CREATE TABLE t (key UInt64, data String) ENGINE = ReplicatedMergeTree('/clickhouse/test/t', '2') ORDER BY tuple()"
    )

    with PartitionManager() as pm:
        node2.query("SYSTEM STOP FETCHES t")
        node1.query(
            "INSERT INTO t SELECT number, '{}' FROM numbers(10000)".format(
                get_random_string(104857)
            )
        )
        pm.add_network_delay(node1, 80)
        node2.query("SYSTEM START FETCHES t")
        fetches_result = []
        for _ in range(1000):
            result = json.loads(
                node2.query("SELECT * FROM system.replicated_fetches FORMAT JSON")
            )
            if not result["data"]:
                if fetches_result:
                    break
                time.sleep(0.1)
            else:
                fetches_result.append(result["data"][0])
                print(fetches_result[-1])
                time.sleep(0.1)

    node2.query("SYSTEM SYNC REPLICA t", timeout=10)
    assert node2.query("SELECT COUNT() FROM t") == "10000\n"

    for elem in fetches_result:
        elem["bytes_read_compressed"] = float(elem["bytes_read_compressed"])
        elem["total_size_bytes_compressed"] = float(elem["total_size_bytes_compressed"])
        elem["progress"] = float(elem["progress"])
        elem["elapsed"] = float(elem["elapsed"])

    assert len(fetches_result) > 0
    first_non_empty = fetches_result[0]

    assert first_non_empty["database"] == "default"
    assert first_non_empty["table"] == "t"
    assert first_non_empty["source_replica_hostname"] == "node1"
    assert first_non_empty["source_replica_port"] == 9009
    assert first_non_empty["source_replica_path"] == "/clickhouse/test/t/replicas/1"
    assert first_non_empty["interserver_scheme"] == "http"
    assert first_non_empty["result_part_name"] == "all_0_0_0"
    assert first_non_empty["result_part_path"].startswith("/var/lib/clickhouse/")
    assert first_non_empty["result_part_path"].endswith("all_0_0_0/")
    assert first_non_empty["partition_id"] == "all"
    assert first_non_empty["URI"].startswith(
        "http://node1:9009/?endpoint=DataPartsExchange"
    )

    for elem in fetches_result:
        assert (
            elem["bytes_read_compressed"] <= elem["total_size_bytes_compressed"]
        ), "Bytes read ({}) more than total bytes ({}). It's a bug".format(
            elem["bytes_read_compressed"], elem["total_size_bytes_compressed"]
        )
        assert (
            0.0 <= elem["progress"] <= 1.0
        ), "Progress shouldn't less than 0 and bigger than 1, got {}".format(
            elem["progress"]
        )
        assert (
            0.0 <= elem["elapsed"]
        ), "Elapsed time must be greater than 0, got {}".format(elem["elapsed"])

    prev_progress = first_non_empty["progress"]
    for elem in fetches_result:
        assert (
            elem["progress"] >= prev_progress
        ), "Progress decreasing prev{}, next {}? It's a bug".format(
            prev_progress, elem["progress"]
        )
        prev_progress = elem["progress"]

    prev_bytes = first_non_empty["bytes_read_compressed"]
    for elem in fetches_result:
        assert (
            elem["bytes_read_compressed"] >= prev_bytes
        ), "Bytes read decreasing prev {}, next {}? It's a bug".format(
            prev_bytes, elem["bytes_read_compressed"]
        )
        prev_bytes = elem["bytes_read_compressed"]

    prev_elapsed = first_non_empty["elapsed"]
    for elem in fetches_result:
        assert (
            elem["elapsed"] >= prev_elapsed
        ), "Elapsed time decreasing prev {}, next {}? It's a bug".format(
            prev_elapsed, elem["elapsed"]
        )
        prev_elapsed = elem["elapsed"]

    node1.query("DROP TABLE IF EXISTS t SYNC")
    node2.query("DROP TABLE IF EXISTS t SYNC")