Skip to content
Open
14 changes: 14 additions & 0 deletions tests/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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:
"""
Expand Down
12 changes: 8 additions & 4 deletions utility.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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}")

Expand Down
Loading