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

25.74

/scoring_engine/web/views/api/team.py

from collections import defaultdict

from flask import jsonify
from flask_login import current_user, login_required
from sqlalchemy import desc, func
from sqlalchemy.orm import subqueryload

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.team import Team
from scoring_engine.sla import get_sla_config

from . import make_cache_key, mod


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.
    E.g., scores [100, 90, 90, 80] -> ranks [1, 2, 2, 4]
    """
    if not score_dict:
        return {}

    # Sort by score descending
    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  # Skip ranks for ties
        ranks[item_id] = current_rank
        prev_score = score

    return ranks


@mod.route("/api/team/<team_id>/stats")
@login_required
@cache.cached(make_cache_key=make_cache_key)
def services_get_team_data(team_id):
    team = db.session.get(Team, team_id)
    if team is None or not current_user.team == team or not current_user.is_blue_team:
        return {"status": "Unauthorized"}, 403

    data = {"place": str(team.place), "current_score": str(team.current_score)}
    return jsonify(data)


@mod.route("/api/team/<team_id>/services")
@login_required
@cache.cached(make_cache_key=make_cache_key)
def api_services(team_id):
    team = db.session.get(Team, team_id)
    if team is None or not current_user.team == team or not current_user.is_blue_team:
        return {"status": "Unauthorized"}, 403

    data = []

    # Get SLA config for dynamic scoring
    sla_config = get_sla_config()
    dynamic_enabled = sla_config.dynamic_enabled

    if dynamic_enabled:
        # Pre-fetch config values for performance
        early_rounds = sla_config.early_rounds
        early_multiplier = sla_config.early_multiplier
        late_start = sla_config.late_start_round
        late_multiplier = sla_config.late_multiplier

        # Query for dynamic scoring: get service scores with round numbers via JOIN
        # This is more efficient than querying all rounds separately
        service_check_data = (
            db.session.query(
                Service.team_id,
                Service.name,
                Service.points,
                Round.number,
                Check.result,
            )
            .select_from(Check)
            .join(Service, Check.service_id == Service.id)
            .join(Round, Check.round_id == Round.id)
            .all()
        )

        # Calculate dynamic scores per service per team
        service_dict = defaultdict(lambda: defaultdict(int))
        service_max_dict = defaultdict(lambda: defaultdict(int))

        for svc_team_id, name, points, round_number, result in service_check_data:
            # Inline multiplier calculation for performance
            if round_number <= early_rounds:
                multiplier = early_multiplier
            elif round_number >= late_start:
                multiplier = late_multiplier
            else:
                multiplier = 1.0

            dynamic_points = int(points * multiplier)
            service_max_dict[name][svc_team_id] += dynamic_points
            if result:
                service_dict[name][svc_team_id] += dynamic_points
    else:
        # Original non-dynamic scoring logic
        service_scores = (
            db.session.query(Service.team_id, Service.name, func.sum(Service.points).label("score"))
            .join(Check)
            .filter(Check.result.is_(True))
            .group_by(Service.team_id, Service.name)
            .order_by(Service.name, desc("score"))
            .all()
        )

        service_dict = defaultdict(lambda: defaultdict(int))
        for svc_team_id, name, points in service_scores:
            service_dict[name][svc_team_id] = points

        service_max_dict = None  # Will calculate per-service below

    # Calculate ranks based on scores
    service_ranks = defaultdict(lambda: defaultdict(int))
    for service_name in service_dict.keys():
        service_ranks[service_name] = calculate_ranks(service_dict[service_name])

    services = (
        db.session.query(Service)
        .options(subqueryload(Service.checks))
        .options(subqueryload(Service.team))
        .filter(Service.team_id == team.id)
        .order_by(Service.id)
        .all()
    )

    for service in services:
        score_earned = service_dict[service.name].get(service.team_id, 0)

        if dynamic_enabled and service_max_dict:
            max_score = service_max_dict[service.name].get(service.team_id, 0)
        else:
            max_score = len(service.checks) * service.points

        percent_earned = "{:.1%}".format(score_earned / max_score if max_score != 0 else 0)

        if not service.checks:
            check = "Undetermined"
        else:
            if service.last_check_result():
                check = "UP"
            else:
                check = "DOWN"
        data.append(
            dict(
                service_id=str(service.id),
                service_name=str(service.name),
                host=str(service.host),
                port=str(service.port),
                check=str(check),
                rank=str(service_ranks[service.name].get(service.team_id, 1)),
                score_earned=str(score_earned),
                max_score=str(max_score),
                percent_earned=percent_earned,
                pts_per_check=str(service.points),
                last_ten_checks=[check.result for check in service.last_ten_checks[::-1]],
            )
        )
    return jsonify(data=data)


@mod.route("/api/team/<team_id>/services/status")
@login_required
@cache.cached(make_cache_key=make_cache_key)
def team_services_status(team_id):
    team = db.session.get(Team, team_id)
    if team is None or not current_user.team == team or not current_user.is_blue_team:
        return {"status": "Unauthorized"}, 403

    data = {}

    round_obj = db.session.query(Round.id).order_by(Round.number.desc()).first()

    # We have no round data, the first round probably hasn't started yet
    if not round_obj:
        return data

    round_id = round_obj[0]

    checks = (
        db.session.query(
            Service.name,
            Check.service_id,
            Check.result,
            Service.check_name,
            Service.host,
        )
        .select_from(Check)
        .join(Service)
        .filter(Service.team_id == team_id)
        .filter(Check.round_id == round_id)
        .order_by(Service.name)
        .all()
    )

    for service_name, service_id, check_result, check_name, host in checks:
        data[service_name] = {
            "id": str(service_id),
            "result": str(check_result),
            "check_name": str(check_name),
            "host": str(host),
        }
    return jsonify(data)

1	from collections import defaultdict	1✔
2
3	from flask import jsonify	1✔
4	from flask_login import current_user, login_required	1✔
5	from sqlalchemy import desc, func	1✔
6	from sqlalchemy.orm import subqueryload	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.team import Team	1✔
14	from scoring_engine.sla import get_sla_config	1✔
15
16	from . import make_cache_key, mod	1✔
17
18
19	def calculate_ranks(score_dict):	1✔
20	"""
21	Calculate ranks for a dict of {id: score} with tie handling.
22
23	Returns dict of {id: rank} where ties get the same rank.
24	E.g., scores [100, 90, 90, 80] -> ranks [1, 2, 2, 4]
25	"""
26	if not score_dict:	×
27	return {}	×
28
29	# Sort by score descending
30	sorted_items = sorted(score_dict.items(), key=lambda x: x[1], reverse=True)	×
31
32	ranks = {}	×
33	current_rank = 1	×
34	prev_score = None	×
35
36	for i, (item_id, score) in enumerate(sorted_items):	×
37	if prev_score is not None and score < prev_score:	×
38	current_rank = i + 1 # Skip ranks for ties	×
39	ranks[item_id] = current_rank	×
40	prev_score = score	×
41
42	return ranks	×
43
44
45	@mod.route("/api/team/<team_id>/stats")	1✔
46	@login_required	1✔
47	@cache.cached(make_cache_key=make_cache_key)	1✔
48	def services_get_team_data(team_id):	1✔
49	team = db.session.get(Team, team_id)	×
50	if team is None or not current_user.team == team or not current_user.is_blue_team:	×
51	return {"status": "Unauthorized"}, 403	×
52
53	data = {"place": str(team.place), "current_score": str(team.current_score)}	×
54	return jsonify(data)	×
55
56
57	@mod.route("/api/team/<team_id>/services")	1✔
58	@login_required	1✔
59	@cache.cached(make_cache_key=make_cache_key)	1✔
60	def api_services(team_id):	1✔
61	team = db.session.get(Team, team_id)	×
62	if team is None or not current_user.team == team or not current_user.is_blue_team:	×
63	return {"status": "Unauthorized"}, 403	×
64
65	data = []	×
66
67	# Get SLA config for dynamic scoring
68	sla_config = get_sla_config()	×
69	dynamic_enabled = sla_config.dynamic_enabled	×
70
71	if dynamic_enabled:	×
72	# Pre-fetch config values for performance
73	early_rounds = sla_config.early_rounds	×
74	early_multiplier = sla_config.early_multiplier	×
75	late_start = sla_config.late_start_round	×
76	late_multiplier = sla_config.late_multiplier	×
77
78	# Query for dynamic scoring: get service scores with round numbers via JOIN
79	# This is more efficient than querying all rounds separately
80	service_check_data = (	×
81	db.session.query(
82	Service.team_id,
83	Service.name,
84	Service.points,
85	Round.number,
86	Check.result,
87	)
88	.select_from(Check)
89	.join(Service, Check.service_id == Service.id)
90	.join(Round, Check.round_id == Round.id)
91	.all()
92	)
93
94	# Calculate dynamic scores per service per team
95	service_dict = defaultdict(lambda: defaultdict(int))	×
96	service_max_dict = defaultdict(lambda: defaultdict(int))	×
97
98	for svc_team_id, name, points, round_number, result in service_check_data:	×
99	# Inline multiplier calculation for performance
100	if round_number <= early_rounds:	×
101	multiplier = early_multiplier	×
102	elif round_number >= late_start:	×
103	multiplier = late_multiplier	×
104	else:
105	multiplier = 1.0	×
106
107	dynamic_points = int(points * multiplier)	×
108	service_max_dict[name][svc_team_id] += dynamic_points	×
109	if result:	×
110	service_dict[name][svc_team_id] += dynamic_points	×
111	else:
112	# Original non-dynamic scoring logic
113	service_scores = (	×
114	db.session.query(Service.team_id, Service.name, func.sum(Service.points).label("score"))
115	.join(Check)
116	.filter(Check.result.is_(True))
117	.group_by(Service.team_id, Service.name)
118	.order_by(Service.name, desc("score"))
119	.all()
120	)
121
122	service_dict = defaultdict(lambda: defaultdict(int))	×
123	for svc_team_id, name, points in service_scores:	×
124	service_dict[name][svc_team_id] = points	×
125
126	service_max_dict = None # Will calculate per-service below	×
127
128	# Calculate ranks based on scores
129	service_ranks = defaultdict(lambda: defaultdict(int))	×
130	for service_name in service_dict.keys():	×
131	service_ranks[service_name] = calculate_ranks(service_dict[service_name])	×
132
133	services = (	×
134	db.session.query(Service)
135	.options(subqueryload(Service.checks))
136	.options(subqueryload(Service.team))
137	.filter(Service.team_id == team.id)
138	.order_by(Service.id)
139	.all()
140	)
141
142	for service in services:	×
143	score_earned = service_dict[service.name].get(service.team_id, 0)	×
144
145	if dynamic_enabled and service_max_dict:	×
146	max_score = service_max_dict[service.name].get(service.team_id, 0)	×
147	else:
148	max_score = len(service.checks) * service.points	×
149
150	percent_earned = "{:.1%}".format(score_earned / max_score if max_score != 0 else 0)	×
151
152	if not service.checks:	×
153	check = "Undetermined"	×
154	else:
155	if service.last_check_result():	×
156	check = "UP"	×
157	else:
158	check = "DOWN"	×
159	data.append(	×
160	dict(
161	service_id=str(service.id),
162	service_name=str(service.name),
163	host=str(service.host),
164	port=str(service.port),
165	check=str(check),
166	rank=str(service_ranks[service.name].get(service.team_id, 1)),
167	score_earned=str(score_earned),
168	max_score=str(max_score),
169	percent_earned=percent_earned,
170	pts_per_check=str(service.points),
171	last_ten_checks=[check.result for check in service.last_ten_checks[::-1]],
172	)
173	)
174	return jsonify(data=data)	×
175
176
177	@mod.route("/api/team/<team_id>/services/status")	1✔
178	@login_required	1✔
179	@cache.cached(make_cache_key=make_cache_key)	1✔
180	def team_services_status(team_id):	1✔
181	team = db.session.get(Team, team_id)	×
182	if team is None or not current_user.team == team or not current_user.is_blue_team:	×
183	return {"status": "Unauthorized"}, 403	×
184
185	data = {}	×
186
187	round_obj = db.session.query(Round.id).order_by(Round.number.desc()).first()	×
188
189	# We have no round data, the first round probably hasn't started yet
190	if not round_obj:	×
191	return data	×
192
193	round_id = round_obj[0]	×
194
195	checks = (	×
196	db.session.query(
197	Service.name,
198	Check.service_id,
199	Check.result,
200	Service.check_name,
201	Service.host,
202	)
203	.select_from(Check)
204	.join(Service)
205	.filter(Service.team_id == team_id)
206	.filter(Check.round_id == round_id)
207	.order_by(Service.name)
208	.all()
209	)
210
211	for service_name, service_id, check_result, check_name, host in checks:	×
212	data[service_name] = {	×
213	"id": str(service_id),
214	"result": str(check_result),
215	"check_name": str(check_name),
216	"host": str(host),
217	}
218	return jsonify(data)	×

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