diff --git a/.env_sample b/.env_sample index bcea691..95b0f9b 100644 --- a/.env_sample +++ b/.env_sample @@ -1,3 +1,9 @@ ENVIRONMENT="Development" FLASK_SECRET_KEY="156a7fbb77c17708e41f044ea5131be2ad592a7deb21866bcb63d7801d2fa13e" FLASK_DATABASE=compy.sqlite +# Password for the admin interface (http://localhost:5000/admin). +# For deployments, remove FLASK_ADMIN_PASSWORD and set FLASK_ADMIN_PASSWORD_HASH +# instead. Generate a hash with: +# python3 -c "from werkzeug.security import generate_password_hash; import getpass; print(generate_password_hash(getpass.getpass()))" +FLASK_ADMIN_PASSWORD="compy-admin" +#FLASK_ADMIN_PASSWORD_HASH="" diff --git a/Readme.md b/Readme.md index 9c1fff1..41ee898 100644 --- a/Readme.md +++ b/Readme.md @@ -57,8 +57,12 @@ Install Weasyprint https://doc.courtbouillon.org/weasyprint/stable/first_steps.h - Execute `git clone https://github.com/Azrael3000/Compy.git` - Switch to the new folder: `cd Compy` - Set up the environmen: `cp .env_sample .env` - - For deployments you MUST edit the .env file and provide a new secret. A new one can be generated e.g. by running - `python3 -c "import secrets; print(secrets.token_hex())"` + - For deployments you MUST edit the .env file: + - Provide a new secret (used to sign the admin session cookie). A new one can be generated e.g. by running + `python3 -c "import secrets; print(secrets.token_hex())"` + - Set your own admin password. Either change `FLASK_ADMIN_PASSWORD`, or (recommended) remove it and set + `FLASK_ADMIN_PASSWORD_HASH` to a password hash generated by running + `python3 -c "from werkzeug.security import generate_password_hash; import getpass; print(generate_password_hash(getpass.getpass()))"` - Start a virtual environment and install required packages: - Linux: `source venv/bin/activate && pip install -r requirements.txt` - Set up the database and run the server: `python3 compy.py --init_db` @@ -73,8 +77,9 @@ Install Weasyprint https://doc.courtbouillon.org/weasyprint/stable/first_steps.h - Linux: `python3 compy.py` - Windows: `python3.exe compy.py` - Navigate your browser to `localhost:5000` - - The admin interface is at `localhost:5000/admin?auth=XXXXXX` where `XXXXXX` are the first 6 - characters of your `FLASK_SECRET_KEY` from `.env` + - The admin interface is at `localhost:5000/admin`. It asks for the admin password configured in + `.env` (`FLASK_ADMIN_PASSWORD` or `FLASK_ADMIN_PASSWORD_HASH`); a login is valid for 12 hours + or until you press "Logout" ## Test data diff --git a/compy_concurrency_test.py b/compy_concurrency_test.py index ad894fa..bb84286 100644 --- a/compy_concurrency_test.py +++ b/compy_concurrency_test.py @@ -21,7 +21,7 @@ class TestConcurrentPages(compy_testing.CompyServerTestCase): @classmethod def setUpClass(cls): super().setUpClass() - session = requests.Session() + session = cls.adminSession() # name the default competition and upload the excel file response = session.post(cls.base_url + "/competition", @@ -119,18 +119,20 @@ def registrationWrites(self, session, round_index): def testConcurrentPagesDoNotInterfere(self): page_simulations = [ - ("admin1", self.adminTabCompOne), - ("admin2", self.adminTabCompTwo), - ("clock", self.clockDisplay), - ("judge", self.judgePhone), - ("results", self.publicResultsPage), - ("registration", self.registrationWrites), + ("admin1", self.adminTabCompOne, True), + ("admin2", self.adminTabCompTwo, True), + ("clock", self.clockDisplay, False), + ("judge", self.judgePhone, False), + ("results", self.publicResultsPage, False), + ("registration", self.registrationWrites, True), ] failures = [] stop_event = threading.Event() - def run_page(page_name, request_round): - page_session = requests.Session() + def run_page(page_name, request_round, is_admin_page): + # admin pages carry a session cookie, public pages must work + # without any authentication + page_session = self.adminSession() if is_admin_page else requests.Session() for round_index in range(N_ROUNDS): if stop_event.is_set(): return @@ -151,7 +153,7 @@ def run_page(page_name, request_round): self.assertEqual(failures, []) # after the storm: comp 1 must be fully intact - session = requests.Session() + session = self.adminSession() response = session.post(self.base_url + "/load_comp", json={"comp_id": self.comp_one_id}) self.assertEqual(response.json()["comp_name"], "Comp One") self.assertEqual(len(response.json()["athletes"]), 30) @@ -175,9 +177,57 @@ def testForgedJudgeHashIsRejected(self): "block": self.first_block, "lane": "1"}) self.assertEqual(response.status_code, 404) # ...and it must not have switched or broken anything + response = self.adminSession().get(self.base_url + "/athletes", + params={"comp_id": self.comp_one_id}) + self.assertEqual(len(response.json()["athletes"]), 30) + + def testAdminEndpointsRequireLogin(self): + # without a session cookie all admin endpoints must refuse to act response = requests.get(self.base_url + "/athletes", params={"comp_id": self.comp_one_id}) - self.assertEqual(len(response.json()["athletes"]), 30) + self.assertEqual(response.status_code, 401) + response = requests.post(self.base_url + "/competition", + json={"comp_name": "Hacked", "overwrite": True, + "comp_id": self.comp_one_id}) + self.assertEqual(response.status_code, 401) + response = requests.delete(self.base_url + "/competition", + json={"comp_id": self.comp_one_id}) + self.assertEqual(response.status_code, 401) + # the admin page itself redirects to the login form + response = requests.get(self.base_url + "/admin", allow_redirects=False) + self.assertEqual(response.status_code, 302) + self.assertTrue(response.headers["Location"].endswith("/admin/login")) + # ...and nothing was changed by the rejected requests + response = self.adminSession().post(self.base_url + "/load_comp", + json={"comp_id": self.comp_one_id}) + self.assertEqual(response.json()["comp_name"], "Comp One") + + def testWrongPasswordIsRejected(self): + session = requests.Session() + response = session.post(self.base_url + "/admin/login", + data={"password": "not-the-password"}) + self.assertEqual(response.status_code, 401) + response = session.get(self.base_url + "/athletes", + params={"comp_id": self.comp_one_id}) + self.assertEqual(response.status_code, 401) + + def testJudgeCanSaveResultWithoutAdminSession(self): + # a judge phone is not logged in as admin; the judge hash from the + # QR code must be enough to save a result, a forged hash must not be + response = requests.get(self.base_url + "/judge/athletes", + params={"comp_id": self.comp_one_id, "judge_id": self.judge_id, + "judge_hash": self.judge_hash, "day": self.first_day, + "block": self.first_block, "lane": "1"}) + start_id = response.json()["lane_list"][0]["s_id"] + result = {"comp_id": self.comp_one_id, "judge_id": self.judge_id, + "id": start_id, "rp": "", "penalty": 0, "card": "WHITE", + "remarks": "", "judge_remarks": ""} + response = requests.put(self.base_url + "/result", + json=result | {"judge_hash": "deadbeef"}) + self.assertEqual(response.status_code, 401) + response = requests.put(self.base_url + "/result", + json=result | {"judge_hash": self.judge_hash}) + self.assertEqual(response.status_code, 200) if __name__ == '__main__': diff --git a/compy_flask.py b/compy_flask.py index d79f548..df8de86 100644 --- a/compy_flask.py +++ b/compy_flask.py @@ -24,11 +24,16 @@ # # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ +import hmac import logging +import time +from datetime import timedelta +from functools import wraps from compy_data import CompyData from compy_config import CompyConfig -from flask import Flask, render_template, request, send_file, Response, make_response, current_app +from flask import Flask, render_template, request, send_file, Response, make_response, session, redirect, url_for from os import path, mkdir +from werkzeug.security import check_password_hash from werkzeug.utils import secure_filename from werkzeug.routing import IntegerConverter try: @@ -58,23 +63,63 @@ def __init__(self, app, db, start_flask): app.config['UPLOAD_FOLDER'] = self.config_.upload_folder app.url_map.converters['signed_int'] = self.SignedIntConverter + # admin sessions are stored in a cookie signed with SECRET_KEY; + # the cookie is not readable by page javascript and not sent on + # cross-site requests (basic CSRF protection) + app.config['SESSION_COOKIE_HTTPONLY'] = True + app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' + # one login lasts a full competition day + app.permanent_session_lifetime = timedelta(hours=12) + + if not app.config.get('ADMIN_PASSWORD_HASH') and not app.config.get('ADMIN_PASSWORD'): + logging.error("Neither FLASK_ADMIN_PASSWORD_HASH nor FLASK_ADMIN_PASSWORD is set " + "in the .env file; logging in to the admin interface is not possible") + + def admin_required(f): + """Only allow the request if this browser has an admin session. + + The admin page itself redirects to the login form, all other + (api) endpoints return 401 so the frontend can react. + """ + @wraps(f) + def wrapper(*args, **kwargs): + if session.get('is_admin'): + return f(*args, **kwargs) + if request.method == 'GET' and request.path == '/admin': + return redirect(url_for('login')) + return self.unauthorized() + return wrapper + @app.route('/admin', methods=['GET']) + @admin_required def admin(): return self.admin() + @app.route('/admin/login', methods=['GET', 'POST']) + def login(): + return self.login() + + @app.route('/admin/logout', methods=['GET']) + def logout(): + return self.logout() + @app.route('/upload_file', methods=['POST']) + @admin_required def uploadFile(): return self.uploadFile() @app.route('/store_results', methods=['POST']) + @admin_required def storeResults(): return self.storeResults() @app.route('/upload_sponsor_img', methods=['POST']) + @admin_required def uploadSponsorImg(): return self.uploadSponsorImg() @app.route('/competition', methods=['POST', 'DELETE']) + @admin_required def changeCompName(): if request.method == 'POST': return self.changeCompName() @@ -82,18 +127,22 @@ def changeCompName(): return self.deleteComp() @app.route('/change_special_ranking_name', methods=['POST']) + @admin_required def changeSpecialRankingName(): return self.changeSpecialRankingName() @app.route('/change_registration', methods=['POST']) + @admin_required def changeRegistration(): return self.changeRegistration() @app.route('/load_comp', methods=['POST']) + @admin_required def loadComp(): return self.loadComp() @app.route('/start_list', methods=['GET', 'PUT']) + @admin_required def startList(): if request.method == 'GET': return self.startList() @@ -101,45 +150,59 @@ def startList(): return self.updateStartList() @app.route('/start_list_pdf', methods=['GET']) + @admin_required def startListPDF(): return self.startListPDF() @app.route('/breaks', methods=['GET']) + @admin_required def breaks(): return self.breaks() @app.route('/lane_list', methods=['GET']) + @admin_required def laneList(): return self.laneList() @app.route('/lane_list_pdf', methods=['GET']) + @admin_required def laneListPDF(): return self.laneListPDF() @app.route('/result', methods=['GET', 'PUT']) def result(): if request.method == 'GET': + # admin page only + if not session.get('is_admin'): + return self.unauthorized() return self.result(False) elif request.method == 'PUT': + # used by the admin page and by judge phones; + # updateResult checks the admin session or the judge hash return self.updateResult() @app.route('/result_pdf', methods=['GET']) + @admin_required def resultPDF(): return self.result(True) @app.route('/change_lane_style', methods=['POST']) + @admin_required def changeLaneStyle(): return self.changeLaneStyle() @app.route('/change_comp_type', methods=['POST']) + @admin_required def changeCompType(): return self.changeCompType() @app.route('/change_selected_country', methods=['POST']) + @admin_required def changeSelectedCountry(): return self.changeSelectedCountry() @app.route('/judge', methods=['DELETE', 'POST']) + @admin_required def judge(): if request.method == 'DELETE': return self.deleteJudge() @@ -147,14 +210,17 @@ def judge(): return self.addJudge() @app.route('/judge/qr_code', methods=['GET']) + @admin_required def judgeQrCode(): return self.getJudgeQrCode() @app.route('/judges', methods=['GET']) + @admin_required def judges(): return self.getJudges() @app.route('/athlete', methods=['DELETE', 'POST']) + @admin_required def athlete(): if request.method == 'DELETE': return self.deleteAthlete() @@ -162,10 +228,12 @@ def athlete(): return self.addAthlete() @app.route('/athletes', methods=['GET']) + @admin_required def athletes(): return self.getAthletes() @app.route('/national_records', methods=['GET']) + @admin_required def nationalRecords(): return self.nationalRecords() @@ -187,10 +255,12 @@ def judgeAthleteResult(): return self.getJudgeAthleteResult() @app.route('/disciplines/', methods=['GET']) + @admin_required def disciplines(federation): return self.disciplines(federation) @app.route('/block', methods=['POST', 'UPDATE', 'DELETE']) + @admin_required def block(): if request.method == 'POST': return self.modifyBlock(True) @@ -204,6 +274,7 @@ def clock(comp_id, current, offset): return self.getClock(comp_id, current, offset) @app.route('/publish_results', methods=['UPDATE']) + @admin_required def publish_results(): return self.updatePublishResults() @@ -609,6 +680,10 @@ def updateResult(self): comp = self.getData(request) if comp is None: return self.badRequest("Failed to load competition") + # results are entered by the admin page (session cookie) or by a + # judge phone (judge id + hash in the request body) + if not session.get('is_admin') and not self.isValidJudge(request, comp): + return self.unauthorized() content, status = self.handleRequest(request, ['id', 'rp', 'penalty', 'card', 'remarks', 'judge_remarks'], CompyData.updateResult, comp) if status != 200: logging.debug("Failed to set result") @@ -926,11 +1001,49 @@ def getClock(self, comp_id, current, offset): "offset": offset} return render_template('clock.html', **content) + def checkAdminPassword(self, password): + """Compare a login attempt against the configured admin password. + + FLASK_ADMIN_PASSWORD_HASH (a werkzeug password hash, recommended + for deployments) takes precedence over the plain text + FLASK_ADMIN_PASSWORD. Both comparisons are constant time. If + neither is configured, logging in is not possible. + """ + pw_hash = self.app_.config.get('ADMIN_PASSWORD_HASH') + if pw_hash: + return check_password_hash(pw_hash, password) + pw = self.app_.config.get('ADMIN_PASSWORD') + if pw: + return hmac.compare_digest(pw.encode('utf-8'), password.encode('utf-8')) + return False + + def login(self): + if session.get('is_admin'): + return redirect(url_for('admin')) + error = None + if request.method == 'POST': + password = request.form.get('password', '') + if self.checkAdminPassword(password): + session.clear() + session['is_admin'] = True + session.permanent = True + logging.info("Admin login from " + str(request.remote_addr)) + return redirect(url_for('admin')) + # throttle brute force attempts + time.sleep(1) + logging.warning("Failed admin login attempt from " + str(request.remote_addr)) + error = "Wrong password" + content = {"version": self.version(), "error": error} + return make_response(render_template('login.html', **content), 401 if error else 200) + + def logout(self): + session.clear() + return redirect(url_for('login')) + + def unauthorized(self): + return {"status": "error", "error_msg": "Authentication required"}, 401 + def admin(self): - auth = request.args.get('auth') - if auth != current_app.config["SECRET_KEY"][:6]: - content = {"version": self.version()} - return make_response(render_template('404.html', **content), 404) all_countries = country_converter.CountryConverter().data["IOC"].dropna().to_list() # the admin frontend loads competition 1 after the page is ready, so # pre-fill the name field with that competition diff --git a/compy_result_entry_test.py b/compy_result_entry_test.py new file mode 100644 index 0000000..cd7b7e0 --- /dev/null +++ b/compy_result_entry_test.py @@ -0,0 +1,309 @@ +"""Tests for entering results through the judging interface. + +Covers the full path a result takes from the judge (or the admin page) +into the database: white cards, yellow cards with manual and automatic +under-AP penalties, red cards with DQ remarks, DNS, and rejection of +invalid input. The data-layer tests check CompyData.updateResult and the +exact values that end up in the start table; the HTTP tests drive the +same flow through the flask endpoints a judge phone uses. +""" +import unittest + +import requests + +import athlete +import compy_testing +import compy_utilities as u + + +class ResultEntryCompetition: + """Mixin that builds a small hand-made competition for result entry. + + One CWT block and one STA block, each athlete has an AP but no result + yet, so every test can enter a result for its own athlete without + interfering with the other tests. + """ + + CWT_ATHLETES = {"Wht": 50, "Yel": 50, "Uap": 50, "Wbp": 50, "Red": 60, + "Dns": 40, "Inv": 50, "Pen": 50, "Ovr": 50, "Neg": 50, + "Ada": 50, "Bel": 50, "Cyd": 55, "Dee": 50, "Eva": 50} + STA_ATHLETES = {"Sta": 120} + + @classmethod + def buildCompetition(cls, comp_name): + import compy_data + with cls.app.app_context(): + data = compy_data.CompyData(cls.db, cls.app) + data.changeName(comp_name, False) + cls.comp_id = data.id_ + cls.cwt_block = cls.db.insert( + "INSERT INTO block (competition_id, day, disciplines) VALUES (?, ?, ?)", + (cls.comp_id, 20260101, data.disciplineListToInt(["CWT"]))) + cls.sta_block = cls.db.insert( + "INSERT INTO block (competition_id, day, disciplines) VALUES (?, ?, ?)", + (cls.comp_id, 20260102, data.disciplineListToInt(["STA"]))) + cls.start_ids = {} + ot = 1000 + for first_name, ap in cls.CWT_ATHLETES.items(): + cls.addStart(first_name, "CWT", cls.cwt_block, ap, ot) + ot += 5 + for first_name, ap in cls.STA_ATHLETES.items(): + cls.addStart(first_name, "STA", cls.sta_block, ap, ot) + ot += 5 + + @classmethod + def addStart(cls, first_name, discipline, block_id, ap, ot): + new_athlete = athlete.Athlete.fromArgs( + "id-" + first_name, first_name, "Diver", "F", "AUT", "", cls.db) + new_athlete.associateWithComp(cls.comp_id) + cls.start_ids[first_name] = cls.db.insert( + '''INSERT INTO start + (competition_athlete_id, discipline, lane, OT, AP, block) + VALUES (?, ?, ?, ?, ?, ?)''', + (new_athlete.comp_athlete_id_, discipline, 1, ot, ap, block_id)) + + +class TestResultEntry(ResultEntryCompetition, compy_testing.CompyDataTestCase): + """CompyData.updateResult: what exactly lands in the start table.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.buildCompetition("Result Entry Test Open") + + def setUp(self): + super().setUp() + self.data = self.newData(self.comp_id) + + def fetchStart(self, first_name): + row = self.db.execute( + "SELECT rp, penalty, card, remarks, judge_remarks FROM start WHERE id == ?", + self.start_ids[first_name]) + self.assertIsNotNone(row) + return {"rp": row[0][0], "penalty": row[0][1], "card": row[0][2], + "remarks": row[0][3], "judge_remarks": row[0][4]} + + def testWhiteCardResultIsStored(self): + ret, _ = self.data.updateResult( + self.start_ids["Wht"], "52", 0, "WHITE", "OK", "clean dive") + self.assertEqual(ret, 0) + start = self.fetchStart("Wht") + self.assertEqual(float(start["rp"]), 52.) + self.assertEqual(start["penalty"], 0.) + self.assertEqual(start["card"], "WHITE") + self.assertEqual(start["remarks"], "OK") + self.assertEqual(start["judge_remarks"], "clean dive") + + def testYellowCardManualPenaltyIsStored(self): + # AP reached, so only the manual penalty (e.g. early start) counts + ret, _ = self.data.updateResult( + self.start_ids["Yel"], "50", "2", "YELLOW", "EARLYSTART", "") + self.assertEqual(ret, 0) + start = self.fetchStart("Yel") + self.assertEqual(start["card"], "YELLOW") + self.assertEqual(start["penalty"], 2.) + + def testYellowCardUnderApPenaltyIsAddedAutomatically(self): + # AP 50, RP 45: 5 penalty points for the missing meters on top of + # the manual penalty of 1 + ret, _ = self.data.updateResult( + self.start_ids["Uap"], "45", 1., "YELLOW", "UNDER AP", "") + self.assertEqual(ret, 0) + self.assertEqual(self.fetchStart("Uap")["penalty"], 6.) + + def testWhiteCardGetsNoUnderApPenalty(self): + ret, _ = self.data.updateResult( + self.start_ids["Wbp"], "45", 0, "WHITE", "OK", "") + self.assertEqual(ret, 0) + self.assertEqual(self.fetchStart("Wbp")["penalty"], 0.) + + def testRedCardDqIsStoredAndScoresZero(self): + ret, _ = self.data.updateResult( + self.start_ids["Red"], "60", 0, "RED", "DQBO-SURFACE", "BO at surface") + self.assertEqual(ret, 0) + start = self.fetchStart("Red") + self.assertEqual(start["card"], "RED") + self.assertEqual(start["remarks"], "DQBO-SURFACE") + # a disqualified athlete scores zero points and gets no rank + ret, content = self.data.getResult("CWT", "F", "International") + self.assertEqual(ret, 0) + red = [r for r in content["results"] if r["Name"] == "Red Diver"][0] + self.assertEqual(red["Points"], "0.00") + self.assertEqual(red["Rank"], "") + + def testDnsIsStored(self): + ret, _ = self.data.updateResult( + self.start_ids["Dns"], "", 0, "RED", "DNS", "") + self.assertEqual(ret, 0) + start = self.fetchStart("Dns") + self.assertEqual(start["remarks"], "DNS") + ret, content = self.data.getResult("CWT", "F", "International") + dns = [r for r in content["results"] if r["Name"] == "Dns Diver"][0] + self.assertEqual(dns["Points"], "0.00") + self.assertEqual(dns["Remarks"], "DNS") + self.assertEqual(dns["Rank"], "") + + def testInvalidCardIsRejectedWithoutStoring(self): + ret, _ = self.data.updateResult( + self.start_ids["Inv"], "50", 0, "PURPLE", "", "") + self.assertEqual(ret, 1) + self.assertIsNone(self.fetchStart("Inv")["rp"]) + self.assertIsNone(self.fetchStart("Inv")["card"]) + + def testMissingCardIsRejected(self): + ret, _ = self.data.updateResult( + self.start_ids["Inv"], "50", 0, None, "", "") + self.assertEqual(ret, 1) + + def testNonNumericPenaltyBecomesZero(self): + ret, _ = self.data.updateResult( + self.start_ids["Pen"], "50", "abc", "WHITE", "OK", "") + self.assertEqual(ret, 0) + self.assertEqual(self.fetchStart("Pen")["penalty"], 0.) + + def testNegativeRpBecomesZero(self): + ret, _ = self.data.updateResult( + self.start_ids["Neg"], "-5", 0, "WHITE", "OK", "") + self.assertEqual(ret, 0) + self.assertEqual(float(self.fetchStart("Neg")["rp"]), 0.) + + def testStaPerformanceIsStoredInSeconds(self): + ret, _ = self.data.updateResult( + self.start_ids["Sta"], "2:05", 0, "WHITE", "OK", "") + self.assertEqual(ret, 0) + self.assertEqual(float(self.fetchStart("Sta")["rp"]), 125.) + # ...and is converted back to a time for display + ret, content = self.data.getAthleteResult(self.start_ids["Sta"]) + self.assertEqual(ret, 0) + self.assertEqual(content["RP"], "2:05") + + def testUnknownStartIdIsRejected(self): + ret, _ = self.data.updateResult(999999, "50", 0, "WHITE", "", "") + self.assertEqual(ret, 1) + + def testResultCanBeCorrected(self): + # the judge first enters a white card, then corrects it to a red + self.data.updateResult( + self.start_ids["Ovr"], "52", 0, "WHITE", "OK", "") + ret, _ = self.data.updateResult( + self.start_ids["Ovr"], "52", 0, "RED", "DQLATESTART", "late start") + self.assertEqual(ret, 0) + start = self.fetchStart("Ovr") + self.assertEqual(start["card"], "RED") + self.assertEqual(start["remarks"], "DQLATESTART") + self.assertEqual(start["judge_remarks"], "late start") + + +class TestJudgeEntryHttp(ResultEntryCompetition, compy_testing.CompyServerTestCase): + """The endpoints a judge phone uses to look up athletes and save results.""" + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.buildCompetition("Judge Entry Open") + with cls.app.app_context(): + import compy_data + data = compy_data.CompyData(cls.db, cls.app, cls.comp_id) + data.addJudge("Judy", "Judge") + judges = {} + data.getJudgeData(judges) + cls.judge_id = judges["judges"][0]["id"] + qr_data = data.getJudgeQrCode(cls.judge_id, cls.base_url + "/") + cls.judge_hash = qr_data[3].split("hash=")[1] + cls.judge_params = {"comp_id": cls.comp_id, "judge_id": cls.judge_id, + "judge_hash": cls.judge_hash} + + def putResult(self, first_name, rp, penalty, card, remarks, + judge_remarks="", auth=None, extra=None): + body = {"id": self.start_ids[first_name], "rp": rp, "penalty": penalty, + "card": card, "remarks": remarks, "judge_remarks": judge_remarks} + body |= self.judge_params if auth is None else auth + body |= extra or {} + return requests.put(self.base_url + "/result", json=body) + + def getAthleteResult(self, first_name): + response = requests.get(self.base_url + "/judge/athlete/result", + params=self.judge_params | + {"s_id": self.start_ids[first_name]}) + self.assertEqual(response.status_code, 200) + return response.json() + + def testJudgeSeesLaneList(self): + response = requests.get(self.base_url + "/judge/athletes", + params=self.judge_params | + {"day": "2026-01-01", + "block": self.cwt_block, "lane": "1"}) + self.assertEqual(response.status_code, 200) + lane_list = response.json()["lane_list"] + self.assertEqual(len(lane_list), len(self.CWT_ATHLETES)) + # sorted by official top time, every entry has a start id to save to + self.assertEqual(lane_list[0]["Name"], "Wht Diver") + self.assertTrue(all("s_id" in entry for entry in lane_list)) + + def testJudgeEntersWhiteResult(self): + response = self.putResult("Ada", "52", 0, "WHITE", "OK", "good dive") + self.assertEqual(response.status_code, 200) + # without discipline/gender/country the athlete's result is returned + self.assertEqual(response.json()["Card"], "WHITE") + stored = self.getAthleteResult("Ada") + self.assertEqual(stored["RP"], "52.0") + self.assertEqual(stored["Card"], "WHITE") + self.assertEqual(stored["Remarks"], "OK") + self.assertEqual(stored["JudgeRemarks"], "good dive") + + def testJudgeEntersYellowWithUnderApPenalty(self): + # AP 50, RP 45, manual penalty 1 -> 6 penalty points in total + response = self.putResult("Bel", "45", 1, "YELLOW", "UNDER AP") + self.assertEqual(response.status_code, 200) + stored = self.getAthleteResult("Bel") + self.assertEqual(stored["Card"], "YELLOW") + self.assertEqual(float(stored["Penalty"]), 6.) + + def testJudgeEntersRedCardDq(self): + response = self.putResult("Cyd", "0", 0, "RED", "DQBO-SURFACE", "BO") + self.assertEqual(response.status_code, 200) + stored = self.getAthleteResult("Cyd") + self.assertEqual(stored["Card"], "RED") + self.assertEqual(stored["Remarks"], "DQBO-SURFACE") + + def testInvalidCardIsRejectedOverHttp(self): + response = self.putResult("Dee", "50", 0, "GREEN", "OK") + self.assertEqual(response.status_code, 400) + # nothing was stored for this athlete + self.assertIsNone(self.getAthleteResult("Dee")["Card"]) + + def testResultEntryWithoutCredentialsIsRejected(self): + response = self.putResult("Ada", "10", 0, "WHITE", "OK", + auth={"comp_id": self.comp_id}) + self.assertEqual(response.status_code, 401) + + def testJudgePageRendersWithValidHash(self): + response = requests.get( + self.base_url + "/judge/%d/%d" % (self.comp_id, self.judge_id), + params={"hash": self.judge_hash}) + self.assertEqual(response.status_code, 200) + self.assertIn("Judge Entry Open", response.text) + self.assertIn("Judy Judge", response.text) + + def testJudgePageWithForgedHashIs404(self): + response = requests.get( + self.base_url + "/judge/%d/%d" % (self.comp_id, self.judge_id), + params={"hash": "deadbeef"}) + self.assertEqual(response.status_code, 404) + + def testAdminEntersResultAndReceivesUpdatedRanking(self): + session = self.adminSession() + body = {"comp_id": self.comp_id, "id": self.start_ids["Eva"], + "rp": "48", "penalty": 0, "card": "WHITE", "remarks": "OK", + "judge_remarks": "", "discipline": "CWT", "gender": "F", + "country": "International"} + response = session.put(self.base_url + "/result", json=body) + self.assertEqual(response.status_code, 200) + # with discipline/gender/country the whole ranking comes back + eva = [r for r in response.json()["results"] if r["Name"] == "Eva Diver"] + self.assertEqual(len(eva), 1) + self.assertEqual(eva[0]["Points"], "48.00") + + +if __name__ == '__main__': + unittest.main() diff --git a/compy_result_view_test.py b/compy_result_view_test.py new file mode 100644 index 0000000..1b266ad --- /dev/null +++ b/compy_result_view_test.py @@ -0,0 +1,160 @@ +"""Tests for the public result view (the pages spectators open). + +Results must only be visible after the organizer publishes them, and the +published ranking must faithfully reflect what the judges entered: rank +order, points with penalties, red cards shown with zero points and no +rank, DNS shown as DNS. Also checks that the admin-only result JSON stays +behind the login. +""" +import unittest + +import requests + +import athlete +import compy_testing + + +class TestPublicResultView(compy_testing.CompyServerTestCase): + + # name -> (AP, RP, penalty, card, remarks) + RESULTS = {"Bea": (52, "55", 0, "WHITE", "OK"), + "Ann": (50, "50", 0, "WHITE", "OK"), + "Yol": (45, "43", 0, "YELLOW", "UNDER AP"), # +2 under-AP penalty + "Cat": (60, "60", 0, "RED", "DQBO-SURFACE"), + "Dot": (40, "", 0, "RED", "DNS")} + + @classmethod + def setUpClass(cls): + super().setUpClass() + with cls.app.app_context(): + import compy_data + data = compy_data.CompyData(cls.db, cls.app) + data.changeName("Result View Open", False) + cls.comp_id = data.id_ + block_id = cls.db.insert( + "INSERT INTO block (competition_id, day, disciplines) VALUES (?, ?, ?)", + (cls.comp_id, 20260101, data.disciplineListToInt(["CWT"]))) + ot = 1000 + for first_name, (ap, rp, penalty, card, remarks) in cls.RESULTS.items(): + new_athlete = athlete.Athlete.fromArgs( + "id-" + first_name, first_name, "Diver", "F", "AUT", "", cls.db) + new_athlete.associateWithComp(cls.comp_id) + s_id = cls.db.insert( + '''INSERT INTO start + (competition_athlete_id, discipline, lane, OT, AP, block) + VALUES (?, ?, ?, ?, ?, ?)''', + (new_athlete.comp_athlete_id_, "CWT", 1, ot, ap, block_id)) + ot += 5 + # enter the results the way the judging interface does + loaded = compy_data.CompyData(cls.db, cls.app, cls.comp_id) + ret, _ = loaded.updateResult(s_id, rp, penalty, card, remarks, "") + if ret != 0: + raise AssertionError("result entry failed in test setup") + + def publish(self, published): + response = self.adminSession().request( + "UPDATE", self.base_url + "/publish_results", + json={"publish_results": published, "comp_id": self.comp_id}) + self.assertEqual(response.status_code, 200) + + def resultsList(self, discipline, expect_status=200): + # discipline 0 is "Overall", 1 is CWT; country 0 is "International" + response = requests.get(self.base_url + "/results_list", + params={"comp_id": self.comp_id, + "discipline": discipline, + "gender": "Female", "country": 0}) + self.assertEqual(response.status_code, expect_status) + return response + + def testResultsPageRequiresPublishing(self): + self.publish(False) + response = requests.get(self.base_url + "/results", + params={"comp_id": self.comp_id}) + self.assertEqual(response.status_code, 400) + self.resultsList(1, expect_status=400) + + def testResultsPageShowsPublishedCompetition(self): + self.publish(True) + response = requests.get(self.base_url + "/results", + params={"comp_id": self.comp_id}) + self.assertEqual(response.status_code, 200) + self.assertIn("Result View Open", response.text) + + def testCompetitionListShowsOnlyPublished(self): + # spectators without a link get a list of published competitions + self.publish(True) + response = requests.get(self.base_url + "/results") + self.assertEqual(response.status_code, 200) + self.assertIn("Result View Open", response.text) + self.publish(False) + response = requests.get(self.base_url + "/results") + self.assertEqual(response.status_code, 200) + self.assertNotIn("Result View Open", response.text) + + def testPublishedRankingMatchesEnteredResults(self): + self.publish(True) + results = self.resultsList(1).json()["results"] + by_name = {r["name"]: r for r in results} + + # ranked athletes come first, in points order + names_in_order = [r["name"] for r in results] + self.assertEqual(names_in_order[:3], ["Bea Diver", "Ann Diver", "Yol Diver"]) + self.assertEqual(by_name["Bea Diver"]["rank"], 1) + self.assertEqual(by_name["Ann Diver"]["rank"], 2) + self.assertEqual(by_name["Yol Diver"]["rank"], 3) + + # points include the yellow card's under-AP penalty (43 - 2 = 41) + self.assertEqual(by_name["Bea Diver"]["points"], "55.00") + self.assertEqual(by_name["Yol Diver"]["points"], "41.00") + self.assertEqual(float(by_name["Yol Diver"]["penalty"]), 2.) + + # a red card is shown with zero points and no rank + self.assertEqual(by_name["Cat Diver"]["card"], "RED") + self.assertEqual(by_name["Cat Diver"]["points"], "0.00") + self.assertEqual(by_name["Cat Diver"]["rank"], "") + self.assertEqual(by_name["Cat Diver"]["remarks"], "DQBO-SURFACE") + + # DNS is shown as DNS, not as a performance + self.assertEqual(by_name["Dot Diver"]["value"], "DNS") + self.assertEqual(by_name["Dot Diver"]["rank"], "") + + def testOverallRankingExcludesRedCardAndDns(self): + self.publish(True) + results = self.resultsList(0).json()["results"] + names = [r["name"] for r in results] + self.assertEqual(names, ["Bea Diver", "Ann Diver", "Yol Diver"]) + self.assertEqual(results[0]["rank"], 1) + self.assertEqual(results[0]["value"], "55.00") + # the overall ranking lists the individual discipline results + self.assertEqual(results[0]["individual_results"][0]["dis"], "CWT") + + def testResultsListRejectsBadParameters(self): + self.publish(True) + response = requests.get(self.base_url + "/results_list", + params={"comp_id": self.comp_id, "discipline": 1, + "gender": "X", "country": 0}) + self.assertEqual(response.status_code, 400) + self.resultsList(99, expect_status=400) + + def testUnpublishingHidesResults(self): + self.publish(True) + self.resultsList(1, expect_status=200) + self.publish(False) + self.resultsList(1, expect_status=400) + + def testAdminResultJsonRequiresLogin(self): + # the detailed result view (with judge remarks) is admin only + params = {"comp_id": self.comp_id, "discipline": "CWT", + "gender": "F", "country": "International"} + response = requests.get(self.base_url + "/result", params=params) + self.assertEqual(response.status_code, 401) + + response = self.adminSession().get(self.base_url + "/result", params=params) + self.assertEqual(response.status_code, 200) + by_name = {r["Name"]: r for r in response.json()["results"]} + self.assertEqual(by_name["Bea Diver"]["Points"], "55.00") + self.assertEqual(by_name["Cat Diver"]["Card"], "RED") + + +if __name__ == '__main__': + unittest.main() diff --git a/compy_testing.py b/compy_testing.py index a1edf61..0420844 100644 --- a/compy_testing.py +++ b/compy_testing.py @@ -19,6 +19,7 @@ import unittest import flask +import requests from werkzeug.serving import make_server import compy_data @@ -27,12 +28,14 @@ REPO_ROOT = os.path.dirname(os.path.abspath(__file__)) TEST_COMPETITION_XLSX = os.path.join(REPO_ROOT, "test_competition.xlsx") +ADMIN_PASSWORD = "compy-test-password" def makeApp(database_path): app = flask.Flask("compy", root_path=REPO_ROOT) app.config["DATABASE"] = database_path app.config["SECRET_KEY"] = "0123456789abcdef0123456789abcdef_compy_test" + app.config["ADMIN_PASSWORD"] = ADMIN_PASSWORD return app @@ -96,3 +99,13 @@ def tearDownClass(cls): cls.server.shutdown() cls.server_thread.join() super().tearDownClass() + + @classmethod + def adminSession(cls): + """A requests session that is logged in to the admin interface.""" + session = requests.Session() + response = session.post(cls.base_url + "/admin/login", + data={"password": ADMIN_PASSWORD}) + if response.status_code != 200 or not session.cookies: + raise AssertionError("admin login failed in test setup") + return session diff --git a/static/compy.js b/static/compy.js index 029f6af..92705a1 100644 --- a/static/compy.js +++ b/static/compy.js @@ -25,6 +25,13 @@ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ */ +// if the admin session has expired (or is missing) every api call returns +// 401; send the user to the login page in that case +$(document).ajaxError(function(event, jqxhr) { + if (jqxhr.status == 401) + window.location.href = "/admin/login"; +}); + var _global_prev_name = ""; var _days_with_disciplines_lanes = null; var _comp_id = null; diff --git a/templates/login.html b/templates/login.html new file mode 100644 index 0000000..1b848f2 --- /dev/null +++ b/templates/login.html @@ -0,0 +1,80 @@ + + + + + Compy {{version}} + + + + + +
+

Compy {{version}}

+

Admin login

+
+ +
+ {% if error %} +

{{ error }}

+ {% endif %} +
+ + diff --git a/templates/template.html b/templates/template.html index e66d478..c3392b1 100644 --- a/templates/template.html +++ b/templates/template.html @@ -56,6 +56,7 @@

Compy {{version}}

Lane lists Results Clock + Logout
diff --git a/tests/compy.resource b/tests/compy.resource index 529a331..53ecefe 100644 --- a/tests/compy.resource +++ b/tests/compy.resource @@ -10,8 +10,9 @@ ${URL} localhost ${PORT} 5000 ${PATH} . ${BASE_URL} http://${URL}:${PORT}/${PATH} -${ADMIN_KEY} 156a7f# -${ADMIN_URL} ${BASE_URL}/admin?auth=${ADMIN_KEY} +# must match FLASK_ADMIN_PASSWORD in the .env used by the server under test +${ADMIN_PASSWORD} compy-admin +${ADMIN_URL} ${BASE_URL}/admin # Elements ${SETTINGS_BUTTON} id=settings_button >> a ${COMP_NAME_FIELD} id=comp_name @@ -23,6 +24,11 @@ Open Admin Page # Debug enable # Open Browser New Page ${ADMIN_URL} + ${login_form} = Get Element Count id=admin_password + IF ${login_form} > 0 + Fill Text id=admin_password ${ADMIN_PASSWORD} + Click id=admin_login_button + END Get Text h1 contains Compy Goto Settings