diff --git a/src/constants.py b/src/constants.py index 94da16d..7eb39a5 100644 --- a/src/constants.py +++ b/src/constants.py @@ -37,6 +37,9 @@ # threshold to generate an alert for violating UDP UDP_SENSITIVITY_THRESHOLD = 0.005 +# threshold to generate an alert for high score +HIGH_SCORE_THRESHOLD_ETH = 10 + # relevant addresses SETTLEMENT_CONTRACT_ADDRESS = "0x9008D19f58AAbD9eD0D60971565AA8510560ab41" MEV_BLOCKER_KICKBACKS_ADDRESSES = [ diff --git a/src/daemon.py b/src/daemon.py index ec3bd6b..1a01cfb 100644 --- a/src/daemon.py +++ b/src/daemon.py @@ -16,6 +16,9 @@ from src.monitoring_tests.mev_blocker_kickbacks_test import ( MEVBlockerRefundsMonitoringTest, ) +from src.monitoring_tests.high_score_test import ( + HighScoreTest, +) from src.constants import SLEEP_TIME_IN_SEC @@ -29,6 +32,7 @@ def main() -> None: tests = [ SolverCompetitionSurplusTest(), MEVBlockerRefundsMonitoringTest(), + HighScoreTest(), ] start_block: Optional[int] = None diff --git a/src/monitoring_tests/high_score_test.py b/src/monitoring_tests/high_score_test.py new file mode 100644 index 0000000..c4eb085 --- /dev/null +++ b/src/monitoring_tests/high_score_test.py @@ -0,0 +1,55 @@ +""" +Checking winner's score and generating an alert if score is very large +""" + +# pylint: disable=logging-fstring-interpolation + +from typing import Any +from src.monitoring_tests.base_test import BaseTest +from src.apis.orderbookapi import OrderbookAPI +from src.constants import HIGH_SCORE_THRESHOLD_ETH + + +class HighScoreTest(BaseTest): + """ + This test checks how large is the winning score and raises an alert if score + is above certain threshold + """ + + def __init__(self) -> None: + super().__init__() + self.orderbook_api = OrderbookAPI() + + def compute_winning_score(self, competition_data: dict[str, Any]) -> bool: + """ + This function simply returns the winning score. + """ + solution = competition_data["solutions"][-1] + score = int(solution["score"]) / 10**18 + log_output = "\t".join( + [ + "Large score test:", + f"Tx Hash: {competition_data['transactionHash']}", + f"Winning Solver: {solution['solver']}", + f"Score in ETH: {score}", + ] + ) + if score > HIGH_SCORE_THRESHOLD_ETH: + self.alert(log_output) + return True + + def run(self, tx_hash: str) -> bool: + """ + Wrapper function for the whole test. Checks if solver competition data is retrievable + and then checks how large the winning score is. + """ + + solver_competition_data = self.orderbook_api.get_solver_competition_data( + tx_hash + ) + if solver_competition_data is None: + return False + + success = self.compute_winning_score(solver_competition_data) + + return success diff --git a/tests/e2e/high_score_test.py b/tests/e2e/high_score_test.py new file mode 100644 index 0000000..3b1a9db --- /dev/null +++ b/tests/e2e/high_score_test.py @@ -0,0 +1,24 @@ +""" +Tests for large score test. +""" + +import unittest +from src.monitoring_tests.high_score_test import ( + HighScoreTest, +) + + +class TestHighScore(unittest.TestCase): + def test_high_score(self) -> None: + high_score_test = HighScoreTest() + # large score tx + tx_hash = "0x5eef22d04a2f30e62df76614decf43e1cc92ab957285a687f182a0191d85d15a" + self.assertTrue(high_score_test.run(tx_hash)) + + # small score tx + tx_hash = "0x37e7dbc2f3df5f31f017634d07855933c6d82f4fda033daebd4377987d2b5ae9" + self.assertTrue(high_score_test.run(tx_hash)) + + +if __name__ == "__main__": + unittest.main()