"""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_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_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_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