maigret/tests/test_sites.py

"""Maigret Database test functions"""

import logging
import re

from typing import Any, Dict

from maigret.sites import MaigretDatabase, MaigretEngine, MaigretSite

EXAMPLE_DB: Dict[str, Any] = {
    'engines': {
        "XenForo": {
            "presenseStrs": ["XenForo"],
            "site": {
                "absenceStrs": [
                    "The specified member cannot be found. Please enter a member's entire name.",
                ],
                "checkType": "message",
                "errors": {"You must be logged-in to do that.": "Login required"},
                "url": "{urlMain}{urlSubpath}/members/?username={username}",
            },
        },
    },
    'sites': {
        "Amperka": {
            "engine": "XenForo",
            "rank": 121613,
            "tags": ["ru"],
            "urlMain": "http://forum.amperka.ru",
            "usernameClaimed": "adam",
            "usernameUnclaimed": "noonewouldeverusethis7",
        },
    },
}


def test_load_empty_db_from_str():
    db = MaigretDatabase()
    db.load_from_str('{"engines": {}, "sites": {}}')

    assert db.sites == []
    assert db.engines == []


def test_load_valid_db():
    db = MaigretDatabase()
    db.load_from_json(EXAMPLE_DB)

    assert len(db.sites) == 1
    assert len(db.engines) == 1

    assert db.sites[0].name == 'Amperka'
    assert db.engines[0].name == 'XenForo'


def test_site_json_dump():
    db = MaigretDatabase()
    db.load_from_json(EXAMPLE_DB)

    init_keys = EXAMPLE_DB['sites']['Amperka'].keys()
    # contains engine data
    obj_keys = db.sites[0].json.keys()

    assert set(init_keys).issubset(set(obj_keys))


def test_site_correct_initialization():
    db = MaigretDatabase()
    db.load_from_json(EXAMPLE_DB)

    xenforo = db.engines[0]
    assert xenforo.name == 'XenForo'
    assert xenforo.site['checkType'] == 'message'

    amperka = db.sites[0]
    assert amperka.name == 'Amperka'
    assert amperka.check_type == 'message'


def test_site_stats_are_instance_local():
    first = MaigretSite('First', {})
    second = MaigretSite('Second', {})

    first.stats['presense_flag'] = 'profile marker'

    assert first.stats is not second.stats
    assert 'presense_flag' not in second.stats


def test_site_strip_engine_data():
    db = MaigretDatabase()
    db.load_from_json(EXAMPLE_DB)

    amperka = db.sites[0]
    amperka_stripped = amperka.strip_engine_data()

    assert amperka_stripped.json == EXAMPLE_DB['sites']['Amperka']


def test_site_strip_engine_data_with_site_prior_updates():
    db = MaigretDatabase()
    UPDATED_EXAMPLE_DB = dict(EXAMPLE_DB)
    UPDATED_EXAMPLE_DB['sites']['Amperka']['absenceStrs'] = ["test"]
    db.load_from_json(UPDATED_EXAMPLE_DB)

    amperka = db.sites[0]
    amperka_stripped = amperka.strip_engine_data()

    assert amperka_stripped.json == UPDATED_EXAMPLE_DB['sites']['Amperka']


def test_saving_site_error():
    db = MaigretDatabase()

    DB = dict(EXAMPLE_DB)
    DB['sites']['Amperka']['errors'] = {'error1': 'text1'}

    db.load_from_json(DB)

    amperka = db.sites[0]
    assert len(amperka.errors) == 2
    assert len(amperka.errors_dict) == 2

    assert amperka.strip_engine_data().errors == {'error1': 'text1'}
    assert amperka.strip_engine_data().json['errors'] == {'error1': 'text1'}


def test_site_scalar_field_wins_over_engine(caplog):
    """An engine is a template; a value the entry states itself is an exception.

    Before this, the engine overwrote the site's scalar fields, so a
    hand-written `urlProbe` was ignored at runtime - and then removed from
    data.json by strip_engine_data on the next save, without a warning.
    """
    site = MaigretSite(
        'Example',
        {
            'urlMain': 'https://example.com',
            'urlProbe': 'https://example.com/site-specific?u={username}',
        },
    )
    engine = MaigretEngine(
        'ExampleEngine',
        {
            'site': {
                'url': '{urlMain}/u/{username}',
                'urlProbe': '{urlMain}/engine-default?u={username}',
                'checkType': 'message',
            },
        },
    )

    with caplog.at_level(logging.WARNING):
        site.update_from_engine(engine)

    assert site.url_probe == 'https://example.com/site-specific?u={username}'
    # fields the site said nothing about still come from the engine
    assert site.url == '{urlMain}/u/{username}'
    assert site.check_type == 'message'
    assert any('overrides engine' in r.getMessage() for r in caplog.records)


def test_strip_engine_data_keeps_site_override():
    """The override has to survive a save, or it disappears from the database."""
    site = MaigretSite(
        'Example',
        {
            'urlMain': 'https://example.com',
            'urlProbe': 'https://example.com/site-specific?u={username}',
        },
    )
    engine = MaigretEngine(
        'ExampleEngine',
        {
            'site': {
                'url': '{urlMain}/u/{username}',
                'urlProbe': '{urlMain}/engine-default?u={username}',
                'checkType': 'message',
            },
        },
    )
    site.update_from_engine(engine)

    stripped = site.strip_engine_data()

    assert stripped.json['urlProbe'] == 'https://example.com/site-specific?u={username}'
    # what did come from the engine is still stripped
    assert 'url' not in stripped.json
    assert 'checkType' not in stripped.json


def test_constructor_defaults_do_not_count_as_site_overrides(caplog):
    """Only what the entry stated counts - not what __init__ filled in.

    `alexa_rank` is set to sys.maxsize for every entry that does not carry one,
    so keying the rule off the instance dict made seventeen op.gg sites keep
    that placeholder instead of the rank their engine supplies.
    """
    site = MaigretSite('Example', {'urlMain': 'https://example.com'})
    engine = MaigretEngine(
        'ExampleEngine',
        {'site': {'url': '{urlMain}/u/{username}', 'alexaRank': 331}},
    )

    with caplog.at_level(logging.WARNING):
        site.update_from_engine(engine)

    assert site.alexa_rank == 331
    assert not [r for r in caplog.records if 'overrides engine' in r.getMessage()]


def test_update_from_engine_warns_on_conflicting_dict_entries(caplog):
    site = MaigretSite(
        'Example',
        {
            'urlMain': 'https://example.com',
            'url': 'https://example.com/{username}',
            'errors': {
                'Shared marker': 'site value',
                'Same marker': 'same value',
            },
        },
    )
    engine = MaigretEngine(
        'ExampleEngine',
        {
            'site': {
                'errors': {
                    'Shared marker': 'engine value',
                    'Same marker': 'same value',
                    'Engine marker': 'engine-only value',
                },
            },
        },
    )

    with caplog.at_level(logging.WARNING):
        site.update_from_engine(engine)

    assert site.errors == {
        'Shared marker': 'engine value',
        'Same marker': 'same value',
        'Engine marker': 'engine-only value',
    }
    assert 'Example' in caplog.text
    assert 'errors' in caplog.text
    assert 'Shared marker' in caplog.text
    assert 'Same marker' not in caplog.text


def test_site_url_detector():
    db = MaigretDatabase()
    db.load_from_json(EXAMPLE_DB)

    assert (
        db.sites[0].url_regexp.pattern
        == r'^https?://(www\.|m\.)?forum\.amperka\.ru/members/\?username=(.+?)$'
    )
    assert (
        db.sites[0].detect_username('http://forum.amperka.ru/members/?username=test')
        == 'test'
    )


def test_extract_id_from_url_skips_none_groups():
    site = MaigretSite(
        "Example",
        {
            "urlMain": "https://example.com",
            "url": "https://example.com/{username}",
        },
    )
    site.url_regexp = re.compile(r"^https://example\.com/([^/?#]+)(?:/(.*))?$")

    assert site.extract_id_from_url("https://example.com/username") == (
        "username",
        "username",
    )


def test_extract_id_from_url_handles_literal_dollar_prefix():
    site = MaigretSite(
        "Cash App",
        {
            "urlMain": "https://cash.app",
            "url": "https://cash.app/${username}",
        },
    )

    assert site.extract_id_from_url("https://cash.app/$alice") == (
        "alice",
        "username",
    )


def test_ranked_sites_dict():
    db = MaigretDatabase()
    db.update_site(MaigretSite('3', {'alexaRank': 1000, 'engine': 'ucoz'}))
    db.update_site(MaigretSite('1', {'alexaRank': 2, 'tags': ['forum']}))
    db.update_site(MaigretSite('2', {'alexaRank': 10, 'tags': ['ru', 'forum']}))

    # sorting
    assert list(db.ranked_sites_dict().keys()) == ['1', '2', '3']
    assert list(db.ranked_sites_dict(top=2).keys()) == ['1', '2']
    assert list(db.ranked_sites_dict(reverse=True, top=2).keys()) == ['3', '2']

    # filtering by tags
    assert list(db.ranked_sites_dict(tags=['ru'], top=2).keys()) == ['2']
    assert list(db.ranked_sites_dict(tags=['forum']).keys()) == ['1', '2']

    # filtering by engine
    assert list(db.ranked_sites_dict(tags=['ucoz']).keys()) == ['3']

    # disjunction
    assert list(db.ranked_sites_dict(names=['2'], tags=['forum']).keys()) == ['2']
    assert list(db.ranked_sites_dict(names=['2'], tags=['ucoz']).keys()) == []
    assert list(db.ranked_sites_dict(names=['4'], tags=['ru']).keys()) == []

    # reverse
    assert list(db.ranked_sites_dict(reverse=True).keys()) == ['3', '2', '1']


def test_ranked_sites_dict_names():
    db = MaigretDatabase()
    db.update_site(MaigretSite('3', {'alexaRank': 30}))
    db.update_site(MaigretSite('1', {'alexaRank': 2}))
    db.update_site(MaigretSite('2', {'alexaRank': 10}))

    # filtering by names
    assert list(db.ranked_sites_dict(names=['1', '2']).keys()) == ['1', '2']
    assert list(db.ranked_sites_dict(names=['2', '3']).keys()) == ['2', '3']


def test_ranked_sites_dict_disabled():
    db = MaigretDatabase()
    db.update_site(MaigretSite('1', {'disabled': True}))
    db.update_site(MaigretSite('2', {}))

    assert len(db.ranked_sites_dict()) == 2
    assert len(db.ranked_sites_dict(disabled=False)) == 1


def test_ranked_sites_dict_id_type():
    db = MaigretDatabase()
    db.update_site(MaigretSite('1', {}))
    db.update_site(MaigretSite('2', {'type': 'username'}))
    db.update_site(MaigretSite('3', {'type': 'gaia_id'}))

    assert len(db.ranked_sites_dict()) == 2
    assert len(db.ranked_sites_dict(id_type='username')) == 2
    assert len(db.ranked_sites_dict(id_type='gaia_id')) == 1


def test_ranked_sites_dict_excluded_tags():
    db = MaigretDatabase()
    db.update_site(MaigretSite('3', {'alexaRank': 1000, 'engine': 'ucoz'}))
    db.update_site(MaigretSite('1', {'alexaRank': 2, 'tags': ['forum']}))
    db.update_site(MaigretSite('2', {'alexaRank': 10, 'tags': ['ru', 'forum']}))

    # excluding by tag
    assert list(db.ranked_sites_dict(excluded_tags=['ru']).keys()) == ['1', '3']
    assert list(db.ranked_sites_dict(excluded_tags=['forum']).keys()) == ['3']

    # excluding by engine
    assert list(db.ranked_sites_dict(excluded_tags=['ucoz']).keys()) == ['1', '2']

    # combining include and exclude tags
    assert list(db.ranked_sites_dict(tags=['forum'], excluded_tags=['ru']).keys()) == [
        '1'
    ]

    # excluding non-existent tag has no effect
    assert list(db.ranked_sites_dict(excluded_tags=['nonexistent']).keys()) == [
        '1',
        '2',
        '3',
    ]

    # exclude all
    assert list(db.ranked_sites_dict(excluded_tags=['forum', 'ucoz']).keys()) == []


def test_ranked_sites_dict_tag_filter_is_case_insensitive():
    # The include (whitelist) tag filter must be case-insensitive, like the
    # exclude (blacklist) filter and every sibling lambda (name/source/engine),
    # all of which lowercase the site-side value. A site tagged 'US' must be
    # found by tags=['us'] just as it is excluded by excluded_tags=['us'].
    db = MaigretDatabase()
    db.update_site(MaigretSite('1', {'alexaRank': 2, 'tags': ['US']}))

    assert list(db.ranked_sites_dict(tags=['us']).keys()) == ['1']
    # the blacklist already treats the same tag case-insensitively
    assert list(db.ranked_sites_dict(excluded_tags=['us']).keys()) == []


def test_ranked_sites_dict_excluded_tags_with_top():
    """Excluded tags should also prevent mirrors from being included."""
    db = MaigretDatabase()
    db.update_site(
        MaigretSite('Parent', {'alexaRank': 1, 'tags': ['forum'], 'type': 'username'})
    )
    db.update_site(
        MaigretSite(
            'Mirror',
            {
                'alexaRank': 999999,
                'source': 'Parent',
                'tags': ['forum'],
                'type': 'username',
            },
        )
    )
    db.update_site(
        MaigretSite('Other', {'alexaRank': 2, 'tags': ['coding'], 'type': 'username'})
    )

    # Without exclusion, mirror should be included
    result = db.ranked_sites_dict(top=1, id_type='username')
    assert 'Parent' in result
    assert 'Mirror' in result

    # With exclusion of 'forum', both Parent and Mirror should be excluded
    result = db.ranked_sites_dict(top=2, excluded_tags=['forum'], id_type='username')
    assert 'Parent' not in result
    assert 'Mirror' not in result
    assert 'Other' in result


def test_ranked_sites_dict_mirrors_disabled_parent():
    """Mirror is included when parent ranks in top N but parent is disabled."""
    db = MaigretDatabase()
    db.update_site(
        MaigretSite(
            'ParentPlatform',
            {'alexaRank': 5, 'disabled': True, 'type': 'username'},
        )
    )
    db.update_site(
        MaigretSite(
            'OtherSite',
            {'alexaRank': 100, 'type': 'username'},
        )
    )
    db.update_site(
        MaigretSite(
            'MirrorSite',
            {
                'alexaRank': 99999999,
                'source': 'ParentPlatform',
                'type': 'username',
            },
        )
    )

    result = db.ranked_sites_dict(top=1, disabled=False, id_type='username')
    assert list(result.keys()) == ['OtherSite', 'MirrorSite']


def test_ranked_sites_dict_mirrors_no_extra_without_parent_in_top():
    db = MaigretDatabase()
    db.update_site(MaigretSite('A', {'alexaRank': 1, 'type': 'username'}))
    db.update_site(
        MaigretSite(
            'B',
            {'alexaRank': 2, 'source': 'NotInDb', 'type': 'username'},
        )
    )

    assert list(db.ranked_sites_dict(top=1, id_type='username').keys()) == ['A']


def test_get_url_template():
    site = MaigretSite(
        "test",
        {
            "urlMain": "https://ya.ru/",
            "url": "{urlMain}{urlSubpath}/members/?username={username}",
        },
    )
    assert (
        site.get_url_template()
        == "{urlMain}{urlSubpath}/members/?username={username} (no engine)"
    )

    site = MaigretSite(
        "test",
        {
            "urlMain": "https://ya.ru/",
            "url": "https://{username}.ya.ru",
        },
    )
    assert site.get_url_template() == "SUBDOMAIN"


def test_update_site_replaces_existing_entry():
    """update_site() must replace the list element, not just rebind a loop variable."""
    db = MaigretDatabase()
    db.update_site(
        MaigretSite('Example', {'urlMain': 'https://example.com', 'disabled': False})
    )

    updated = MaigretSite(
        'Example', {'urlMain': 'https://example.com', 'disabled': True}
    )
    db.update_site(updated)

    # The database must contain exactly one entry and it must be the updated one
    assert len(db.sites) == 1
    assert db.sites_dict['Example'].disabled is True


def test_update_site_appends_when_name_not_found():
    """update_site() must append when no site with that name exists."""
    db = MaigretDatabase()
    db.update_site(MaigretSite('Alpha', {'urlMain': 'https://alpha.com'}))
    db.update_site(MaigretSite('Beta', {'urlMain': 'https://beta.com'}))

    assert len(db.sites) == 2
    assert 'Alpha' in db.sites_dict
    assert 'Beta' in db.sites_dict


def test_has_site_url_or_name(default_db):
    # by the same url or partial match
    assert default_db.has_site("https://aback.com.ua/user/") == True
    assert default_db.has_site("https://aback.com.ua") == True

    # acceptable partial match
    assert default_db.has_site("https://aback.com.ua/use") == True
    assert default_db.has_site("https://aback.com") == True

    # by name
    assert default_db.has_site("Aback") == True

    # false
    assert default_db.has_site("https://aeifgoai3h4g8a3u4g5") == False
    assert default_db.has_site("aeifgoai3h4g8a3u4g5") == False