diff --git a/tests/base.py b/tests/base.py index f15c81ab..7f6e0d19 100644 --- a/tests/base.py +++ b/tests/base.py @@ -11,6 +11,7 @@ from sqlalchemy import text from werkzeug.datastructures import Headers +import utility from database import create_session from mod_auth.models import Role, User from mod_customized.models import CustomizedTest, TestFork @@ -278,6 +279,15 @@ def create_app(self, mock_config, mock_storage_client): def setUp(self): """Set up all entities.""" + from datetime import datetime + + import utility + + # Reset the state directly for test isolation. No mocks needed! + utility.cached_load_time = datetime(1970, 1, 1) + utility.cached_web_hook_blocks = [] + + super().setUp() self.app.preprocess_request() g.db = create_session( self.app.config['DATABASE_URI'], drop_tables=True) @@ -396,6 +406,10 @@ def setUp(self): g.db.add_all(forbidden_ext) g.db.commit() + def tearDown(self): + """Clean up after every test.""" + super().tearDown() + @staticmethod def create_login_form_data(email, password) -> dict: """ diff --git a/utility.py b/utility.py index 96308e41..bd341307 100644 --- a/utility.py +++ b/utility.py @@ -78,17 +78,18 @@ def decorated_function(*args, **kwargs): cached_web_hook_blocks: List[str] = [] +cached_load_time: datetime = datetime(1970, 1, 1) -def cache_has_expired() -> bool: +def cache_has_expired(load_time: datetime) -> bool: """ Check if the cache expired. :return: True if the cache was last updated more than one hour ago. :rtype: bool """ - cached_load_time = datetime(1970, 1, 1) - return cached_load_time + timedelta(hours=1) < datetime.now() + from datetime import datetime, timedelta + return load_time + timedelta(hours=1) < datetime.now() def is_github_web_hook_ip(request_ip: Union[IPv4Address, IPv6Address]) -> bool: @@ -115,15 +116,18 @@ def get_cached_web_hook_blocks() -> List[str]: :rtype: List[str] """ global cached_web_hook_blocks + global cached_load_time from run import config - if len(cached_web_hook_blocks) == 0 or cache_has_expired(): + if len(cached_web_hook_blocks) == 0 or cache_has_expired(cached_load_time): client_id = config.get('GITHUB_CLIENT_ID', '') client_secret = config.get('GITHUB_CLIENT_KEY', '') meta_json = requests.get( 'https://api.github.com/meta', auth=(client_id, client_secret)).json() try: cached_web_hook_blocks = meta_json['hooks'] + # We successfully fetched the IPs so we reset the clock + cached_load_time = datetime.now() except KeyError: g.log.critical(f"Failed to retrieve hook IP's from GitHub! API returned {meta_json}")