diff --git a/bases/bot_detector/runemetrics_scraper/core.py b/bases/bot_detector/runemetrics_scraper/core.py index 41d2ec5..46cd9af 100644 --- a/bases/bot_detector/runemetrics_scraper/core.py +++ b/bases/bot_detector/runemetrics_scraper/core.py @@ -13,7 +13,11 @@ ) from bot_detector.event_queue.core import Queue, QueueProducer from bot_detector.event_queue.factory import QueueFactory -from bot_detector.event_queue.structs import NotFoundStruct, ScrapedStruct +from bot_detector.event_queue.structs import ( + NotFoundStruct, + PlayerBannedStruct, + ScrapedStruct, +) from bot_detector.proxy_manager import ProxyManager from bot_detector.proxy_manager import Settings as ProxySettings from bot_detector.retry_tracker import RetryTracker @@ -191,6 +195,7 @@ async def work( rate_limiter: RateLimiter, player_nf_queue: Queue[NotFoundStruct], player_sc_producer: QueueProducer[ScrapedStruct], + player_banned_producer: QueueProducer[PlayerBannedStruct], ): retry_tracker = RetryTracker(base_delay=10.0, max_delay=300.0, decay_window=300.0) async with ClientSession() as session: @@ -257,10 +262,25 @@ async def work( continue # update player data + old_label = player_data.label_jagex player_data = await update_player( player_data=player_data, runemetrics_response=runemetrics_response, ) + + # emit a ban event only on the transition into label_jagex = 2 + if old_label != 2 and player_data.label_jagex == 2: + banned_event = PlayerBannedStruct( + metadata=MetaData(version=1, source="runemetrics_scraper"), + player_id=player_data.id, + name=player_data.name, + ) + produce_error = await player_banned_producer.put([banned_event]) + if produce_error: + logger.error( + f"[{worker_id}]: Failed to produce banned player: {produce_error}" + ) + # create hiscore data try: scraped_data = ScrapedStruct( @@ -338,15 +358,30 @@ async def main(): ), ), ) - for queue in (player_nf_queue, player_sc_producer): + player_banned_producer = QueueFactory.create_queue( + model=PlayerBannedStruct, + queue_type="producer", + backend_type="kafka", + config=KafkaConfig( + topic="players.banned", + bootstrap_servers=b_server, + producer=True, + producer_config=KafkaProducerConfig( + partition_key_fn=lambda message: str(message.player_id % 10) + ), + ), + ) + for queue in (player_nf_queue, player_sc_producer, player_banned_producer): if isinstance(queue, Exception): raise queue assert isinstance(player_nf_queue, Queue) assert isinstance(player_sc_producer, QueueProducer) + assert isinstance(player_banned_producer, QueueProducer) await player_nf_queue.start() await player_sc_producer.start() + await player_banned_producer.start() # start workers workers = [ @@ -360,6 +395,7 @@ async def main(): ), player_nf_queue=player_nf_queue, player_sc_producer=player_sc_producer, + player_banned_producer=player_banned_producer, ) ) for worker_id in range(len(proxies)) diff --git a/test/bases/bot_detector/runemetrics_scraper/test_core.py b/test/bases/bot_detector/runemetrics_scraper/test_core.py index fbcccd0..8710454 100644 --- a/test/bases/bot_detector/runemetrics_scraper/test_core.py +++ b/test/bases/bot_detector/runemetrics_scraper/test_core.py @@ -112,6 +112,7 @@ async def test_work_commits_offset_after_successful_handle( rate_limiter=AsyncMock(), player_nf_queue=player_nf_queue, player_sc_producer=player_sc_producer, + player_banned_producer=AsyncMock(), ) player_nf_queue.commit.assert_awaited_once() @@ -161,6 +162,7 @@ async def test_work_commits_only_after_successful_requeue( rate_limiter=AsyncMock(), player_nf_queue=player_nf_queue, player_sc_producer=player_sc_producer, + player_banned_producer=AsyncMock(), ) assert player_nf_queue.commit.await_count == expected_commit_calls @@ -252,6 +254,7 @@ async def test_work_backoff_increases_across_consecutive_errors( rate_limiter=AsyncMock(), player_nf_queue=player_nf_queue, player_sc_producer=player_sc_producer, + player_banned_producer=AsyncMock(), ) delays = [c[0][0] for c in sleep_mock.call_args_list] @@ -319,6 +322,7 @@ async def test_work_backoff_resets_after_success( rate_limiter=AsyncMock(), player_nf_queue=player_nf_queue, player_sc_producer=player_sc_producer, + player_banned_producer=AsyncMock(), ) delays = [c[0][0] for c in sleep_mock.call_args_list] @@ -375,7 +379,79 @@ async def test_work_validation_error_requeues_without_backoff( rate_limiter=AsyncMock(), player_nf_queue=player_nf_queue, player_sc_producer=player_sc_producer, + player_banned_producer=AsyncMock(), ) player_nf_queue.put.assert_awaited_once() sleep_mock.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "old_label,scrape_error,should_emit", + [ + (1, "NOT_A_MEMBER", True), + (0, "NOT_A_MEMBER", True), + (2, "NOT_A_MEMBER", False), + (0, None, False), + (0, "NO_PROFILE", False), + (1, None, False), + ], +) +async def test_work_emits_banned_event_only_on_transition_into_banned( + monkeypatch: pytest.MonkeyPatch, + player_struct: PlayerStruct, + old_label: int, + scrape_error: str | None, + should_emit: bool, +): + player = player_struct.model_copy(update={"label_jagex": old_label}) + player_message = NotFoundStruct( + metadata=MetaData(version=1, source="test"), + player_data=player, + ) + player_nf_queue = AsyncMock() + player_nf_queue.get_one = AsyncMock(return_value=player_message) + player_nf_queue.commit = AsyncMock(return_value=None) + player_sc_producer = AsyncMock() + player_sc_producer.put = AsyncMock(return_value=None) + player_banned_producer = AsyncMock() + player_banned_producer.put = AsyncMock(return_value=None) + + runemetrics_response = RuneMetricsResponse() + if scrape_error is not None: + runemetrics_response.error = RuneMetricsError( + error=scrape_error, + loggedIn=False, + ) + + monkeypatch.setattr(core, "ClientSession", _DummySession) + monkeypatch.setattr( + core, + "get_proxy", + AsyncMock(side_effect=["http://user@proxy", asyncio.CancelledError()]), + ) + monkeypatch.setattr( + core, + "scrape_player", + AsyncMock(return_value=(runemetrics_response, 0.1, None)), + ) + monkeypatch.setattr(core.asyncio, "sleep", AsyncMock(return_value=None)) + + with pytest.raises(asyncio.CancelledError): + await core.work( + worker_id=1, + proxy_manager=AsyncMock(), + rate_limiter=AsyncMock(), + player_nf_queue=player_nf_queue, + player_sc_producer=player_sc_producer, + player_banned_producer=player_banned_producer, + ) + + if should_emit: + player_banned_producer.put.assert_awaited_once() + event = player_banned_producer.put.call_args.args[0][0] + assert event.player_id == player.id + assert event.name == player.name + else: + player_banned_producer.put.assert_not_awaited()