23385248069

Committed 21 Mar 2026 05:50PM UTC coverage: 73.202% (-2.5%) from 75.69%

Build # 23385248069

Build Type

push

github

Committed by

RustyBower

Commit Message

Fix test to match DB fallback behavior for missing output files

The endpoint now returns check.output from DB (200) instead of 404
when the on-disk file doesn't exist.

Coverage Stats

3726 of 5090 relevant lines covered (73.2%)

0.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.33

/scoring_engine/cache_helper.py

"""Helper functions for clearing and updating application caches."""

from flask import current_app
from flask_caching.backends import NullCache

from scoring_engine.cache import cache


def update_all_cache(app_or_ctx=None):
    """Clear and rebuild all cached values.

    Parameters
    ----------
    app_or_ctx : Flask app or app context, optional
        If provided, the cache will be cleared within this application's
        context.  If omitted, the current application context will be used.
    """

    if app_or_ctx is None:
        app_or_ctx = current_app

    context_manager = app_or_ctx.app_context() if hasattr(app_or_ctx, "app_context") else app_or_ctx

    with context_manager:
        cache.clear()

    update_overview_data()
    update_scoreboard_data()
    update_team_stats()
    update_services_navbar()
    update_service_data()
    update_services_data()
    update_announcements_data()
    update_sla_data()
    update_flags_data()
    update_stats()


def update_overview_data():
    from scoring_engine.web.views.api.overview import (
        overview_get_data,
        overview_get_round_data,
        _get_overview_data_cached,
        _get_table_columns_cached,
    )

    cache.delete_memoized(overview_get_data)
    cache.delete_memoized(overview_get_round_data)
    cache.delete_memoized(_get_overview_data_cached)
    cache.delete_memoized(_get_table_columns_cached)


def update_scoreboard_data():
    from scoring_engine.web.views.api.scoreboard import _get_bar_data_cached, _get_line_data_cached

    cache.delete_memoized(_get_bar_data_cached)
    cache.delete_memoized(_get_line_data_cached)


def update_team_stats(team_id=None):
    # corresponds with file scoring_engine.web.views.api.team function services_get_team_data

    if team_id is not None:
        cache.delete(f"/api/team/{team_id}/stats_team_{team_id}")
    elif not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/team/*/stats_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_services_navbar(team_id=None):
    # corresponds with file scoring_engine.web.views.api.team function team_services_status

    if team_id is not None:
        cache.delete(f"/api/team/{team_id}/services/status_team_{team_id}")
    elif not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/team/*/services/status_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_service_data(service_id=None):
    # corresponds with file scoring_engine.web.views.api.service function service_get_checks

    if not isinstance(cache.cache, NullCache):
        pattern = f"*/api/service/{service_id}/checks_*" if service_id is not None else "*/api/service/*/checks_*"
        for key in cache.cache._write_client.scan_iter(match=pattern):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_services_data(team_id=None):
    # corresponds with file scoring_engine.web.views.api.team function api_services

    if team_id is not None:
        cache.delete(f"/api/team/{team_id}/services_team_{team_id}")
    elif not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/team/*/services_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_announcements_data():
    """Clear cached announcement data for all visibility contexts."""
    if not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/announcements_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_inject_data(inject_id, team_id=None):
    """Clear cached inject detail for the given inject.

    The cache key for ``/api/inject/<id>`` is ``/api/inject/<id>_team_<team_id>``
    for blue teams or ``/api/inject/<id>_white`` for white team.
    Both the owning team and the white team can view an inject, so we clear
    all matching keys when ``team_id`` is not provided.
    """
    if team_id is not None:
        cache.delete(f"/api/inject/{inject_id}_team_{team_id}")
    elif not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match=f"*/api/inject/{inject_id}_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_inject_comments(inject_id, team_id=None):
    """Clear cached inject comments for the given inject."""
    if team_id is not None:
        cache.delete(f"/api/inject/{inject_id}/comments_{team_id}")
    elif not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match=f"*/api/inject/{inject_id}/comments_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_inject_files(inject_id, team_id=None):
    """Clear cached inject files for the given inject."""
    if team_id is not None:
        cache.delete(f"/api/inject/{inject_id}/files_{team_id}")
    elif not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match=f"*/api/inject/{inject_id}/files_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_all_inject_data():
    """Clear all cached inject detail and inject list data for all teams."""
    if not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/inject*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_sla_data():
    # Clear cached /api/sla responses (keyed per-team/role)
    if not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/sla*_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_flags_data():
    # Clear cached /api/flags responses (keyed per-team/role)
    if not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/flags*_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))


def update_stats():
    # Clear cached /api/stats responses (keyed per-team/role)
    if not isinstance(cache.cache, NullCache):
        for key in cache.cache._write_client.scan_iter(match="*/api/stats_*"):
            cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))

1	"""Helper functions for clearing and updating application caches."""
2
3	from flask import current_app	1✔
4	from flask_caching.backends import NullCache	1✔
5
6	from scoring_engine.cache import cache	1✔
7
8
9	def update_all_cache(app_or_ctx=None):	1✔
10	"""Clear and rebuild all cached values.
11
12	Parameters
13	----------
14	app_or_ctx : Flask app or app context, optional
15	If provided, the cache will be cleared within this application's
16	context. If omitted, the current application context will be used.
17	"""
18
19	if app_or_ctx is None:	1✔
20	app_or_ctx = current_app	×
21
22	context_manager = app_or_ctx.app_context() if hasattr(app_or_ctx, "app_context") else app_or_ctx	1✔
23
24	with context_manager:	1✔
25	cache.clear()	1✔
26
27	update_overview_data()	1✔
28	update_scoreboard_data()	1✔
29	update_team_stats()	1✔
30	update_services_navbar()	1✔
31	update_service_data()	1✔
32	update_services_data()	1✔
33	update_announcements_data()	1✔
34	update_sla_data()	1✔
35	update_flags_data()	1✔
36	update_stats()	1✔
37
38
39	def update_overview_data():	1✔
40	from scoring_engine.web.views.api.overview import (	1✔
41	overview_get_data,
42	overview_get_round_data,
43	_get_overview_data_cached,
44	_get_table_columns_cached,
45	)
46
47	cache.delete_memoized(overview_get_data)	1✔
48	cache.delete_memoized(overview_get_round_data)	1✔
49	cache.delete_memoized(_get_overview_data_cached)	1✔
50	cache.delete_memoized(_get_table_columns_cached)	1✔
51
52
53	def update_scoreboard_data():	1✔
54	from scoring_engine.web.views.api.scoreboard import _get_bar_data_cached, _get_line_data_cached	1✔
55
56	cache.delete_memoized(_get_bar_data_cached)	1✔
57	cache.delete_memoized(_get_line_data_cached)	1✔
58
59
60	def update_team_stats(team_id=None):	1✔
61	# corresponds with file scoring_engine.web.views.api.team function services_get_team_data
62
63	if team_id is not None:	1✔
64	cache.delete(f"/api/team/{team_id}/stats_team_{team_id}")	1✔
65	elif not isinstance(cache.cache, NullCache):	1✔
66	for key in cache.cache._write_client.scan_iter(match="/api/team//stats_*"):	×
67	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
68
69
70	def update_services_navbar(team_id=None):	1✔
71	# corresponds with file scoring_engine.web.views.api.team function team_services_status
72
73	if team_id is not None:	1✔
74	cache.delete(f"/api/team/{team_id}/services/status_team_{team_id}")	1✔
75	elif not isinstance(cache.cache, NullCache):	1✔
76	for key in cache.cache._write_client.scan_iter(match="/api/team//services/status_*"):	×
77	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
78
79
80	def update_service_data(service_id=None):	1✔
81	# corresponds with file scoring_engine.web.views.api.service function service_get_checks
82
83	if not isinstance(cache.cache, NullCache):	1✔
84	pattern = f"/api/service/{service_id}/checks_" if service_id is not None else "/api/service//checks_*"	×
85	for key in cache.cache._write_client.scan_iter(match=pattern):	×
86	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
87
88
89	def update_services_data(team_id=None):	1✔
90	# corresponds with file scoring_engine.web.views.api.team function api_services
91
92	if team_id is not None:	1✔
93	cache.delete(f"/api/team/{team_id}/services_team_{team_id}")	1✔
94	elif not isinstance(cache.cache, NullCache):	1✔
95	for key in cache.cache._write_client.scan_iter(match="/api/team//services_*"):	×
96	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
97
98
99	def update_announcements_data():	1✔
100	"""Clear cached announcement data for all visibility contexts."""
101	if not isinstance(cache.cache, NullCache):	1✔
102	for key in cache.cache._write_client.scan_iter(match="/api/announcements_"):	×
103	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
104
105
106	def update_inject_data(inject_id, team_id=None):	1✔
107	"""Clear cached inject detail for the given inject.
108
109	The cache key for ``/api/inject/<id>`` is ``/api/inject/<id>_team_<team_id>``
110	for blue teams or ``/api/inject/<id>_white`` for white team.
111	Both the owning team and the white team can view an inject, so we clear
112	all matching keys when ``team_id`` is not provided.
113	"""
114	if team_id is not None:	×
115	cache.delete(f"/api/inject/{inject_id}_team_{team_id}")	×
116	elif not isinstance(cache.cache, NullCache):	×
117	for key in cache.cache._write_client.scan_iter(match=f"/api/inject/{inject_id}_"):	×
118	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
119
120
121	def update_inject_comments(inject_id, team_id=None):	1✔
122	"""Clear cached inject comments for the given inject."""
123	if team_id is not None:	1✔
124	cache.delete(f"/api/inject/{inject_id}/comments_{team_id}")	×
125	elif not isinstance(cache.cache, NullCache):	1✔
126	for key in cache.cache._write_client.scan_iter(match=f"/api/inject/{inject_id}/comments_"):	×
127	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
128
129
130	def update_inject_files(inject_id, team_id=None):	1✔
131	"""Clear cached inject files for the given inject."""
132	if team_id is not None:	×
133	cache.delete(f"/api/inject/{inject_id}/files_{team_id}")	×
134	elif not isinstance(cache.cache, NullCache):	×
135	for key in cache.cache._write_client.scan_iter(match=f"/api/inject/{inject_id}/files_"):	×
136	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
137
138
139	def update_all_inject_data():	1✔
140	"""Clear all cached inject detail and inject list data for all teams."""
141	if not isinstance(cache.cache, NullCache):	1✔
142	for key in cache.cache._write_client.scan_iter(match="/api/inject"):	×
143	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
144
145
146	def update_sla_data():	1✔
147	# Clear cached /api/sla responses (keyed per-team/role)
148	if not isinstance(cache.cache, NullCache):	1✔
149	for key in cache.cache._write_client.scan_iter(match="/api/sla_*"):	×
150	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
151
152
153	def update_flags_data():	1✔
154	# Clear cached /api/flags responses (keyed per-team/role)
155	if not isinstance(cache.cache, NullCache):	1✔
156	for key in cache.cache._write_client.scan_iter(match="/api/flags_*"):	×
157	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×
158
159
160	def update_stats():	1✔
161	# Clear cached /api/stats responses (keyed per-team/role)
162	if not isinstance(cache.cache, NullCache):	1✔
163	for key in cache.cache._write_client.scan_iter(match="/api/stats_"):	×
164	cache.delete(key.decode("utf-8").removeprefix(cache.cache.key_prefix))	×

scoringengine / scoringengine / 23385248069

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous