"""Profile edit: the page, the phone format, the diff against AD, audit and errors (FR-9)."""

import logging
import re

import pytest
from django.test import Client

from portal.adapters.directory import DirectoryUnavailable, Forbidden, ValidationFailed
from portal.audit import services as audit_services
from portal.audit.models import AuditEntry
from portal.profile.forms import InvalidNumber, normalize_phone
from portal.tests.helpers import sign_in

pytestmark = pytest.mark.django_db(transaction=True)

VALID = {
    "first_name": "Alice",
    "last_name": "Anders",
    "display_name": "Alice Anders",
    "telephone": "+49301234567",
    "mobile": "intern 23",
}


@pytest.fixture
def people(adapters):
    d = adapters.directory
    d.add_user(
        "alice",
        given_name="Alice",
        surname="Anders",
        display_name="Alice Anders",
        mail="alice@example.test",
        telephone="+49301234567",
        mobile="intern 23",  # an old value that is not a number
    )
    d.add_user("bob", given_name="Bob", surname="Brown", display_name="Bob Brown")
    return d


def signed_in(adapters, name="alice"):
    client, _ = sign_in(adapters, name)
    return client


def save(client, **changes):
    return client.post("/profile", {**VALID, **changes})


def profile_entries():
    return AuditEntry.objects.filter(action__startswith="profile.")


def audit_actions():
    return list(profile_entries().values_list("action", flat=True))


def stored(adapters, sam="alice"):
    return adapters.directory.find_user(sam)


class Spy:
    def __init__(self, directory):
        self.writes = []
        original = directory.update_attributes

        def wrapper(sam, changes):
            self.writes.append((sam, dict(changes)))
            return original(sam, changes)

        directory.update_attributes = wrapper


# -- the page ------------------------------------------------------------------------------


def test_the_page_shows_the_ad_values_and_plain_text_for_username_and_mail(people, adapters):
    html = signed_in(adapters).get("/profile").content.decode()
    for value in ("Alice", "Anders", "Alice Anders", "+49301234567", "intern 23"):
        assert f'value="{value}"' in html
    assert "alice@example.test" in html
    assert 'name="username"' not in html and 'name="mail"' not in html
    assert re.search(r'<a href="/profile" class="current" aria-current="page">', html)


def test_the_page_speaks_german_in_both_registers(people, adapters):
    client = signed_in(adapters)
    client.cookies["pm_lang"] = "de"
    assert "Diese Angaben sind in Ihrem Konto" in client.get("/profile").content.decode()
    client.cookies["pm_form"] = "du"
    assert "in deinem Konto" in client.get("/profile").content.decode()


def test_the_page_is_unavailable_when_the_directory_is_down(people, adapters):
    client = signed_in(adapters)
    adapters.directory.unavailable = True
    response = client.get("/profile")
    assert response.status_code == 503 and b'name="first_name"' not in response.content


def test_access_rules(people, adapters):
    assert Client().get("/profile").status_code == 302
    assert Client().post("/profile", VALID).status_code == 302
    assert signed_in(adapters).put("/profile").status_code == 405
    client = signed_in(adapters)
    strict = Client(enforce_csrf_checks=True)
    strict.cookies = client.cookies
    assert strict.post("/profile", {**VALID, "first_name": "Eve"}).status_code == 403
    assert stored(adapters).given_name == "Alice"


# -- saving ----------------------------------------------------------------------------------


def test_only_the_changed_fields_are_written_and_audited_without_values(people, adapters):
    spy = Spy(adapters.directory)
    client = signed_in(adapters)
    response = save(client, first_name="Alicia", mobile="0171 / 123456")
    assert response.status_code == 302 and response["Location"] == "/profile"
    assert spy.writes == [("alice", {"givenName": "Alicia", "mobile": "+49171123456"})]
    account = stored(adapters)
    assert (account.given_name, account.mobile) == ("Alicia", "+49171123456")
    assert (account.surname, account.telephone, account.mail) == (
        "Anders",
        "+49301234567",
        "alice@example.test",
    )
    entry = AuditEntry.objects.get(action="profile.updated")
    assert entry.actor == "alice" and entry.ip == "127.0.0.1"
    assert entry.detail == {"fields": ["first_name", "mobile"]}
    assert profile_entries().count() == 1


def test_the_saved_notice_shows_once_with_the_stored_number_form(people, adapters):
    client = signed_in(adapters)
    response = save(client, telephone="0049 (0) 30 1234568")
    page = client.get(response["Location"]).content.decode()
    assert "Your changes were saved." in page and 'value="+49301234568"' in page
    assert "Your changes were saved." not in client.get("/profile").content.decode()


def test_nothing_changed_makes_no_write_and_no_audit(people, adapters):
    spy = Spy(adapters.directory)
    client = signed_in(adapters)
    response = save(client)
    assert response.status_code == 302 and spy.writes == [] and profile_entries().count() == 0
    assert "Nothing to change." in client.get("/profile").content.decode()


def test_a_number_equal_once_normalized_is_not_a_change(people, adapters):
    spy = Spy(adapters.directory)
    save(signed_in(adapters), telephone="+49 30 1234567")
    assert spy.writes == [] and profile_entries().count() == 0


def test_an_empty_number_clears_the_attribute(people, adapters):
    spy = Spy(adapters.directory)
    save(signed_in(adapters), telephone="")
    assert spy.writes == [("alice", {"telephoneNumber": ""})]
    assert stored(adapters).telephone == ""


def test_an_old_value_that_is_not_a_number_never_blocks_another_field(people, adapters):
    save(signed_in(adapters), last_name="Ander")
    account = stored(adapters)
    assert account.surname == "Ander" and account.mobile == "intern 23"


def test_old_values_that_break_the_rules_are_left_alone_but_changed_ones_are_checked(
    people, adapters
):
    long_name = "x" * 100
    adapters.directory.add_user("carol", given_name=long_name, surname="", display_name="Carol")
    client = signed_in(adapters, "carol")
    spy = Spy(adapters.directory)
    old = {
        "first_name": long_name,
        "last_name": "",
        "display_name": "Carol",
        "telephone": "",
        "mobile": "",
    }
    assert client.post("/profile", {**old, "mobile": "5"}).status_code == 400
    assert client.post("/profile", {**old, "display_name": "Caro"}).status_code == 302
    assert spy.writes == [("carol", {"displayName": "Caro"})]


def test_the_session_name_follows_the_saved_display_name(people, adapters):
    client = signed_in(adapters)
    assert "Alice Anders" in client.get("/").content.decode()
    save(client, display_name="Ally")
    assert "Ally" in client.get("/profile").content.decode()
    assert "Alice Anders" not in client.get("/").content.decode()


def test_the_save_compares_with_a_fresh_read_not_with_the_page(people, adapters):
    spy = Spy(adapters.directory)
    client = signed_in(adapters)
    client.get("/profile")
    adapters.directory.update_attributes("alice", {"sn": "Elsewhere"})
    save(client, display_name="Ally", last_name="Elsewhere")  # same as AD now: not written
    assert spy.writes[1:] == [("alice", {"displayName": "Ally"})]


# -- phone numbers ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    ("text", "result"),
    [
        ("0171 / 123456", "+49171123456"),
        ("0171123456", "+49171123456"),
        ("+49 171 123456", "+49171123456"),
        ("+49 (0) 171-123456", "+49171123456"),
        ("0049 171 123456", "+49171123456"),
        ("0049 (0) 171 123456", "+49171123456"),
        ("+43 660 1234567", "+436601234567"),
        ("0043 660 1234567", "+436601234567"),
        ("(0171) 123.456", "+49171123456"),
        ("+49171123456", "+49171123456"),
    ],
)
def test_phone_inputs_and_their_stored_form(text, result):
    assert normalize_phone(text) == result
    assert normalize_phone(result) == result


@pytest.mark.parametrize(
    "text",
    [
        "171123456",
        "abc",
        "0171 abc",
        "+49 171 12+3456",
        "+49 1711 23456 7890 1234",
        "0123",
        "+49 12",
        "++49171123456",
        "intern 23",
        "",
        "*)(uid=*",
    ],
)
def test_phone_inputs_that_are_refused(text):
    with pytest.raises(InvalidNumber):
        normalize_phone(text)


@pytest.mark.parametrize("field", ["telephone", "mobile"])
@pytest.mark.parametrize("text", ["171123456", "0171 abc", "+49 12", "+49 171 12+3456"])
def test_a_bad_number_is_refused_under_its_field(people, adapters, field, text):
    spy = Spy(adapters.directory)
    response = save(signed_in(adapters), **{field: text})
    html = response.content.decode()
    assert response.status_code == 400 and spy.writes == []
    assert "must be a number that starts with + and the country code, or with 0" in html
    assert 'class="error"' in html and profile_entries().count() == 0


# -- invalid values ---------------------------------------------------------------------------


@pytest.mark.parametrize("field", ["first_name", "last_name", "display_name"])
def test_a_required_name_cannot_be_empty(people, adapters, field):
    spy = Spy(adapters.directory)
    response = save(signed_in(adapters), **{field: "   "})
    assert response.status_code == 400 and spy.writes == [] and profile_entries().count() == 0
    assert "is required." in response.content.decode()


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("first_name", "x" * 65),
        ("last_name", "x" * 65),
        ("display_name", "x" * 257),
        ("first_name", "A\nB"),
        ("last_name", "A\tB"),
        ("display_name", "A\x00B"),
    ],
)
def test_too_long_or_control_characters_are_refused_and_other_input_stays(
    people, adapters, field, value
):
    spy = Spy(adapters.directory)
    response = save(signed_in(adapters), mobile="0171 5", **{field: value})
    html = response.content.decode()
    assert response.status_code == 400 and spy.writes == [] and profile_entries().count() == 0
    assert "Your changes were not saved." in html
    assert 'value="0171 5"' in html  # what else was typed stays in the form
    assert f'<a href="#id_{field}">' in html


def test_the_limits_are_ads_limits(people, adapters):
    save(signed_in(adapters), first_name="x" * 64, display_name="y" * 256)
    account = stored(adapters)
    assert len(account.given_name) == 64 and len(account.display_name) == 256


def test_values_are_trimmed(people, adapters):
    save(signed_in(adapters), first_name="  Alicia  ")
    assert stored(adapters).given_name == "Alicia"


@pytest.mark.parametrize("text", ["*)(uid=*", "\\00", "CN=x,DC=y", "a)(|(cn=*"])
def test_injection_strings_are_plain_text_and_touch_nothing_else(people, adapters, text):
    spy = Spy(adapters.directory)
    before_bob = stored(adapters, "bob")
    assert save(signed_in(adapters), display_name=text).status_code == 302
    assert spy.writes == [("alice", {"displayName": text})]
    assert stored(adapters).display_name == text
    assert stored(adapters, "bob") == before_bob


# -- who is edited -----------------------------------------------------------------------------


def test_a_posted_username_is_ignored_and_another_user_is_untouched(people, adapters):
    before_bob = stored(adapters, "bob")
    response = save(signed_in(adapters), first_name="Alicia", username="bob", sam="bob", dn="x")
    assert response.status_code == 302
    assert stored(adapters).given_name == "Alicia" and stored(adapters, "bob") == before_bob
    assert AuditEntry.objects.get(action="profile.updated").actor == "alice"


# -- failures ------------------------------------------------------------------------------------


def test_ad_refusing_a_value_changes_nothing_and_is_audited_as_failure(people, adapters):
    def refuse(sam, changes):
        raise ValidationFailed("constraint")

    adapters.directory.update_attributes = refuse
    response = save(signed_in(adapters), first_name="Alicia")
    html = response.content.decode()
    assert response.status_code == 400
    assert "A value was not accepted. Please check your entries." in html
    assert "constraint" not in html and 'value="Alicia"' in html
    assert stored(adapters).given_name == "Alice"
    assert audit_actions() == ["profile.update_failed"]
    assert profile_entries().get().detail == {"reason": "refused"}


def test_directory_down_on_save_is_503_and_audited(people, adapters):
    client = signed_in(adapters)
    adapters.directory.unavailable = True
    response = save(client, first_name="Alicia")
    assert response.status_code == 503
    assert "This is not possible right now. Please try again later." in response.content.decode()
    adapters.directory.unavailable = False
    assert stored(adapters).given_name == "Alice"
    assert audit_actions() == ["profile.update_failed"]
    assert profile_entries().get().detail == {"reason": "unavailable"}


def test_directory_failing_during_the_write_rolls_back_and_audits(people, adapters):
    def down(sam, changes):
        raise DirectoryUnavailable("gone")

    adapters.directory.update_attributes = down
    assert save(signed_in(adapters), first_name="Alicia").status_code == 503
    assert audit_actions() == ["profile.update_failed"]


def test_a_failure_after_the_ad_call_is_logged_and_the_user_told_it_is_unclear(
    people, adapters, monkeypatch, caplog
):
    real_record = audit_services.record

    def broken(*args, **kwargs):
        raise RuntimeError("database gone: Alicia")

    client = signed_in(adapters)
    monkeypatch.setattr(audit_services, "record", broken)
    with caplog.at_level(logging.ERROR):
        response = save(client, first_name="Alicia", mobile="0171 5555555")
    monkeypatch.setattr(audit_services, "record", real_record)
    assert response.status_code == 500 and "may have been changed" in response.content.decode()
    assert stored(adapters).given_name == "Alicia"  # AD did change
    assert audit_actions() == ["profile.update_failed"]  # no "updated" entry survived
    log = next(r.getMessage() for r in caplog.records if r.levelno == logging.ERROR)
    assert "user=alice" in log and "fields=first_name,mobile" in log and "step=audit" in log
    assert "Alicia" not in log and "5555555" not in log


def test_a_missing_write_right_is_the_not_possible_page_and_a_failure_entry(
    people, adapters, caplog
):
    def forbidden(sam, changes):
        raise Forbidden("insufficient access: Alicia")

    client = signed_in(adapters)
    adapters.directory.update_attributes = forbidden
    with caplog.at_level(logging.ERROR):
        response = save(client, first_name="Alicia")
    assert response.status_code == 503
    assert "This is not possible right now." in response.content.decode()
    assert audit_actions() == ["profile.update_failed"]
    assert profile_entries().get().detail == {"reason": "forbidden"}
    assert not any("Alicia" in r.getMessage() for r in caplog.records)


def test_an_invalid_save_reads_the_directory_once(people, adapters):
    client = signed_in(adapters)
    reads = []
    original = adapters.directory.find_user
    adapters.directory.find_user = lambda sam: reads.append(sam) or original(sam)
    response = save(client, first_name="  ")
    assert response.status_code == 400 and len(reads) == 1
    assert "alice@example.test" in response.content.decode()