diff options
Diffstat (limited to 'tests/managers')
| -rw-r--r-- | tests/managers/thl/test_ledger/test_thl_lm_tx.py | 48 | ||||
| -rw-r--r-- | tests/managers/thl/test_ledger/test_user_txs.py | 11 | ||||
| -rw-r--r-- | tests/managers/thl/test_maxmind.py | 495 | ||||
| -rw-r--r-- | tests/managers/thl/test_survey.py | 10 | ||||
| -rw-r--r-- | tests/managers/thl/test_survey_penalty.py | 6 | ||||
| -rw-r--r-- | tests/managers/thl/test_wall_manager.py | 18 |
6 files changed, 295 insertions, 293 deletions
diff --git a/tests/managers/thl/test_ledger/test_thl_lm_tx.py b/tests/managers/thl/test_ledger/test_thl_lm_tx.py index 1130621..90e5469 100644 --- a/tests/managers/thl/test_ledger/test_thl_lm_tx.py +++ b/tests/managers/thl/test_ledger/test_thl_lm_tx.py @@ -206,9 +206,7 @@ class TestThlLedgerTxManager: # Update the finished timestamp, but nothing else. This means that # there is no financial changes needed - session.update( - finished=datetime.now(tz=UTC) + timedelta(minutes=10) - ) + session.update(finished=datetime.now(tz=UTC) + timedelta(minutes=10)) assert session.finished with caplog.at_level(logging.INFO): tx = thl_lm.create_tx_bp_adjustment(session=session) @@ -827,7 +825,11 @@ class TestThlLedgerTxManagerFlows: status, status_code_1 = s1.determine_session_status() thl_net, commission_amount, bp_pay, user_pay = s1.determine_payments() s1.update( - status=status, status_code_1=status_code_1, finished=s1.started + timedelta(minutes=10), payout=bp_pay, user_payout=user_pay + status=status, + status_code_1=status_code_1, + finished=s1.started + timedelta(minutes=10), + payout=bp_pay, + user_payout=user_pay, ) print(thl_net, commission_amount, bp_pay, user_pay) thl_lm.create_tx_bp_payment(session=s1, created=w1.started) @@ -884,7 +886,11 @@ class TestThlLedgerTxManagerFlows: status, status_code_1 = session.determine_session_status() thl_net, commission_amount, bp_pay, user_pay = session.determine_payments() session.update( - status=status, status_code_1=status_code_1, finished=session.started + timedelta(minutes=10), payout=bp_pay, user_payout=user_pay + status=status, + status_code_1=status_code_1, + finished=session.started + timedelta(minutes=10), + payout=bp_pay, + user_payout=user_pay, ) print(thl_net, commission_amount, bp_pay, user_pay) @@ -916,7 +922,11 @@ class TestThlLedgerTxManagerFlows: status, status_code_1 = session.determine_session_status() # thl_net, commission_amount, bp_pay, user_pay = session.determine_payments() session.update( - status=status, status_code_1=status_code_1, finished=session.started + timedelta(minutes=10), payout=Decimal("1.53"), user_payout=Decimal("1.53") + status=status, + status_code_1=status_code_1, + finished=session.started + timedelta(minutes=10), + payout=Decimal("1.53"), + user_payout=Decimal("1.53"), ) thl_lm.create_tx_bp_payment(session=session, created=wall1.started) @@ -951,7 +961,11 @@ class TestThlLedgerTxManagerFlows: status, status_code_1 = session.determine_session_status() # thl_net, commission_amount, bp_pay, user_pay = session.determine_payments() session.update( - status=status, status_code_1=status_code_1, finished=session.started + timedelta(minutes=10), payout=Decimal("0.39"), user_payout=Decimal("0.26") + status=status, + status_code_1=status_code_1, + finished=session.started + timedelta(minutes=10), + payout=Decimal("0.39"), + user_payout=Decimal("0.26"), ) # with pytest.logs(logger, level=logging.WARNING) as cm: # tx = thl_lm.create_transaction_bp_payment(session, created=wall1.started) @@ -1284,7 +1298,11 @@ class TestThlLedgerManagerAdj: status, status_code_1 = session.determine_session_status() thl_net, commission_amount, bp_pay, user_pay = session.determine_payments() session.update( - status=status, status_code_1=status_code_1, finished=utc_hour_ago + timedelta(minutes=10), payout=bp_pay, user_payout=user_pay + status=status, + status_code_1=status_code_1, + finished=utc_hour_ago + timedelta(minutes=10), + payout=bp_pay, + user_payout=user_pay, ) thl_lm.create_tx_bp_payment(session, created=wall1.started) @@ -1452,7 +1470,11 @@ class TestThlLedgerManagerAdj: assert user_pay == Decimal("1.52") session.update( - status=status, status_code_1=status_code_1, finished=now + timedelta(minutes=10), payout=bp_pay, user_payout=user_pay + status=status, + status_code_1=status_code_1, + finished=now + timedelta(minutes=10), + payout=bp_pay, + user_payout=user_pay, ) tx = thl_lm.create_tx_bp_adjustment(session=session, created=wall1.started) @@ -1624,9 +1646,13 @@ class TestThlLedgerManagerAdj: session = Session(started=wall1.started, user=user, wall_events=[wall1, wall2]) status, status_code_1 = session.determine_session_status() - thl_net, commission_amount, bp_pay, user_pay = session.determine_payments() + _, _, bp_pay, user_pay = session.determine_payments() session.update( - status=status, status_code_1=status_code_1, finished=utc_hour_ago + timedelta(minutes=10), payout=bp_pay, user_payout=user_pay + status=status, + status_code_1=status_code_1, + finished=utc_hour_ago + timedelta(minutes=10), + payout=bp_pay, + user_payout=user_pay, ) thl_lm.create_tx_bp_payment(session, created=wall1.started) diff --git a/tests/managers/thl/test_ledger/test_user_txs.py b/tests/managers/thl/test_ledger/test_user_txs.py index f83641e..881eb03 100644 --- a/tests/managers/thl/test_ledger/test_user_txs.py +++ b/tests/managers/thl/test_ledger/test_user_txs.py @@ -60,7 +60,7 @@ def test_user_txs( amount=5, created=utc_now, payout_type=PayoutType.AMT_HIT, - request_data=dict(), + request_data={}, ) thl_lm.create_tx_user_payout_request( user=user, @@ -73,7 +73,7 @@ def test_user_txs( amount=127, created=utc_now, payout_type=PayoutType.AMT_BONUS, - request_data=dict(), + request_data={}, ) thl_lm.create_tx_user_payout_request( user=user, @@ -140,12 +140,7 @@ def test_user_txs_pagination( product_amt_true: Product, create_main_accounts: Callable[..., None], thl_lm: ThlLedgerManager, - lm: LedgerManager, delete_ledger_db: Callable[..., None], - session_with_tx_factory: Callable[..., Session], - adj_to_fail_with_tx_factory, - user_payout_event_manager, - utc_now: datetime, ): delete_ledger_db() create_main_accounts() @@ -253,7 +248,7 @@ def test_user_txs_rolling_balance( cashout_method_uuid=settings.amt_bonus_cashout_method_id, amount=150, payout_type=PayoutType.AMT_BONUS, - request_data=dict(), + request_data={}, ) thl_lm.create_tx_user_payout_request( user=user, diff --git a/tests/managers/thl/test_maxmind.py b/tests/managers/thl/test_maxmind.py index 75bf0e9..3e85cc3 100644 --- a/tests/managers/thl/test_maxmind.py +++ b/tests/managers/thl/test_maxmind.py @@ -1,18 +1,6 @@ -import json -import logging -from collections.abc import Callable - -import pytest from faker import Faker from faker.providers.address.en_US import Provider as USAddressProvider -from generalresearch.managers.thl.ipinfo import GeoIpInfoManager -from generalresearch.models.thl.ipinfo import ( - GeoIPInformation, - normalize_ip, -) -from generalresearch.models.thl.maxmind.definitions import UserType - fake = Faker() US_STATES = {x.lower() for x in USAddressProvider.states} @@ -24,245 +12,244 @@ IP_v6_US = "2600:1700:ece0:9410:55d:faf3:c15d:6e4" IP_v6_US_SAME_64 = "2600:1700:ece0:9410:55d:faf3:c15d:aaaa" -@pytest.fixture(scope="session") -def delete_ipinfo(thl_web_rw) -> Callable: - def _delete_ipinfo(ip): - thl_web_rw.execute_write( - query="DELETE FROM thl_geoname WHERE geoname_id IN (SELECT geoname_id FROM thl_ipinformation WHERE ip = %s);", - params=[ip], - ) - thl_web_rw.execute_write( - query="DELETE FROM thl_ipinformation WHERE ip = %s;", - params=[ip], - ) - - return _delete_ipinfo - - -class TestMaxmindBasicManager: - - def test_init(self, maxmind_basic_manager): - - assert isinstance(maxmind_basic_manager, MaxmindBasicManager) - - def test_get_basic_ip_information(self, maxmind_basic_manager): - ip = IP_v4_INDIA - maxmind_basic_manager.run_update_geoip_db() - - res1 = maxmind_basic_manager.get_basic_ip_information(ip_address=ip) - assert isinstance(res1, geoip2.models.Country) - assert res1.country.iso_code == "IN" - assert res1.country.name == "India" - - res2 = maxmind_basic_manager.get_basic_ip_information( - ip_address=fake.ipv4_private() - ) - assert res2 is None - - def test_get_country_iso_from_ip_geoip2db(self, maxmind_basic_manager): - ip = IP_v4_INDIA - maxmind_basic_manager.run_update_geoip_db() - - res1 = maxmind_basic_manager.get_country_iso_from_ip_geoip2db(ip=ip) - assert res1 == "in" - - res2 = maxmind_basic_manager.get_country_iso_from_ip_geoip2db( - ip=fake.ipv4_private() - ) - assert res2 is None - - def test_get_basic_ip_information_ipv6(self, maxmind_basic_manager): - ip = IP_v6_INDIA - maxmind_basic_manager.run_update_geoip_db() - - res1 = maxmind_basic_manager.get_basic_ip_information(ip_address=ip) - assert isinstance(res1, geoip2.models.Country) - assert res1.country.iso_code == "IN" - assert res1.country.name == "India" - - -class TestMaxmindManager: - - def test_init(self, thl_web_rr, thl_redis_config, maxmind_manager: MaxmindManager): - instance = MaxmindManager(pg_config=thl_web_rr, redis_config=thl_redis_config) - assert isinstance(instance, MaxmindManager) - assert isinstance(maxmind_manager, MaxmindManager) - - def test_create_basic( - self, - maxmind_manager: MaxmindManager, - geoipinfo_manager: GeoIpInfoManager, - delete_ipinfo, - ): - # This is (currently) an IP in India, and so it should only do the basic lookup - ip = IP_v4_INDIA - delete_ipinfo(ip) - geoipinfo_manager.clear_cache(ip) - assert geoipinfo_manager.get_cache(ip) is None - assert geoipinfo_manager.get_mysql_if_exists(ip) is None - - maxmind_manager.run_ip_information(ip, force_insights=False) - # Check that it is in the cache and in mysql - res = geoipinfo_manager.get_cache(ip) - assert res.ip == ip - assert res.basic - res = geoipinfo_manager.get_mysql(ip) - assert res.ip == ip - assert res.basic - - def test_create_basic_ipv6( - self, - maxmind_manager: MaxmindManager, - geoipinfo_manager: GeoIpInfoManager, - delete_ipinfo, - ): - # This is (currently) an IP in India, and so it should only do the basic lookup - ip = IP_v6_INDIA - normalized_ip, lookup_prefix = normalize_ip(ip) - delete_ipinfo(ip) - geoipinfo_manager.clear_cache(ip) - delete_ipinfo(normalized_ip) - geoipinfo_manager.clear_cache(normalized_ip) - assert geoipinfo_manager.get_cache(ip) is None - assert geoipinfo_manager.get_cache(normalized_ip) is None - assert geoipinfo_manager.get_mysql_if_exists(ip) is None - assert geoipinfo_manager.get_mysql_if_exists(normalized_ip) is None - - maxmind_manager.run_ip_information(ip, force_insights=False) - - # Check that it is in the cache - res = geoipinfo_manager.get_cache(ip) - # The looked up IP (/128) is returned, - assert res.ip == ip - assert res.lookup_prefix == "/64" - assert res.basic - - # ... but the normalized version was stored (/64) - assert geoipinfo_manager.get_cache_raw(ip) is None - res = json.loads(geoipinfo_manager.get_cache_raw(normalized_ip)) - assert res["ip"] == normalized_ip - - # Check mysql - res = geoipinfo_manager.get_mysql(ip) - assert res.ip == ip - assert res.lookup_prefix == "/64" - assert res.basic - with pytest.raises(AssertionError): - geoipinfo_manager.get_mysql_raw(ip) - res = geoipinfo_manager.get_mysql_raw(normalized_ip) - assert res["ip"] == normalized_ip - - def test_create_insights( - self, - maxmind_manager: MaxmindManager, - geoipinfo_manager: GeoIpInfoManager, - delete_ipinfo, - ): - # This is (currently) an IP in the US, so it should do insights - ip = IP_v4_US - delete_ipinfo(ip) - geoipinfo_manager.clear_cache(ip) - assert geoipinfo_manager.get_cache(ip) is None - assert geoipinfo_manager.get_mysql_if_exists(ip) is None - - res1 = maxmind_manager.run_ip_information(ip, force_insights=False) - assert isinstance(res1, GeoIPInformation) - - # Check that it is in the cache and in mysql - res2 = geoipinfo_manager.get_cache(ip) - assert isinstance(res2, GeoIPInformation) - assert res2.ip == ip - assert not res2.basic - - res3 = geoipinfo_manager.get_mysql(ip) - assert isinstance(res3, GeoIPInformation) - assert res3.ip == ip - assert not res3.basic - assert res3.is_anonymous is False - assert res3.subdivision_1_name.lower() in US_STATES - # this might change ... - assert res3.user_type == UserType.CELLULAR - - assert res1 == res2 == res3, "runner, cache, mysql all return same instance" - - def test_create_insights_ipv6( - self, - maxmind_manager: MaxmindManager, - geoipinfo_manager: GeoIpInfoManager, - delete_ipinfo, - ): - # This is (currently) an IP in the US, so it should do insights - ip = IP_v6_US - normalized_ip, lookup_prefix = normalize_ip(ip) - delete_ipinfo(ip) - geoipinfo_manager.clear_cache(ip) - delete_ipinfo(normalized_ip) - geoipinfo_manager.clear_cache(normalized_ip) - assert geoipinfo_manager.get_cache(ip) is None - assert geoipinfo_manager.get_cache(normalized_ip) is None - assert geoipinfo_manager.get_mysql_if_exists(ip) is None - assert geoipinfo_manager.get_mysql_if_exists(normalized_ip) is None - - res1 = maxmind_manager.run_ip_information(ip, force_insights=False) - assert isinstance(res1, GeoIPInformation) - assert res1.lookup_prefix == "/64" - - # Check that it is in the cache and in mysql - res2 = geoipinfo_manager.get_cache(ip) - assert isinstance(res2, GeoIPInformation) - assert res2.ip == ip - assert not res2.basic - - res3 = geoipinfo_manager.get_mysql(ip) - assert isinstance(res3, GeoIPInformation) - assert res3.ip == ip - assert not res3.basic - assert res3.is_anonymous is False - assert res3.subdivision_1_name.lower() in US_STATES - # this might change ... - assert res3.user_type == UserType.RESIDENTIAL - - assert res1 == res2 == res3, "runner, cache, mysql all return same instance" - - def test_get_or_create_ip_information(self, maxmind_manager): - ip = IP_v4_US - - res1 = maxmind_manager.get_or_create_ip_information(ip_address=ip) - assert isinstance(res1, GeoIPInformation) - - res2 = maxmind_manager.get_or_create_ip_information( - ip_address=fake.ipv4_private() - ) - assert res2 is None - - def test_get_or_create_ip_information_ipv6( - self, maxmind_manager, delete_ipinfo, geoipinfo_manager, caplog - ): - ip = IP_v6_US - normalized_ip, lookup_prefix = normalize_ip(ip) - delete_ipinfo(normalized_ip) - geoipinfo_manager.clear_cache(normalized_ip) - - with caplog.at_level(logging.INFO): - res1 = maxmind_manager.get_or_create_ip_information(ip_address=ip) - assert isinstance(res1, GeoIPInformation) - assert res1.ip == ip - # It looks up in insight using the normalize IP! - assert f"get_insights_ip_information: {normalized_ip}" in caplog.text - - # And it should NOT do the lookup again with an ipv6 in the same /64 block! - ip = IP_v6_US_SAME_64 - caplog.clear() - with caplog.at_level(logging.INFO): - res2 = maxmind_manager.get_or_create_ip_information(ip_address=ip) - assert isinstance(res2, GeoIPInformation) - assert res2.ip == ip - assert "get_insights_ip_information" not in caplog.text - - def test_run_ip_information(self, maxmind_manager): - ip = IP_v4_US - - res = maxmind_manager.run_ip_information(ip_address=ip) - assert isinstance(res, GeoIPInformation) - assert res.country_name == "United States" - assert res.country_iso == "us" +# @pytest.fixture(scope="session") +# def delete_ipinfo(thl_web_rw) -> Callable: +# def _delete_ipinfo(ip): +# thl_web_rw.execute_write( +# query="DELETE FROM thl_geoname WHERE geoname_id IN (SELECT geoname_id FROM thl_ipinformation WHERE ip = %s);", +# params=[ip], +# ) +# thl_web_rw.execute_write( +# query="DELETE FROM thl_ipinformation WHERE ip = %s;", +# params=[ip], +# ) + +# return _delete_ipinfo + + +# @pytest.skip("TODO: Replace with GRIP Client") +# class TestMaxmindBasicManager: + +# def test_init(self,): + +# def test_get_basic_ip_information(self, maxmind_basic_manager): +# ip = IP_v4_INDIA +# maxmind_basic_manager.run_update_geoip_db() + +# res1 = maxmind_basic_manager.get_basic_ip_information(ip_address=ip) +# # assert isinstance(res1, geoip2.models.Country) +# assert res1.country.iso_code == "IN" +# assert res1.country.name == "India" + +# res2 = maxmind_basic_manager.get_basic_ip_information( +# ip_address=fake.ipv4_private() +# ) +# assert res2 is None + +# def test_get_country_iso_from_ip_geoip2db(self, maxmind_basic_manager): +# ip = IP_v4_INDIA +# maxmind_basic_manager.run_update_geoip_db() + +# res1 = maxmind_basic_manager.get_country_iso_from_ip_geoip2db(ip=ip) +# assert res1 == "in" + +# res2 = maxmind_basic_manager.get_country_iso_from_ip_geoip2db( +# ip=fake.ipv4_private() +# ) +# assert res2 is None + +# def test_get_basic_ip_information_ipv6(self, maxmind_basic_manager): +# ip = IP_v6_INDIA +# maxmind_basic_manager.run_update_geoip_db() + +# res1 = maxmind_basic_manager.get_basic_ip_information(ip_address=ip) +# assert isinstance(res1, geoip2.models.Country) +# assert res1.country.iso_code == "IN" +# assert res1.country.name == "India" + + +# class TestMaxmindManager: + +# def test_init(self, thl_web_rr, thl_redis_config, maxmind_manager: MaxmindManager): +# instance = MaxmindManager(pg_config=thl_web_rr, redis_config=thl_redis_config) +# assert isinstance(instance, MaxmindManager) +# assert isinstance(maxmind_manager, MaxmindManager) + +# def test_create_basic( +# self, +# maxmind_manager: MaxmindManager, +# geoipinfo_manager: GeoIpInfoManager, +# delete_ipinfo, +# ): +# # This is (currently) an IP in India, and so it should only do the basic lookup +# ip = IP_v4_INDIA +# delete_ipinfo(ip) +# geoipinfo_manager.clear_cache(ip) +# assert geoipinfo_manager.get_cache(ip) is None +# assert geoipinfo_manager.get_mysql_if_exists(ip) is None + +# maxmind_manager.run_ip_information(ip, force_insights=False) +# # Check that it is in the cache and in mysql +# res = geoipinfo_manager.get_cache(ip) +# assert res.ip == ip +# assert res.basic +# res = geoipinfo_manager.get_mysql(ip) +# assert res.ip == ip +# assert res.basic + +# def test_create_basic_ipv6( +# self, +# maxmind_manager: MaxmindManager, +# geoipinfo_manager: GeoIpInfoManager, +# delete_ipinfo, +# ): +# # This is (currently) an IP in India, and so it should only do the basic lookup +# ip = IP_v6_INDIA +# normalized_ip, lookup_prefix = normalize_ip(ip) +# delete_ipinfo(ip) +# geoipinfo_manager.clear_cache(ip) +# delete_ipinfo(normalized_ip) +# geoipinfo_manager.clear_cache(normalized_ip) +# assert geoipinfo_manager.get_cache(ip) is None +# assert geoipinfo_manager.get_cache(normalized_ip) is None +# assert geoipinfo_manager.get_mysql_if_exists(ip) is None +# assert geoipinfo_manager.get_mysql_if_exists(normalized_ip) is None + +# maxmind_manager.run_ip_information(ip, force_insights=False) + +# # Check that it is in the cache +# res = geoipinfo_manager.get_cache(ip) +# # The looked up IP (/128) is returned, +# assert res.ip == ip +# assert res.lookup_prefix == "/64" +# assert res.basic + +# # ... but the normalized version was stored (/64) +# assert geoipinfo_manager.get_cache_raw(ip) is None +# res = json.loads(geoipinfo_manager.get_cache_raw(normalized_ip)) +# assert res["ip"] == normalized_ip + +# # Check mysql +# res = geoipinfo_manager.get_mysql(ip) +# assert res.ip == ip +# assert res.lookup_prefix == "/64" +# assert res.basic +# with pytest.raises(AssertionError): +# geoipinfo_manager.get_mysql_raw(ip) +# res = geoipinfo_manager.get_mysql_raw(normalized_ip) +# assert res["ip"] == normalized_ip + +# def test_create_insights( +# self, +# maxmind_manager: MaxmindManager, +# geoipinfo_manager: GeoIpInfoManager, +# delete_ipinfo, +# ): +# # This is (currently) an IP in the US, so it should do insights +# ip = IP_v4_US +# delete_ipinfo(ip) +# geoipinfo_manager.clear_cache(ip) +# assert geoipinfo_manager.get_cache(ip) is None +# assert geoipinfo_manager.get_mysql_if_exists(ip) is None + +# res1 = maxmind_manager.run_ip_information(ip, force_insights=False) +# assert isinstance(res1, GeoIPInformation) + +# # Check that it is in the cache and in mysql +# res2 = geoipinfo_manager.get_cache(ip) +# assert isinstance(res2, GeoIPInformation) +# assert res2.ip == ip +# assert not res2.basic + +# res3 = geoipinfo_manager.get_mysql(ip) +# assert isinstance(res3, GeoIPInformation) +# assert res3.ip == ip +# assert not res3.basic +# assert res3.is_anonymous is False +# assert res3.subdivision_1_name.lower() in US_STATES +# # this might change ... +# assert res3.user_type == UserType.CELLULAR + +# assert res1 == res2 == res3, "runner, cache, mysql all return same instance" + +# def test_create_insights_ipv6( +# self, +# maxmind_manager: MaxmindManager, +# geoipinfo_manager: GeoIpInfoManager, +# delete_ipinfo, +# ): +# # This is (currently) an IP in the US, so it should do insights +# ip = IP_v6_US +# normalized_ip, lookup_prefix = normalize_ip(ip) +# delete_ipinfo(ip) +# geoipinfo_manager.clear_cache(ip) +# delete_ipinfo(normalized_ip) +# geoipinfo_manager.clear_cache(normalized_ip) +# assert geoipinfo_manager.get_cache(ip) is None +# assert geoipinfo_manager.get_cache(normalized_ip) is None +# assert geoipinfo_manager.get_mysql_if_exists(ip) is None +# assert geoipinfo_manager.get_mysql_if_exists(normalized_ip) is None + +# res1 = maxmind_manager.run_ip_information(ip, force_insights=False) +# assert isinstance(res1, GeoIPInformation) +# assert res1.lookup_prefix == "/64" + +# # Check that it is in the cache and in mysql +# res2 = geoipinfo_manager.get_cache(ip) +# assert isinstance(res2, GeoIPInformation) +# assert res2.ip == ip +# assert not res2.basic + +# res3 = geoipinfo_manager.get_mysql(ip) +# assert isinstance(res3, GeoIPInformation) +# assert res3.ip == ip +# assert not res3.basic +# assert res3.is_anonymous is False +# assert res3.subdivision_1_name.lower() in US_STATES +# # this might change ... +# assert res3.user_type == UserType.RESIDENTIAL + +# assert res1 == res2 == res3, "runner, cache, mysql all return same instance" + +# def test_get_or_create_ip_information(self, maxmind_manager): +# ip = IP_v4_US + +# res1 = maxmind_manager.get_or_create_ip_information(ip_address=ip) +# assert isinstance(res1, GeoIPInformation) + +# res2 = maxmind_manager.get_or_create_ip_information( +# ip_address=fake.ipv4_private() +# ) +# assert res2 is None + +# def test_get_or_create_ip_information_ipv6( +# self, maxmind_manager, delete_ipinfo, geoipinfo_manager, caplog +# ): +# ip = IP_v6_US +# normalized_ip, lookup_prefix = normalize_ip(ip) +# delete_ipinfo(normalized_ip) +# geoipinfo_manager.clear_cache(normalized_ip) + +# with caplog.at_level(logging.INFO): +# res1 = maxmind_manager.get_or_create_ip_information(ip_address=ip) +# assert isinstance(res1, GeoIPInformation) +# assert res1.ip == ip +# # It looks up in insight using the normalize IP! +# assert f"get_insights_ip_information: {normalized_ip}" in caplog.text + +# # And it should NOT do the lookup again with an ipv6 in the same /64 block! +# ip = IP_v6_US_SAME_64 +# caplog.clear() +# with caplog.at_level(logging.INFO): +# res2 = maxmind_manager.get_or_create_ip_information(ip_address=ip) +# assert isinstance(res2, GeoIPInformation) +# assert res2.ip == ip +# assert "get_insights_ip_information" not in caplog.text + +# def test_run_ip_information(self, maxmind_manager): +# ip = IP_v4_US + +# res = maxmind_manager.run_ip_information(ip_address=ip) +# assert isinstance(res, GeoIPInformation) +# assert res.country_name == "United States" +# assert res.country_iso == "us" diff --git a/tests/managers/thl/test_survey.py b/tests/managers/thl/test_survey.py index 37f0b66..117a6aa 100644 --- a/tests/managers/thl/test_survey.py +++ b/tests/managers/thl/test_survey.py @@ -161,9 +161,9 @@ class TestSurvey: calc_answers={"i:adhoc_13126": ("3", "4")}, ), ] - uqad = dict() + uqad = {} for uqa in uqas: - for k, v in uqa.calc_answers.items(): + for k, _ in uqa.calc_answers.items(): if k in qualifying_questions: uqad[k] = uqa uqad[uqa.property_code] = uqa @@ -234,7 +234,7 @@ class TestSurveyStat: ): survey = surveys_fixture[0].model_copy() surveys = [] - for idx in range(20_000): + for _ in range(20_000): s = survey.model_copy() s.survey_id = uuid.uuid4().hex surveys.append(s) @@ -269,9 +269,9 @@ class TestSurveyStat: s.conv_beta = 20 s.updated_at = now # and 1,000 don't change - print(f"----c-----: {datetime.now().isoformat()}") + print(f"----c-----: {datetime.now(tz=UTC).isoformat()}") res2 = surveystat_manager.update_or_create(ss) - print(f"----d-----: {datetime.now().isoformat()}") + print(f"----d-----: {datetime.now(tz=UTC).isoformat()}") assert len(res2) == 20_000 def test_ymsp( diff --git a/tests/managers/thl/test_survey_penalty.py b/tests/managers/thl/test_survey_penalty.py index 2a3cdc2..c7862bb 100644 --- a/tests/managers/thl/test_survey_penalty.py +++ b/tests/managers/thl/test_survey_penalty.py @@ -88,10 +88,8 @@ class TestSurveyPenalty: ) assert res == {"t:a": 0.1, "t:b": 0.2, "u:b": 0.1} assert surveypenalty_manager.cache.currsize == 1 - cached_key = tuple(list(list(surveypenalty_manager.cache.keys())[0])[1:]) - assert cached_key == tuple( - ["product_id", product_uuid, "team_id", team_id_random] - ) + cached_key = tuple(list(next(iter(surveypenalty_manager.cache.keys())))[1:]) + assert cached_key == ("product_id", product_uuid, "team_id", team_id_random) # Both don't exist, return nothing res = surveypenalty_manager.get_penalties_for( diff --git a/tests/managers/thl/test_wall_manager.py b/tests/managers/thl/test_wall_manager.py index 5abc648..067a29e 100644 --- a/tests/managers/thl/test_wall_manager.py +++ b/tests/managers/thl/test_wall_manager.py @@ -77,12 +77,12 @@ class TestWallManager: assert isinstance(res, list) assert len(res) == 50 - res1 = list(set([w.session_id for w in res])) + res1 = list({w.session_id for w in res}) res1.sort() assert session_ids == res1 - def test_create_wall(self, wall_manager, session_manager, user, session): + def test_create_wall(self, wall_manager, user, session): w = wall_manager.create( session_id=session.id, user_id=user.user_id, @@ -98,9 +98,7 @@ class TestWallManager: w2 = wall_manager.get_from_uuid(wall_uuid=w.uuid) assert w == w2 - def test_report_wall_abandon( - self, wall_manager, session_manager, user, session, utc_hour_ago - ): + def test_report_wall_abandon(self, wall_manager, user, session, utc_hour_ago): w1 = wall_manager.create( session_id=session.id, user_id=user.user_id, @@ -176,12 +174,10 @@ class TestWallManager: assert Status.COMPLETE == w2.status assert "This survey blows!" == w2.report_notes - def test_filter_wall_attempts( - self, wall_manager, session_manager, user, session, utc_hour_ago - ): + def test_filter_wall_attempts(self, wall_manager, user, session, utc_hour_ago): res = wall_manager.filter_wall_attempts(user_id=user.user_id) assert len(res) == 0 - w1 = wall_manager.create( + wall_manager.create( session_id=session.id, user_id=user.user_id, uuid_id=uuid4().hex, @@ -193,7 +189,7 @@ class TestWallManager: ) res = wall_manager.filter_wall_attempts(user_id=user.user_id) assert len(res) == 1 - w2 = wall_manager.create( + wall_manager.create( session_id=session.id, user_id=user.user_id, uuid_id=uuid4().hex, @@ -221,7 +217,7 @@ class TestWallCacheManager: start3 = datetime.now(UTC) - timedelta(hours=1) session = session_manager.create_dummy(started=start1, user=user) - wall1 = wall_manager.create_dummy( + wall_manager.create_dummy( session_id=session.id, user_id=session.user_id, started=start1, |