import ipaddress
from unittest import mock

import pytest

from binderhub import utils


def test_rendezvous_rank():
    # check that a key doesn't move if its assigned bucket remains but the
    # other buckets are removed
    key = "crazy frog is a crazy key"
    first_round = utils.rendezvous_rank(["b1", "b2", "b3"], key)
    second_round = utils.rendezvous_rank([first_round[0], first_round[1]], key)

    assert first_round[0] == second_round[0], key


def test_rendezvous_independence():
    # check that the relative ranking of 80 buckets doesn't depend on the
    # presence of 20 extra buckets
    key = "k1"
    eighty_buckets = utils.rendezvous_rank([f"b{i}" for i in range(80)], key)
    hundred_buckets = utils.rendezvous_rank([f"b{i}" for i in range(100)], key)

    for i in range(80, 100):
        hundred_buckets.remove(f"b{i}")

    assert eighty_buckets == hundred_buckets


def test_rendezvous_redistribution():
    # check that approximately a third of keys move to the new bucket
    # when one is added
    n_keys = 3000

    # count how many keys were moved, which bucket a key started from and
    # which bucket a key was moved from (to the new bucket)
    n_moved = 0
    from_bucket = {"b1": 0, "b2": 0}
    start_in = {"b1": 0, "b2": 0}

    for i in range(n_keys):
        key = f"key-{i}"
        two_buckets = utils.rendezvous_rank(["b1", "b2"], key)
        start_in[two_buckets[0]] += 1
        three_buckets = utils.rendezvous_rank(["b1", "b2", "b3"], key)

        if two_buckets[0] != three_buckets[0]:
            n_moved += 1
            from_bucket[two_buckets[0]] += 1

            # should always move to the newly added bucket
            assert three_buckets[0] == "b3"

    # because of statistical fluctuations we have to leave some room when
    # making this comparison
    assert 0.31 < n_moved / n_keys < 0.35
    # keys should move from the two original buckets with approximately
    # equal probability. We pick 30 because it is "about right"
    assert abs(from_bucket["b1"] - from_bucket["b2"]) < 30
    # the initial distribution of keys should be roughly the same
    # We pick 30 because it is "about right"
    assert abs(start_in["b1"] - start_in["b2"]) < 30


def test_cache():
    cache = utils.Cache(max_size=2)
    cache.set("a", 1)
    cache.set("b", 2)
    assert "a" in cache
    cache.set("c", 3)
    assert "a" not in cache
    cache.set("b", 3)
    assert "b" in cache
    cache.set("d", 4)
    assert "b" in cache
    assert "c" not in cache
    assert len(cache) == 2


def test_cache_expiry():
    cache = utils.Cache(2, max_age=10)
    before_now = cache._now

    def later():
        return before_now() + 20

    expired = mock.patch.object(cache, "_now", later)

    cache.set("a", 1)
    assert "a" in cache
    assert cache.get("a")
    with expired:
        assert not cache.get("a")
        assert "a" not in cache
        assert "a" not in cache._ages

    cache.set("a", 1)
    # no max age means no expiry
    with expired, mock.patch.object(cache, "max_age", 0):
        assert cache.get("a")

    # retrieving an item does *not* update the age
    before_age = cache._ages["a"]
    with mock.patch.object(cache, "_now", lambda: before_now() + 1):
        cache.get("a")
    assert cache._ages["a"] == before_age


@pytest.mark.parametrize(
    "ip, cidrs, found",
    [
        ("192.168.1.1", ["192.168.1.1/32", "255.255.0.0/16"], True),
        ("192.168.1.2", ["192.168.1.1/32", "255.255.0.0/16"], False),
        ("192.168.1.2", ["192.168.1.0/24", "255.255.0.0/16"], True),
        ("192.168.1.2", ["255.255.0.0/16", "192.168.1.0/24"], True),
        ("192.168.1.2", [], False),
    ],
)
def test_ip_in_networks(ip, cidrs, found):
    networks = {ipaddress.ip_network(cidr): f"message {cidr}" for cidr in cidrs}
    if networks:
        min_prefix = min(net.prefixlen for net in networks)
    else:
        min_prefix = 1
    match = utils.ip_in_networks(ip, networks, min_prefix)
    if found:
        assert match
        net, message = match
        assert message == f"message {net}"
        assert ipaddress.ip_address(ip) in net
    else:
        assert match is False


def test_ip_in_networks_invalid():
    with pytest.raises(ValueError):
        utils.ip_in_networks("1.2.3.4", {}, 0)
