<commit_before># Copyright (C) 2022  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information

import datetime
from io import StringIO
from typing import Dict

from psycopg2.extras import execute_values
import pytest

from django.core.management import call_command
from django.db import transaction

from swh.model.model import Person
from swh.web.auth.models import UserMailmap
from swh.web.auth.utils import MAILMAP_PERMISSION
from swh.web.common.utils import reverse
from swh.web.tests.utils import (
    check_api_post_response,
    check_http_get_response,
    create_django_permission,
)


@pytest.fixture
def mailmap_user(regular_user):
    regular_user.user_permissions.add(create_django_permission(MAILMAP_PERMISSION))
    return regular_user


@pytest.mark.django_db(transaction=True)
@pytest.mark.parametrize("view_name", ["profile-mailmap-add", "profile-mailmap-update"])
def test_mailmap_endpoints_anonymous_user(api_client, view_name):
    url = reverse(view_name)
    check_api_post_response(api_client, url, status_code=403)


@pytest.mark.django_db(transaction=True)
def test_mailmap_endpoints_user_with_permission(api_client, mailmap_user):
    api_client.force_login(mailmap_user)
    for view_name in ("profile-mailmap-add", "profile-mailmap-update"):
        url = reverse(view_name)
        check_api_post_response(
            api_client,
            url,
            data={"from_email": "<EMAIL>", "display_name": "bar"},
            status_code=200,
        )

    # FIXME: use check_api_get_responses; currently this crashes without
    # content_type="application/json"
    resp = check_http_get_response(
        api_client,
        reverse("profile-mailmap-list"),
        status_code=200,
        content_type="application/json",
    ).data
    assert len(resp) == 1
    assert resp[0]["from_email"] == "<EMAIL>"
    assert resp[0]["display_name"] == "bar"


@pytest.mark.django_db(transaction=True)
def test_mailmap_add_duplicate(api_client, mailmap_user):
    api_client.force_login(mailmap_user)

    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "bar"},
        status_code=200,
    )
    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "baz"},
        status_code=400,
    )


@pytest.mark.django_db(transaction=True)
def test_mailmap_endpoints_error_response(api_client, mailmap_user):
    api_client.force_login(mailmap_user)

    url = reverse("profile-mailmap-add")
    resp = check_api_post_response(api_client, url, status_code=400)
    assert b"from_email" in resp.content

    url = reverse("profile-mailmap-update")
    resp = check_api_post_response(api_client, url, status_code=400)
    assert b"from_email" in resp.content


@pytest.mark.django_db(transaction=True)
def test_mailmap_update(api_client, mailmap_user):
    api_client.force_login(mailmap_user)

    before_add = datetime.datetime.now(tz=datetime.timezone.utc)

    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "Display Name 1"},
        status_code=200,
    )
    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "Display Name 2"},
        status_code=200,
    )
    after_add = datetime.datetime.now(tz=datetime.timezone.utc)

    mailmaps = list(UserMailmap.objects.order_by("from_email").all())
    assert len(mailmaps) == 2, mailmaps

    assert mailmaps[0].from_email == "<EMAIL>", mailmaps
    assert mailmaps[0].display_name == "Display Name 1", mailmaps
    assert before_add <= mailmaps[0].last_update_date <= after_add

    assert mailmaps[1].from_email == "<EMAIL>", mailmaps
    assert mailmaps[1].display_name == "Display Name 2", mailmaps
    assert before_add <= mailmaps[0].last_update_date <= after_add

    before_update = datetime.datetime.now(tz=datetime.timezone.utc)

    check_api_post_response(
        api_client,
        reverse("profile-mailmap-update"),
        data={"from_email": "<EMAIL>", "display_name": "Display Name 1b"},
        status_code=200,
    )

    after_update = datetime.datetime.now(tz=datetime.timezone.utc)

    mailmaps = list(UserMailmap.objects.order_by("from_email").all())
    assert len(mailmaps) == 2, mailmaps

    assert mailmaps[0].from_email == "<EMAIL>", mailmaps
    assert mailmaps[0].display_name == "Display Name 1b", mailmaps
    assert before_update <= mailmaps[0].last_update_date <= after_update

    assert mailmaps[1].from_email == "<EMAIL>", mailmaps
    assert mailmaps[1].display_name == "Display Name 2", mailmaps
    assert before_add <= mailmaps[1].last_update_date <= after_add


def populate_mailmap():
    for (verified, activated) in (
        (False, False),
        (False, True),
        (True, False),
        (True, True),
    ):
        verified_str = "V" if verified else ""
        activated_str = "A" if activated else ""
        UserMailmap.objects.create(
            from_email=f"from_email{verified_str}{<EMAIL>",
            display_name=f"Display Name {verified_str} {activated_str}".strip(),
            from_email_verified=verified,
            display_name_activated=activated,
        )


def call_sync_mailmaps(*args) -> str:
    out = StringIO()
    err = StringIO()

    call_command("sync_mailmaps", *args, stdout=out, stderr=err)

    out.seek(0)
    err.seek(0)
    assert err.read() == ""
    return out.read()


MAILMAP_KNOWN_FULLNAMES = (
    "Original Name <<EMAIL>>",
    "Original Name V <<EMAIL>>",
    "Original Name A <<EMAIL>>",
    "Original Name V A <<EMAIL>>",
    "Original Name V A 2 <<EMAIL>>",
    "Original Name V A 3 <<EMAIL>>",
)


MAILMAP_KNOWN_PEOPLE = tuple(
    Person.from_fullname(f.encode()) for f in MAILMAP_KNOWN_FULLNAMES
)


def init_stub_storage_db(postgresql):
    cur = postgresql.cursor()
    cur.execute(
        """
        CREATE TABLE person (
          fullname bytea PRIMARY KEY,
          name bytea,
          email bytea,
          displayname bytea
        )
        """
    )
    execute_values(
        cur,
        "INSERT INTO person (fullname, name, email) VALUES %s",
        (p.to_dict() for p in MAILMAP_KNOWN_PEOPLE),
        template="(%(fullname)s, %(name)s, %(email)s)",
    )
    cur.execute("CREATE INDEX ON person (email)")
    postgresql.commit()
    cur.close()

    return postgresql.dsn


def get_displaynames(postgresql) -> Dict[str, str]:
    with postgresql.cursor() as cur:
        cur.execute(
            "SELECT fullname, displayname FROM person WHERE displayname IS NOT NULL"
        )

        return {bytes(f).decode("utf-8"): bytes(d).decode("utf-8") for (f, d) in cur}


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_dry_run(postgresql):
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    out = call_sync_mailmaps(dsn)

    assert "(dry run)" in out
    assert "Synced 1 mailmaps to swh.storage database" in out

    assert get_displaynames(postgresql) == {}

    assert (
        UserMailmap.objects.filter(
            from_email_verified=True,
            display_name_activated=True,
            mailmap_last_processing_date__isnull=False,
        ).count()
        == 0
    )


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_perform(postgresql):
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    out = call_sync_mailmaps("--perform", dsn)

    assert "(dry run)" not in out
    assert "Synced 1 mailmaps to swh.storage database" in out

    expected_displaynames = {
        "Original Name V A <<EMAIL>>": "Display Name V A",
        "Original Name V A 2 <<EMAIL>>": "Display Name V A",
        "Original Name V A 3 <<EMAIL>>": "Display Name V A",
    }

    assert get_displaynames(postgresql) == expected_displaynames

    assert (
        UserMailmap.objects.filter(
            from_email_verified=True,
            display_name_activated=True,
            mailmap_last_processing_date__isnull=False,
        ).count()
        == 1
    )


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_with_to_email(postgresql):
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    call_sync_mailmaps("--perform", dsn)

    expected_displaynames = {
        "Original Name V A <<EMAIL>>": "Display Name V A",
        "Original Name V A 2 <<EMAIL>>": "Display Name V A",
        "Original Name V A 3 <<EMAIL>>": "Display Name V A",
    }

    assert get_displaynames(postgresql) == expected_displaynames

    # Add a non-valid to_email
    with transaction.atomic():
        for mailmap in UserMailmap.objects.filter(
            from_email_verified=True, display_name_activated=True
        ):
            mailmap.to_email = "<EMAIL>"
            mailmap.save()

    call_sync_mailmaps("--perform", dsn)

    assert get_displaynames(postgresql) == expected_displaynames

    # Verify the relevant to_email
    with transaction.atomic():
        for mailmap in UserMailmap.objects.filter(
            from_email_verified=True, display_name_activated=True
        ):
            mailmap.to_email_verified = True
            mailmap.save()

    call_sync_mailmaps("--perform", dsn)

    new_displayname = "Display Name V A <<EMAIL>>"
    expected_displaynames = {
        "Original Name V A <<EMAIL>>": new_displayname,
        "Original Name V A 2 <<EMAIL>>": new_displayname,
        "Original Name V A 3 <<EMAIL>>": new_displayname,
    }

    assert get_displaynames(postgresql) == expected_displaynames


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_disable(postgresql):
    """Check that disabling a mailmap only happens once"""
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    # Do the initial mailmap sync
    call_sync_mailmaps("--perform", dsn)

    assert len(get_displaynames(postgresql)) == 3

    updated = 0
    # Disable a display name
    with transaction.atomic():
        # Cannot use update() because `last_update_date` would not be updated
        for mailmap in UserMailmap.objects.filter(
            from_email_verified=True, display_name_activated=True
        ):
            mailmap.display_name_activated = False

            mailmap.save()
            updated += 1

    assert updated == 1

    # Sync mailmaps again
    out = call_sync_mailmaps("--perform", dsn)
    assert "1 mailmaps to disable" in out

    assert get_displaynames(postgresql) == {}

    # Update a displayname by hand
    with postgresql.cursor() as cur:
        cur.execute(
            "UPDATE person SET displayname='Manual Display Name' "
            "WHERE fullname='<NAME> <<EMAIL>>'"
        )

    expected_displaynames = {
        "Original Name V A <<EMAIL>>": "Manual Display Name"
    }

    assert get_displaynames(postgresql) == expected_displaynames

    # Sync mailmaps one last time. No mailmaps should be disabled
    out = call_sync_mailmaps("--perform", dsn)
    assert "0 mailmaps to disable" in out

    assert get_displaynames(postgresql) == expected_displaynames
<commit_msg>mailmaps: Add test_mailmap_add_full<commit_after># Copyright (C) 2022  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information

import datetime
from io import StringIO
from typing import Dict

from psycopg2.extras import execute_values
import pytest

from django.core.management import call_command
from django.db import transaction

from swh.model.model import Person
from swh.web.auth.models import UserMailmap
from swh.web.auth.utils import MAILMAP_PERMISSION
from swh.web.common.utils import reverse
from swh.web.tests.utils import (
    check_api_post_response,
    check_http_get_response,
    create_django_permission,
)


@pytest.fixture
def mailmap_user(regular_user):
    regular_user.user_permissions.add(create_django_permission(MAILMAP_PERMISSION))
    return regular_user


@pytest.mark.django_db(transaction=True)
@pytest.mark.parametrize("view_name", ["profile-mailmap-add", "profile-mailmap-update"])
def test_mailmap_endpoints_anonymous_user(api_client, view_name):
    url = reverse(view_name)
    check_api_post_response(api_client, url, status_code=403)


@pytest.mark.django_db(transaction=True)
def test_mailmap_endpoints_user_with_permission(api_client, mailmap_user):
    api_client.force_login(mailmap_user)
    for view_name in ("profile-mailmap-add", "profile-mailmap-update"):
        url = reverse(view_name)
        check_api_post_response(
            api_client,
            url,
            data={"from_email": "<EMAIL>", "display_name": "bar"},
            status_code=200,
        )

    # FIXME: use check_api_get_responses; currently this crashes without
    # content_type="application/json"
    resp = check_http_get_response(
        api_client,
        reverse("profile-mailmap-list"),
        status_code=200,
        content_type="application/json",
    ).data
    assert len(resp) == 1
    assert resp[0]["from_email"] == "<EMAIL>"
    assert resp[0]["display_name"] == "bar"


@pytest.mark.django_db(transaction=True)
def test_mailmap_add_duplicate(api_client, mailmap_user):
    api_client.force_login(mailmap_user)

    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "bar"},
        status_code=200,
    )
    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "baz"},
        status_code=400,
    )


@pytest.mark.django_db(transaction=True)
def test_mailmap_add_full(api_client, mailmap_user):
    api_client.force_login(mailmap_user)

    request_data = {
        "from_email": "<EMAIL>",
        "from_email_verified": True,
        "from_email_verification_request_date": "2021-02-07T14:04:15Z",
        "display_name": "bar",
        "display_name_activated": True,
        "to_email": "<EMAIL>",
        "to_email_verified": True,
        "to_email_verification_request_date": "2021-02-07T15:54:59Z",
    }

    check_api_post_response(
        api_client, reverse("profile-mailmap-add"), data=request_data, status_code=200,
    )

    resp = check_http_get_response(
        api_client,
        reverse("profile-mailmap-list"),
        status_code=200,
        content_type="application/json",
    ).data
    assert len(resp) == 1
    assert resp[0].items() >= request_data.items()


@pytest.mark.django_db(transaction=True)
def test_mailmap_endpoints_error_response(api_client, mailmap_user):
    api_client.force_login(mailmap_user)

    url = reverse("profile-mailmap-add")
    resp = check_api_post_response(api_client, url, status_code=400)
    assert b"from_email" in resp.content

    url = reverse("profile-mailmap-update")
    resp = check_api_post_response(api_client, url, status_code=400)
    assert b"from_email" in resp.content


@pytest.mark.django_db(transaction=True)
def test_mailmap_update(api_client, mailmap_user):
    api_client.force_login(mailmap_user)

    before_add = datetime.datetime.now(tz=datetime.timezone.utc)

    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "Display Name 1"},
        status_code=200,
    )
    check_api_post_response(
        api_client,
        reverse("profile-mailmap-add"),
        data={"from_email": "<EMAIL>", "display_name": "Display Name 2"},
        status_code=200,
    )
    after_add = datetime.datetime.now(tz=datetime.timezone.utc)

    mailmaps = list(UserMailmap.objects.order_by("from_email").all())
    assert len(mailmaps) == 2, mailmaps

    assert mailmaps[0].from_email == "<EMAIL>", mailmaps
    assert mailmaps[0].display_name == "Display Name 1", mailmaps
    assert before_add <= mailmaps[0].last_update_date <= after_add

    assert mailmaps[1].from_email == "<EMAIL>", mailmaps
    assert mailmaps[1].display_name == "Display Name 2", mailmaps
    assert before_add <= mailmaps[0].last_update_date <= after_add

    before_update = datetime.datetime.now(tz=datetime.timezone.utc)

    check_api_post_response(
        api_client,
        reverse("profile-mailmap-update"),
        data={"from_email": "<EMAIL>", "display_name": "Display Name 1b"},
        status_code=200,
    )

    after_update = datetime.datetime.now(tz=datetime.timezone.utc)

    mailmaps = list(UserMailmap.objects.order_by("from_email").all())
    assert len(mailmaps) == 2, mailmaps

    assert mailmaps[0].from_email == "<EMAIL>", mailmaps
    assert mailmaps[0].display_name == "Display Name 1b", mailmaps
    assert before_update <= mailmaps[0].last_update_date <= after_update

    assert mailmaps[1].from_email == "<EMAIL>", mailmaps
    assert mailmaps[1].display_name == "Display Name 2", mailmaps
    assert before_add <= mailmaps[1].last_update_date <= after_add


def populate_mailmap():
    for (verified, activated) in (
        (False, False),
        (False, True),
        (True, False),
        (True, True),
    ):
        verified_str = "V" if verified else ""
        activated_str = "A" if activated else ""
        UserMailmap.objects.create(
            from_email=f"from_email{verified_str}{activated_<EMAIL>",
            display_name=f"Display Name {verified_str} {activated_str}".strip(),
            from_email_verified=verified,
            display_name_activated=activated,
        )


def call_sync_mailmaps(*args) -> str:
    out = StringIO()
    err = StringIO()

    call_command("sync_mailmaps", *args, stdout=out, stderr=err)

    out.seek(0)
    err.seek(0)
    assert err.read() == ""
    return out.read()


MAILMAP_KNOWN_FULLNAMES = (
    "Original Name <<EMAIL>>",
    "Original Name V <<EMAIL>>",
    "Original Name A <<EMAIL>>",
    "Original Name V A <<EMAIL>>",
    "Original Name V A 2 <<EMAIL>>",
    "Original Name V A 3 <<EMAIL>>",
)


MAILMAP_KNOWN_PEOPLE = tuple(
    Person.from_fullname(f.encode()) for f in MAILMAP_KNOWN_FULLNAMES
)


def init_stub_storage_db(postgresql):
    cur = postgresql.cursor()
    cur.execute(
        """
        CREATE TABLE person (
          fullname bytea PRIMARY KEY,
          name bytea,
          email bytea,
          displayname bytea
        )
        """
    )
    execute_values(
        cur,
        "INSERT INTO person (fullname, name, email) VALUES %s",
        (p.to_dict() for p in MAILMAP_KNOWN_PEOPLE),
        template="(%(fullname)s, %(name)s, %(email)s)",
    )
    cur.execute("CREATE INDEX ON person (email)")
    postgresql.commit()
    cur.close()

    return postgresql.dsn


def get_displaynames(postgresql) -> Dict[str, str]:
    with postgresql.cursor() as cur:
        cur.execute(
            "SELECT fullname, displayname FROM person WHERE displayname IS NOT NULL"
        )

        return {bytes(f).decode("utf-8"): bytes(d).decode("utf-8") for (f, d) in cur}


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_dry_run(postgresql):
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    out = call_sync_mailmaps(dsn)

    assert "(dry run)" in out
    assert "Synced 1 mailmaps to swh.storage database" in out

    assert get_displaynames(postgresql) == {}

    assert (
        UserMailmap.objects.filter(
            from_email_verified=True,
            display_name_activated=True,
            mailmap_last_processing_date__isnull=False,
        ).count()
        == 0
    )


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_perform(postgresql):
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    out = call_sync_mailmaps("--perform", dsn)

    assert "(dry run)" not in out
    assert "Synced 1 mailmaps to swh.storage database" in out

    expected_displaynames = {
        "Original Name V A <<EMAIL>>": "Display Name V A",
        "Original Name V A 2 <<EMAIL>>": "Display Name V A",
        "Original Name V A 3 <<EMAIL>>": "Display Name V A",
    }

    assert get_displaynames(postgresql) == expected_displaynames

    assert (
        UserMailmap.objects.filter(
            from_email_verified=True,
            display_name_activated=True,
            mailmap_last_processing_date__isnull=False,
        ).count()
        == 1
    )


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_with_to_email(postgresql):
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    call_sync_mailmaps("--perform", dsn)

    expected_displaynames = {
        "Original Name V A <<EMAIL>>": "Display Name V A",
        "Original Name V A 2 <<EMAIL>>": "Display Name V A",
        "Original Name V A 3 <<EMAIL>>": "Display Name V A",
    }

    assert get_displaynames(postgresql) == expected_displaynames

    # Add a non-valid to_email
    with transaction.atomic():
        for mailmap in UserMailmap.objects.filter(
            from_email_verified=True, display_name_activated=True
        ):
            mailmap.to_email = "<EMAIL>"
            mailmap.save()

    call_sync_mailmaps("--perform", dsn)

    assert get_displaynames(postgresql) == expected_displaynames

    # Verify the relevant to_email
    with transaction.atomic():
        for mailmap in UserMailmap.objects.filter(
            from_email_verified=True, display_name_activated=True
        ):
            mailmap.to_email_verified = True
            mailmap.save()

    call_sync_mailmaps("--perform", dsn)

    new_displayname = "Display Name V A <<EMAIL>>"
    expected_displaynames = {
        "Original Name V A <<EMAIL>>": new_displayname,
        "Original Name V A 2 <<EMAIL>>": new_displayname,
        "Original Name V A 3 <<EMAIL>>": new_displayname,
    }

    assert get_displaynames(postgresql) == expected_displaynames


@pytest.mark.django_db(transaction=True)
def test_sync_mailmaps_disable(postgresql):
    """Check that disabling a mailmap only happens once"""
    with transaction.atomic():
        populate_mailmap()

    dsn = init_stub_storage_db(postgresql)

    # Do the initial mailmap sync
    call_sync_mailmaps("--perform", dsn)

    assert len(get_displaynames(postgresql)) == 3

    updated = 0
    # Disable a display name
    with transaction.atomic():
        # Cannot use update() because `last_update_date` would not be updated
        for mailmap in UserMailmap.objects.filter(
            from_email_verified=True, display_name_activated=True
        ):
            mailmap.display_name_activated = False

            mailmap.save()
            updated += 1

    assert updated == 1

    # Sync mailmaps again
    out = call_sync_mailmaps("--perform", dsn)
    assert "1 mailmaps to disable" in out

    assert get_displaynames(postgresql) == {}

    # Update a displayname by hand
    with postgresql.cursor() as cur:
        cur.execute(
            "UPDATE person SET displayname='Manual Display Name' "
            "WHERE fullname='Original Name V A <<EMAIL>>'"
        )

    expected_displaynames = {
        "Original Name V A <<EMAIL>>": "Manual Display Name"
    }

    assert get_displaynames(postgresql) == expected_displaynames

    # Sync mailmaps one last time. No mailmaps should be disabled
    out = call_sync_mailmaps("--perform", dsn)
    assert "0 mailmaps to disable" in out

    assert get_displaynames(postgresql) == expected_displaynames

{'fraction_non_alphanumeric': 0.11074511112037694, 'fraction_numerical': 0.008339240295209107, 'mean_word_length': 3.250221513379408, 'pattern_counts': {'":': 52, '<': 120, '<?xml version=': 0, '>': 109, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6594228', 'n_tokens_mistral': 7768, 'n_tokens_neox': 7491, 'n_words': 1637}