# -*- coding: utf-8 -*-
"""
tests.functional.test_web
~~~~~~~~~~~~~~~~~~~~~~~~~

This module defines basic web tests.

:copyright: (c) 2017 European Spallation Source ERIC
:license: BSD 2-Clause, see LICENSE for more details.

"""
import pytest
import re
from app import models


def login(client, username, password):
    data = {"username": username, "password": password}
    return client.post("/user/login", data=data, follow_redirects=True)


def logout(client):
    return client.get("/user/logout", follow_redirects=True)


@pytest.fixture
def logged_client(client):
    login(client, "user_ro", "userro")
    yield client
    logout(client)


@pytest.fixture
def logged_rw_client(client):
    login(client, "user_rw", "userrw")
    yield client
    logout(client)


@pytest.fixture
def no_login_check_client(request, app):
    app.config["LOGIN_DISABLED"] = True
    client = app.test_client()
    # We still need to login, otherwise an AnonymousUserMixin is returned
    # An AnonymousUser doesn't have all the User methods
    login(client, "user_ro", "userro")
    yield client
    app.config["LOGIN_DISABLED"] = False
    logout(client)


def test_login_logout(client):
    response = login(client, "unknown", "invalid")
    assert b"<title>Login - CSEntry</title>" in response.data
    response = login(client, "user_rw", "invalid")
    assert b"<title>Login - CSEntry</title>" in response.data
    response = login(client, "user_rw", "userrw")
    assert b"Control System Entry" in response.data
    assert b"User RW" in response.data
    response = logout(client)
    assert b"<title>Login - CSEntry</title>" in response.data


def test_index(logged_client):
    response = logged_client.get("/")
    assert b"Control System Entry" in response.data
    assert b"User RO" in response.data


@pytest.mark.parametrize("url", ["/", "/inventory/items", "/network/networks"])
def test_protected_url_get(url, client):
    response = client.get(url)
    assert response.status_code == 302
    assert "/user/login" in response.headers["Location"]
    login(client, "user_ro", "userro")
    response = client.get(url)
    assert response.status_code == 200


@pytest.mark.parametrize(
    "url", ["/inventory/_retrieve_items", "/network/_retrieve_hosts"]
)
def test_protected_url_post(url, client):
    response = client.post(url)
    assert response.status_code == 302
    assert "/user/login" in response.headers["Location"]
    login(client, "user_ro", "userro")
    response = client.post(url)
    assert response.status_code == 200


def test_retrieve_items(logged_client, item_factory):
    response = logged_client.post("/inventory/_retrieve_items")
    assert response.get_json()["data"] == []
    serial_numbers = ("12345", "45678")
    for sn in serial_numbers:
        item_factory(serial_number=sn)
    response = logged_client.post("/inventory/_retrieve_items")
    items = response.get_json()["data"]
    assert set(serial_numbers) == set(item["serial_number"] for item in items)
    assert len(items[0]) == 18


def test_retrieve_items_pagination(logged_client, item_factory):
    for sn in range(1000, 1030):
        item_factory(serial_number=sn)
    response = logged_client.post(
        "/inventory/_retrieve_items", data={"draw": "50", "length": 10, "start": 0}
    )
    r = response.get_json()
    assert r["draw"] == 50
    assert r["recordsTotal"] == 30
    assert r["recordsFiltered"] == 30
    assert len(r["data"]) == 10
    serial_numbers = [item["serial_number"] for item in r["data"]]
    response = logged_client.post(
        "/inventory/_retrieve_items", data={"draw": "51", "length": 10, "start": 10}
    )
    serial_numbers.extend(
        [item["serial_number"] for item in response.get_json()["data"]]
    )
    response = logged_client.post(
        "/inventory/_retrieve_items", data={"draw": "52", "length": 10, "start": 20}
    )
    serial_numbers.extend(
        [item["serial_number"] for item in response.get_json()["data"]]
    )
    assert sorted(serial_numbers) == list(str(i) for i in range(1000, 1030))


def test_retrieve_items_filter(logged_client, item_factory):
    for sn in range(1000, 1010):
        item_factory(serial_number=sn)
    response = logged_client.post(
        "/inventory/_retrieve_items",
        data={
            "draw": "50",
            "length": 20,
            "start": 0,
            "search[value]": "serial_number:1005",
        },
    )
    r = response.get_json()
    assert r["recordsTotal"] == 10
    assert r["recordsFiltered"] == 1
    assert len(r["data"]) == 1
    assert r["data"][0]["serial_number"] == "1005"


def test_retrieve_items_sort(logged_client, item_factory):
    serial_numbers = ["AAA001", "AAB034", "AAA100", "AAB001"]
    for sn in serial_numbers:
        item_factory(serial_number=sn)
    response = logged_client.post(
        "/inventory/_retrieve_items",
        data={
            "draw": "50",
            "length": 20,
            "start": 0,
            "order[0][column]": "3",
            "columns[3][data]": "serial_number",
        },
    )
    items = response.get_json()["data"]
    assert sorted(serial_numbers) == [item["serial_number"] for item in items]
    response = logged_client.post(
        "/inventory/_retrieve_items",
        data={
            "draw": "50",
            "length": 20,
            "start": 0,
            "order[0][column]": "3",
            "order[0][dir]": "desc",
            "columns[3][data]": "serial_number",
        },
    )
    items = response.get_json()["data"]
    assert sorted(serial_numbers, reverse=True) == [
        item["serial_number"] for item in items
    ]


def test_retrieve_items_case_insensitive(logged_client, model_factory, item_factory):
    juniper_model = model_factory(name="Juniper")
    item_factory(serial_number="BBB001", model=juniper_model)
    item_factory(serial_number="ABB042")
    response = logged_client.post(
        "/inventory/_retrieve_items",
        data={"draw": "50", "length": 20, "start": 0, "search[value]": "juniper"},
    )
    r = response.get_json()
    assert r["recordsTotal"] == 2
    assert r["recordsFiltered"] == 1
    assert len(r["data"]) == 1
    assert r["data"][0]["model"] == "Juniper"


def test_retrieve_items_one_word(logged_client, manufacturer_factory, item_factory):
    manufacturer = manufacturer_factory(name="Concurrent Technologies")
    item_factory(serial_number="AAA001", manufacturer=manufacturer)
    item_factory(serial_number="ABB042")
    response = logged_client.post(
        "/inventory/_retrieve_items",
        data={"draw": "50", "length": 20, "start": 0, "search[value]": "concurrent"},
    )
    r = response.get_json()
    assert r["recordsTotal"] == 2
    assert r["recordsFiltered"] == 1
    assert len(r["data"]) == 1
    assert r["data"][0]["manufacturer"] == "Concurrent Technologies"


def test_generate_random_mac(logged_client):
    response = logged_client.get("/network/_generate_random_mac")
    mac = response.get_json()["data"]["mac"]
    assert re.match("^(?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}$", mac) is not None
    assert mac.startswith("02:42:42")


def test_retrieve_hosts(logged_client, interface_factory, host_factory):
    response = logged_client.post("/network/_retrieve_hosts")
    assert response.get_json()["data"] == []
    host1 = host_factory(name="host1")
    host2 = host_factory(name="host2")
    interface_factory(name="host1", host=host1)
    interface_factory(name="host2", host=host2)
    response = logged_client.post("/network/_retrieve_hosts")
    hosts = response.get_json()["data"]
    assert {host1.name, host2.name} == set(host["name"] for host in hosts)
    assert len(hosts[0]) == 14
    assert len(hosts[0]["interfaces"][0]) == 15


def test_retrieve_hosts_by_ip(logged_client, interface_factory):
    interface1 = interface_factory()
    interface_factory()
    response = logged_client.post(
        "/network/_retrieve_hosts",
        data={"draw": "50", "length": 20, "start": 0, "search[value]": interface1.ip},
    )
    r = response.get_json()
    assert r["recordsTotal"] == 2
    assert r["recordsFiltered"] == 1
    assert len(r["data"]) == 1
    assert r["data"][0]["name"] == interface1.host.name


def test_delete_interface_from_index(
    no_login_check_client, interface_factory, host_factory
):
    host1 = host_factory(name="host1")
    interface_factory(name="host1", host=host1)
    interface2 = interface_factory(name="host1b", host=host1)
    # The interface is in the index
    instances, nb = models.Host.search("host1b")
    assert list(instances) == [host1]
    assert nb == 1
    # Delete the interface
    response = no_login_check_client.post(
        "/network/interfaces/delete", data={"interface_id": interface2.id}
    )
    assert response.status_code == 302
    # It's not in the database anymore
    assert models.Interface.query.get(interface2.id) is None
    # Neither in the index
    instances, nb = models.Host.search("host1b")
    assert list(instances) == []
    assert nb == 0
    # But host1 is still in the index
    instances, nb = models.Host.search("host1")
    assert list(instances) == [host1]
    assert nb == 1


def test_edit_item_comment_in_index(
    logged_rw_client, item_factory, item_comment_factory
):
    item1 = item_factory(ics_id="AAA001")
    comment = item_comment_factory(body="Hello", item_id=item1.id)
    assert item1.comments == [comment]
    # Edit the comment
    body = "Hello world!"
    response = logged_rw_client.post(
        f"/inventory/items/comment/edit/{comment.id}", data={"body": body}
    )
    assert response.status_code == 302
    # The comment was updated in the database
    updated_comment = models.ItemComment.query.get(comment.id)
    assert updated_comment.body == body
    # And in the index
    instances, nb = models.Item.search("world")
    assert list(instances) == [item1]


def test_create_host(client, domain_factory, network_factory, device_type):
    domain = domain_factory(name="prod.example.org")
    network = network_factory(
        address="192.168.1.0/24",
        first_ip="192.168.1.10",
        last_ip="192.168.1.250",
        domain=domain,
    )
    name = "myhost"
    ip = "192.168.1.11"
    mac = "02:42:42:45:3c:89"
    form = {
        "network_id": network.id,
        "name": name,
        "device_type_id": device_type.id,
        "is_ioc": False,
        "ip": ip,
        "mac": mac,
        "description": "test",
        "ansible_vars": "",
        "ansible_groups": [],
        "random_mac": False,
        "cnames_string": "",
    }
    # Invalid network_id with user_lab user
    # (form validation error because the network is not part of the choices
    # for this user)
    login(client, "user_lab", "userlab")
    response = client.post(f"/network/hosts/create", data=form)
    assert response.status_code == 200
    # The host wasn't created
    assert models.Host.query.filter_by(name=name).first() is None
    logout(client)
    # Success with user_prod user
    login(client, "user_prod", "userprod")
    response = client.post(f"/network/hosts/create", data=form, follow_redirects=True)
    assert response.status_code == 200
    # The host was created
    assert b"created!" in response.data
    assert b"View host" in response.data
    host = models.Host.query.filter_by(name=name).first()
    assert host is not None
    assert host.interfaces[0].ip == ip
    assert host.interfaces[0].mac == mac
    assert host.interfaces[0].name == name


def test_create_host_invalid_fields(
    session, client, domain_factory, network_factory, device_type
):
    domain = domain_factory(name="prod.example.org")
    network = network_factory(
        address="192.168.1.0/24",
        first_ip="192.168.1.10",
        last_ip="192.168.1.250",
        domain=domain,
    )
    name = "myhost"
    ip = "192.168.1.11"
    mac = "02:42:42:45:3c:89"
    form = {
        "network_id": network.id,
        "name": name,
        "device_type_id": device_type.id,
        "is_ioc": False,
        "ip": ip,
        "mac": mac,
        "description": "test",
        "ansible_vars": "",
        "ansible_groups": [],
        "random_mac": False,
        "cnames_string": "",
    }
    login(client, "user_prod", "userprod")
    # Invalid mac
    data = form.copy()
    data["mac"] = "ea:ea:60:45:a8:96:se"
    response = client.post(f"/network/hosts/create", data=data, follow_redirects=True)
    assert response.status_code == 200
    assert b"Register new host" in response.data
    assert b"Invalid MAC address" in response.data
    # An exception was raised during validation (on Select in the Unique Validator),
    # so we need to rollback.
    session.rollback()
    # Invalid hostname
    data = form.copy()
    data["name"] = "invalid_host"
    response = client.post(f"/network/hosts/create", data=data, follow_redirects=True)
    assert response.status_code == 200
    assert b"Register new host" in response.data
    assert b"Invalid input" in response.data


def test_create_interface(
    client, host_factory, domain_factory, network_factory, interface_factory
):
    host = host_factory(name="myhost")
    domain = domain_factory(name="prod.example.org")
    network1 = network_factory(domain=domain)
    interface_factory(network=network1, host=host)
    network2 = network_factory(
        address="192.168.2.0/24",
        first_ip="192.168.2.10",
        last_ip="192.168.2.250",
        domain=domain,
    )
    name = host.name + "-2"
    ip = "192.168.2.11"
    mac = "02:42:42:46:3c:75"
    form = {
        "host_id": host.id,
        "interface_name": name,
        "network_id": network2.id,
        "random_mac": False,
        "ip": ip,
        "mac": mac,
        "cnames_string": "",
    }
    # Permission denied
    # user_lab doesn't have permissions for the host domain: prod.example.org
    login(client, "user_lab", "userlab")
    response = client.post(f"/network/interfaces/create/{host.name}", data=form)
    assert response.status_code == 403
    # The host wasn't created
    assert models.Interface.query.filter_by(name=name).first() is None
    logout(client)
    # Success with user_prod user
    login(client, "user_prod", "userprod")
    response = client.post(f"/network/interfaces/create/{host.name}", data=form)
    assert response.status_code == 302
    # The interface was created
    interface = models.Interface.query.filter_by(name=name).first()
    assert interface is not None
    assert interface.ip == ip
    assert interface.mac == mac
    assert interface.name == name
    assert interface.host == host


def check_vm_creation_response(response, success=True):
    assert response.status_code == 200
    assert (b"View task" in response.data) is success
    assert (b"View host" in response.data) is not success
    assert (b"Please contact an admin user" in response.data) is not success


def test_create_vm(
    client,
    domain_factory,
    network_factory,
    device_type_factory,
    host_factory,
    interface_factory,
):
    virtualmachine = device_type_factory(name="VirtualMachine")
    domain_prod = domain_factory(name="prod.example.org")
    domain_lab = domain_factory(name="lab.example.org")
    network_prod = network_factory(domain=domain_prod)
    network_lab = network_factory(domain=domain_lab)
    vm_prod = host_factory(device_type=virtualmachine)
    interface_factory(name=vm_prod.name, host=vm_prod, network=network_prod)
    vioc_prod = host_factory(device_type=virtualmachine, is_ioc=True)
    interface_factory(name=vioc_prod.name, host=vioc_prod, network=network_prod)
    vm_lab = host_factory(device_type=virtualmachine)
    interface_factory(name=vm_lab.name, host=vm_lab, network=network_lab)
    vioc_lab = host_factory(device_type=virtualmachine, is_ioc=True)
    interface_factory(name=vioc_lab.name, host=vioc_lab, network=network_lab)
    form = {"cores": 1, "memory": 4, "disk": 15, "osversion": "centos7"}
    # User has access to the lab networks and can create VM and VIOC there
    login(client, "user_lab", "userlab")
    response = client.post(
        f"/network/hosts/view/{vm_prod.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=False)
    response = client.post(
        f"/network/hosts/view/{vioc_prod.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=False)
    response = client.post(
        f"/network/hosts/view/{vm_lab.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=True)
    response = client.post(
        f"/network/hosts/view/{vioc_lab.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=True)
    logout(client)
    # User has access to the prod networks but can only create VIOC due to ALLOWED_VM_CREATION_DOMAINS
    login(client, "user_prod", "userprod")
    response = client.post(
        f"/network/hosts/view/{vm_prod.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=False)
    response = client.post(
        f"/network/hosts/view/{vioc_prod.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=True)
    response = client.post(
        f"/network/hosts/view/{vm_lab.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=False)
    response = client.post(
        f"/network/hosts/view/{vioc_lab.name}", data=form, follow_redirects=True
    )
    check_vm_creation_response(response, success=False)