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

76.76

/scoring_engine/web/views/api/overview.py

from collections import defaultdict

from flask import jsonify
from flask_login import current_user
from sqlalchemy import desc
from sqlalchemy.sql import func

from scoring_engine.cache import cache
from scoring_engine.db import db
from scoring_engine.models.check import Check
from scoring_engine.models.round import Round
from scoring_engine.models.service import Service
from scoring_engine.models.setting import Setting
from scoring_engine.models.team import Team
from scoring_engine.sla import apply_dynamic_scoring_to_round, calculate_team_total_penalties, get_sla_config

from . import mod


def get_anonymize_mode():
    """
    Determine how team names should be displayed.
    Returns: (anonymize, show_both) tuple
        - (True, False): Show only anonymous names (public/blue teams)
        - (False, True): Show "RealName (Anonymous)" (white team when enabled)
        - (False, False): Show only real names (setting disabled)
    """
    setting = Setting.get_setting("anonymize_team_names")
    anonymize_enabled = setting.value is True if setting else False

    if not anonymize_enabled:
        return (False, False)

    # White team sees both names when anonymization is enabled
    if current_user.is_authenticated and current_user.is_white_team:
        return (False, True)

    # Everyone else sees only anonymous names
    return (True, False)


def calculate_ranks(score_dict):
    """
    Calculate ranks for a dict of {id: score} with tie handling.
    Returns dict of {id: rank} where ties get the same rank.
    """
    if not score_dict:
        return {}
    sorted_items = sorted(score_dict.items(), key=lambda x: x[1], reverse=True)
    ranks = {}
    current_rank = 1
    prev_score = None
    for i, (item_id, score) in enumerate(sorted_items):
        if prev_score is not None and score < prev_score:
            current_rank = i + 1
        ranks[item_id] = current_rank
        prev_score = score
    return ranks


@cache.memoize()
def _get_table_columns_cached(anonymize, show_both):
    """Internal cached function for table columns."""
    blue_teams = Team.get_all_blue_teams()
    team_name_map = Team.get_team_name_mapping(anonymize=anonymize, show_both=show_both)

    columns = []
    columns.append({"title": "", "data": "", "color": None})
    for team in blue_teams:
        display_name = team_name_map.get(team.id, team.name)
        columns.append({"title": display_name, "data": display_name, "color": team.rgb_color})
    return columns


@cache.memoize()
def _get_overview_data_cached(anonymize, show_both):
    """Internal cached function for overview data with host/port info."""
    data = defaultdict(lambda: defaultdict(dict))
    team_name_map = Team.get_team_name_mapping(anonymize=anonymize, show_both=show_both)

    round_obj = db.session.query(Round).order_by(Round.number.desc()).first()
    if round_obj:
        checks = (
            db.session.query(Check.service_id, Check.result)
            .join(Round)
            .filter(Round.number == round_obj.number)
            .subquery()
        )
        res = (
            db.session.query(Team.name, Service.name, Service.host, Service.port, checks.c.result)
            .join(Team)
            .filter(checks.c.service_id == Service.id)
            .all()
        )

        for r in res:
            team_name = r[0]
            display_name = team_name_map.get(team_name, team_name)
            data[display_name][r[1]] = {
                "host": r[2],
                "port": r[3],
                "passing": r[4],
            }

    # Convert defaultdict to regular dict for JSON serialization
    return {k: dict(v) for k, v in data.items()}


@mod.route("/api/overview/get_round_data")
@cache.memoize()
def overview_get_round_data():
    round_obj = db.session.query(Round).order_by(Round.number.desc()).first()
    if round_obj:
        round_start = round_obj.local_round_start
        number = round_obj.number
    else:
        round_start = ""
        number = 0
    engine_paused = Setting.get_setting("engine_paused").value
    data = {"round_start": round_start, "number": number, "engine_paused": engine_paused}
    return jsonify(data)


@mod.route("/api/overview/data")
def overview_data():
    """Get overview data with host/port info. Cached by user type."""
    anonymize, show_both = get_anonymize_mode()
    return jsonify(_get_overview_data_cached(anonymize, show_both))


@mod.route("/api/overview/get_columns")
def overview_get_columns():
    """Get table columns. Cached by user type."""
    anonymize, show_both = get_anonymize_mode()
    return jsonify(columns=_get_table_columns_cached(anonymize, show_both))


@mod.route("/api/overview/get_data")
@cache.memoize()
def overview_get_data():
    """Get overview table data. This endpoint returns positional data matching columns."""
    data = []
    blue_teams = db.session.query(Team).filter(Team.color == "Blue").order_by(Team.id).all()
    blue_team_ids = [team.id for team in blue_teams]
    blue_teams_dict = {team.id: team for team in blue_teams}
    last_round = Round.get_last_round_num()

    # Get SLA configuration
    sla_config = get_sla_config()

    current_scores = ["Current Score"]
    current_places = ["Current Place"]
    service_ratios = ["Up/Down Ratio"]
    sla_penalties_row = ["SLA Penalties"]

    num_up_services = dict(
        db.session.query(
            Service.team_id,
            func.count(Service.team_id),
        )
        .join(Check)
        .join(Round)
        .filter(Check.result.is_(True))
        .filter(Round.number == last_round)
        .group_by(Service.team_id)
        .all()
    )

    num_down_services = dict(
        db.session.query(
            Service.team_id,
            func.count(Service.team_id),
        )
        .join(Check)
        .join(Round)
        .filter(Check.result.is_(False))
        .filter(Round.number == last_round)
        .group_by(Service.team_id)
        .all()
    )

    if len(blue_team_ids) > 0:
        # Calculate team scores with dynamic scoring multipliers
        if sla_config.dynamic_enabled:
            # Query scores per round for dynamic scoring
            round_scores = (
                db.session.query(
                    Service.team_id,
                    Check.round_id,
                    func.sum(Service.points).label("round_score"),
                )
                .join(Check)
                .filter(Check.result.is_(True))
                .group_by(Service.team_id, Check.round_id)
                .all()
            )

            # Get round numbers for each round_id
            rounds_map = {r.id: r.number for r in db.session.query(Round.id, Round.number).all()}

            # Calculate totals with multipliers
            team_scores = defaultdict(int)
            for team_id, round_id, round_score in round_scores:
                round_number = rounds_map.get(round_id, 0)
                adjusted_score = apply_dynamic_scoring_to_round(round_number, round_score, sla_config)
                team_scores[team_id] += adjusted_score
            team_scores = dict(team_scores)
        else:
            # No dynamic scoring - use simple sum
            team_scores = dict(
                db.session.query(Service.team_id, func.sum(Service.points).label("score"))
                .join(Check)
                .filter(Check.result.is_(True))
                .group_by(Service.team_id)
                .order_by(desc("score"))
                .all()
            )

        # Calculate adjusted scores with SLA penalties
        adjusted_scores_dict = {}
        penalties_dict = {}
        for blue_team_id in blue_team_ids:
            base_score = team_scores.get(blue_team_id, 0)
            if sla_config.sla_enabled:
                team = blue_teams_dict[blue_team_id]
                penalty = calculate_team_total_penalties(team, sla_config)
                penalties_dict[blue_team_id] = penalty
                if sla_config.allow_negative:
                    adjusted_scores_dict[blue_team_id] = base_score - penalty
                else:
                    adjusted_scores_dict[blue_team_id] = max(0, base_score - penalty)
            else:
                penalties_dict[blue_team_id] = 0
                adjusted_scores_dict[blue_team_id] = base_score

        # Use adjusted scores for ranking when SLA is enabled
        scores_for_ranking = adjusted_scores_dict if sla_config.sla_enabled else team_scores

        # Calculate ranks with tie handling
        ranks_dict = calculate_ranks(scores_for_ranking)

        for blue_team_id in blue_team_ids:
            # Show adjusted score when SLA is enabled, base score otherwise
            if sla_config.sla_enabled:
                current_scores.append(str(adjusted_scores_dict.get(blue_team_id, 0)))
            else:
                current_scores.append(str(team_scores.get(blue_team_id, 0)))
            current_places.append(str(ranks_dict.get(blue_team_id, 0)))
            service_ratios.append(
                '<span class="text-success">{0} <i class="bi bi-arrow-up"></i></span> / '
                '<span class="text-danger">{1} <i class="bi bi-arrow-down"></i></span>'.format(
                    num_up_services.get(blue_team_id, 0),
                    num_down_services.get(blue_team_id, 0),
                )
            )
            # Add penalty display (negative number if penalty exists)
            penalty = penalties_dict.get(blue_team_id, 0)
            if penalty > 0:
                sla_penalties_row.append('<span class="text-danger">-{}</span>'.format(penalty))
            else:
                sla_penalties_row.append("0")

        data.append(current_scores)
        data.append(current_places)
        # Show SLA penalties row when SLA is enabled
        if sla_config.sla_enabled:
            data.append(sla_penalties_row)
        data.append(service_ratios)

        checks = (
            db.session.query(Service.name, Check.result)
            .join(Service)
            .join(Round)
            .filter(Round.number == last_round)
            .order_by(Service.name, Service.team_id)
            .all()
        )

        service_dict = defaultdict(list)

        for service, status in checks:
            service_dict[service].append(status)

        # Loop through dictionary to create datatables formatted list
        for k, v in service_dict.items():
            data.append([k] + v)
        return jsonify(data=data)
    else:
        return "{}"

1	from collections import defaultdict	1✔
2
3	from flask import jsonify	1✔
4	from flask_login import current_user	1✔
5	from sqlalchemy import desc	1✔
6	from sqlalchemy.sql import func	1✔
7
8	from scoring_engine.cache import cache	1✔
9	from scoring_engine.db import db	1✔
10	from scoring_engine.models.check import Check	1✔
11	from scoring_engine.models.round import Round	1✔
12	from scoring_engine.models.service import Service	1✔
13	from scoring_engine.models.setting import Setting	1✔
14	from scoring_engine.models.team import Team	1✔
15	from scoring_engine.sla import apply_dynamic_scoring_to_round, calculate_team_total_penalties, get_sla_config	1✔
16
17	from . import mod	1✔
18
19
20	def get_anonymize_mode():	1✔
21	"""
22	Determine how team names should be displayed.
23	Returns: (anonymize, show_both) tuple
24	- (True, False): Show only anonymous names (public/blue teams)
25	- (False, True): Show "RealName (Anonymous)" (white team when enabled)
26	- (False, False): Show only real names (setting disabled)
27	"""
28	setting = Setting.get_setting("anonymize_team_names")	1✔
29	anonymize_enabled = setting.value is True if setting else False	1✔
30
31	if not anonymize_enabled:	1✔
32	return (False, False)	1✔
33
34	# White team sees both names when anonymization is enabled
35	if current_user.is_authenticated and current_user.is_white_team:	×
36	return (False, True)	×
37
38	# Everyone else sees only anonymous names
39	return (True, False)	×
40
41
42	def calculate_ranks(score_dict):	1✔
43	"""
44	Calculate ranks for a dict of {id: score} with tie handling.
45	Returns dict of {id: rank} where ties get the same rank.
46	"""
47	if not score_dict:	1✔
48	return {}	×
49	sorted_items = sorted(score_dict.items(), key=lambda x: x[1], reverse=True)	1✔
50	ranks = {}	1✔
51	current_rank = 1	1✔
52	prev_score = None	1✔
53	for i, (item_id, score) in enumerate(sorted_items):	1✔
54	if prev_score is not None and score < prev_score:	1✔
55	current_rank = i + 1	1✔
56	ranks[item_id] = current_rank	1✔
57	prev_score = score	1✔
58	return ranks	1✔
59
60
61	@cache.memoize()	1✔
62	def _get_table_columns_cached(anonymize, show_both):	1✔
63	"""Internal cached function for table columns."""
64	blue_teams = Team.get_all_blue_teams()	×
65	team_name_map = Team.get_team_name_mapping(anonymize=anonymize, show_both=show_both)	×
66
67	columns = []	×
68	columns.append({"title": "", "data": "", "color": None})	×
69	for team in blue_teams:	×
70	display_name = team_name_map.get(team.id, team.name)	×
71	columns.append({"title": display_name, "data": display_name, "color": team.rgb_color})	×
72	return columns	×
73
74
75	@cache.memoize()	1✔
76	def _get_overview_data_cached(anonymize, show_both):	1✔
77	"""Internal cached function for overview data with host/port info."""
78	data = defaultdict(lambda: defaultdict(dict))	1✔
79	team_name_map = Team.get_team_name_mapping(anonymize=anonymize, show_both=show_both)	1✔
80
81	round_obj = db.session.query(Round).order_by(Round.number.desc()).first()	1✔
82	if round_obj:	1✔
83	checks = (	1✔
84	db.session.query(Check.service_id, Check.result)
85	.join(Round)
86	.filter(Round.number == round_obj.number)
87	.subquery()
88	)
89	res = (	1✔
90	db.session.query(Team.name, Service.name, Service.host, Service.port, checks.c.result)
91	.join(Team)
92	.filter(checks.c.service_id == Service.id)
93	.all()
94	)
95
96	for r in res:	1✔
97	team_name = r[0]	1✔
98	display_name = team_name_map.get(team_name, team_name)	1✔
99	data[display_name][r[1]] = {	1✔
100	"host": r[2],
101	"port": r[3],
102	"passing": r[4],
103	}
104
105	# Convert defaultdict to regular dict for JSON serialization
106	return {k: dict(v) for k, v in data.items()}	1✔
107
108
109	@mod.route("/api/overview/get_round_data")	1✔
110	@cache.memoize()	1✔
111	def overview_get_round_data():	1✔
112	round_obj = db.session.query(Round).order_by(Round.number.desc()).first()	×
113	if round_obj:	×
114	round_start = round_obj.local_round_start	×
115	number = round_obj.number	×
116	else:
117	round_start = ""	×
118	number = 0	×
119	engine_paused = Setting.get_setting("engine_paused").value	×
120	data = {"round_start": round_start, "number": number, "engine_paused": engine_paused}	×
121	return jsonify(data)	×
122
123
124	@mod.route("/api/overview/data")	1✔
125	def overview_data():	1✔
126	"""Get overview data with host/port info. Cached by user type."""
127	anonymize, show_both = get_anonymize_mode()	1✔
128	return jsonify(_get_overview_data_cached(anonymize, show_both))	1✔
129
130
131	@mod.route("/api/overview/get_columns")	1✔
132	def overview_get_columns():	1✔
133	"""Get table columns. Cached by user type."""
134	anonymize, show_both = get_anonymize_mode()	×
135	return jsonify(columns=_get_table_columns_cached(anonymize, show_both))	×
136
137
138	@mod.route("/api/overview/get_data")	1✔
139	@cache.memoize()	1✔
140	def overview_get_data():	1✔
141	"""Get overview table data. This endpoint returns positional data matching columns."""
142	data = []	1✔
143	blue_teams = db.session.query(Team).filter(Team.color == "Blue").order_by(Team.id).all()	1✔
144	blue_team_ids = [team.id for team in blue_teams]	1✔
145	blue_teams_dict = {team.id: team for team in blue_teams}	1✔
146	last_round = Round.get_last_round_num()	1✔
147
148	# Get SLA configuration
149	sla_config = get_sla_config()	1✔
150
151	current_scores = ["Current Score"]	1✔
152	current_places = ["Current Place"]	1✔
153	service_ratios = ["Up/Down Ratio"]	1✔
154	sla_penalties_row = ["SLA Penalties"]	1✔
155
156	num_up_services = dict(	1✔
157	db.session.query(
158	Service.team_id,
159	func.count(Service.team_id),
160	)
161	.join(Check)
162	.join(Round)
163	.filter(Check.result.is_(True))
164	.filter(Round.number == last_round)
165	.group_by(Service.team_id)
166	.all()
167	)
168
169	num_down_services = dict(	1✔
170	db.session.query(
171	Service.team_id,
172	func.count(Service.team_id),
173	)
174	.join(Check)
175	.join(Round)
176	.filter(Check.result.is_(False))
177	.filter(Round.number == last_round)
178	.group_by(Service.team_id)
179	.all()
180	)
181
182	if len(blue_team_ids) > 0:	1✔
183	# Calculate team scores with dynamic scoring multipliers
184	if sla_config.dynamic_enabled:	1✔
185	# Query scores per round for dynamic scoring
186	round_scores = (	1✔
187	db.session.query(
188	Service.team_id,
189	Check.round_id,
190	func.sum(Service.points).label("round_score"),
191	)
192	.join(Check)
193	.filter(Check.result.is_(True))
194	.group_by(Service.team_id, Check.round_id)
195	.all()
196	)
197
198	# Get round numbers for each round_id
199	rounds_map = {r.id: r.number for r in db.session.query(Round.id, Round.number).all()}	1✔
200
201	# Calculate totals with multipliers
202	team_scores = defaultdict(int)	1✔
203	for team_id, round_id, round_score in round_scores:	1✔
204	round_number = rounds_map.get(round_id, 0)	1✔
205	adjusted_score = apply_dynamic_scoring_to_round(round_number, round_score, sla_config)	1✔
206	team_scores[team_id] += adjusted_score	1✔
207	team_scores = dict(team_scores)	1✔
208	else:
209	# No dynamic scoring - use simple sum
210	team_scores = dict(	1✔
211	db.session.query(Service.team_id, func.sum(Service.points).label("score"))
212	.join(Check)
213	.filter(Check.result.is_(True))
214	.group_by(Service.team_id)
215	.order_by(desc("score"))
216	.all()
217	)
218
219	# Calculate adjusted scores with SLA penalties
220	adjusted_scores_dict = {}	1✔
221	penalties_dict = {}	1✔
222	for blue_team_id in blue_team_ids:	1✔
223	base_score = team_scores.get(blue_team_id, 0)	1✔
224	if sla_config.sla_enabled:	1✔
225	team = blue_teams_dict[blue_team_id]	×
226	penalty = calculate_team_total_penalties(team, sla_config)	×
227	penalties_dict[blue_team_id] = penalty	×
228	if sla_config.allow_negative:	×
229	adjusted_scores_dict[blue_team_id] = base_score - penalty	×
230	else:
231	adjusted_scores_dict[blue_team_id] = max(0, base_score - penalty)	×
232	else:
233	penalties_dict[blue_team_id] = 0	1✔
234	adjusted_scores_dict[blue_team_id] = base_score	1✔
235
236	# Use adjusted scores for ranking when SLA is enabled
237	scores_for_ranking = adjusted_scores_dict if sla_config.sla_enabled else team_scores	1✔
238
239	# Calculate ranks with tie handling
240	ranks_dict = calculate_ranks(scores_for_ranking)	1✔
241
242	for blue_team_id in blue_team_ids:	1✔
243	# Show adjusted score when SLA is enabled, base score otherwise
244	if sla_config.sla_enabled:	1✔
245	current_scores.append(str(adjusted_scores_dict.get(blue_team_id, 0)))	×
246	else:
247	current_scores.append(str(team_scores.get(blue_team_id, 0)))	1✔
248	current_places.append(str(ranks_dict.get(blue_team_id, 0)))	1✔
249	service_ratios.append(	1✔
250	'<span class="text-success">{0} <i class="bi bi-arrow-up"></i></span> / '
251	'<span class="text-danger">{1} <i class="bi bi-arrow-down"></i></span>'.format(
252	num_up_services.get(blue_team_id, 0),
253	num_down_services.get(blue_team_id, 0),
254	)
255	)
256	# Add penalty display (negative number if penalty exists)
257	penalty = penalties_dict.get(blue_team_id, 0)	1✔
258	if penalty > 0:	1✔
259	sla_penalties_row.append('<span class="text-danger">-{}</span>'.format(penalty))	×
260	else:
261	sla_penalties_row.append("0")	1✔
262
263	data.append(current_scores)	1✔
264	data.append(current_places)	1✔
265	# Show SLA penalties row when SLA is enabled
266	if sla_config.sla_enabled:	1✔
267	data.append(sla_penalties_row)	×
268	data.append(service_ratios)	1✔
269
270	checks = (	1✔
271	db.session.query(Service.name, Check.result)
272	.join(Service)
273	.join(Round)
274	.filter(Round.number == last_round)
275	.order_by(Service.name, Service.team_id)
276	.all()
277	)
278
279	service_dict = defaultdict(list)	1✔
280
281	for service, status in checks:	1✔
282	service_dict[service].append(status)	1✔
283
284	# Loop through dictionary to create datatables formatted list
285	for k, v in service_dict.items():	1✔
286	data.append([k] + v)	1✔
287	return jsonify(data=data)	1✔
288	else:
289	return "{}"	×

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