# -*- coding: utf-8 -*- r'''Tests for leaderboard_app/data.py. Tests CSV loading, main leaderboard building with first-row-per-team logic, team filtering, numeric rounding, and team history sorting. ''' # System modules from pathlib import Path # External modules import pandas as pd import pytest # Internal modules from leaderboard_app.data import ( LEADERBOARD_COLUMNS, HISTORY_COLUMNS, _empty_display, _empty_submissions, _format_timestamps, _normalize_submissions, _round_numeric, build_main_leaderboard, build_team_history, load_submissions, ) CSV_HEADER = ( "team_name,score,mae_era5_region1,mae_era5_region2," "crps_aimip_region1,crps_aimip_region2,submitted_at\n" ) # ----------------------------------------------------------- # Helper functions # ----------------------------------------------------------- def _create_submissions_frame( rows: list[dict[str, object]], ) -> pd.DataFrame: r''' Create a submissions DataFrame for testing. Parameters ---------- rows : list[dict[str, object]] List of row dictionaries. Returns ------- pandas.DataFrame Submissions frame with required columns. ''' frame = pd.DataFrame(rows) frame["submitted_at"] = pd.to_datetime(frame["submitted_at"], utc=True) frame["_source_order"] = range(len(frame)) return frame # ----------------------------------------------------------- # Unit tests # ----------------------------------------------------------- class TestLeaderboardDataUnittest: r'''Isolated unit tests for leaderboard data helpers.''' def test_empty_submissions_has_required_columns(self) -> None: r'''Verifies _empty_submissions includes all required columns.''' # Act frame = _empty_submissions() # Assert assert "team_name" in frame.columns assert "score" in frame.columns assert "submitted_at" in frame.columns assert "_source_order" in frame.columns assert frame.empty def test_empty_display_has_requested_columns(self) -> None: r'''Verifies _empty_display creates frame with specified columns.''' # Arrange columns = ["team_name", "score"] # Act frame = _empty_display(columns) # Assert assert list(frame.columns) == columns assert frame.empty def test_round_numeric_rounds_to_three_decimals(self) -> None: r'''Verifies _round_numeric rounds numeric columns to 3 decimals.''' # Arrange frame = pd.DataFrame( { "team_name": ["TeamA"], "score": [0.123456789], "mae_era5_region1": [1.987654321], } ) # Act result = _round_numeric(frame) # Assert assert result.loc[0, "score"] == 0.123 assert result.loc[0, "mae_era5_region1"] == 1.988 def test_format_timestamps_converts_to_utc_strings(self) -> None: r'''Verifies _format_timestamps converts datetime to UTC strings.''' # Arrange series = pd.Series( pd.to_datetime(["2026-04-24T10:30:00Z"], utc=True) ) # Act result = _format_timestamps(series) # Assert assert result.iloc[0] == "2026-04-24 10:30:00 UTC" def test_normalize_submissions_raises_on_missing_columns(self) -> None: r'''Missing columns raise during normalization.''' # Arrange frame = pd.DataFrame({"team_name": ["TeamA"]}) # Act & Assert with pytest.raises(ValueError, match="Missing required columns"): _normalize_submissions(frame) def test_normalize_submissions_adds_source_order(self) -> None: r'''Verifies _normalize_submissions adds _source_order column.''' # Arrange frame = pd.DataFrame( { "team_name": ["TeamA", "TeamB"], "score": [0.9, 0.8], "mae_era5_region1": [1.0, 1.1], "mae_era5_region2": [1.0, 1.1], "crps_aimip_region1": [1.0, 1.1], "crps_aimip_region2": [1.0, 1.1], "submitted_at": ["2026-04-24", "2026-04-23"], } ) # Act result = _normalize_submissions(frame) # Assert assert "_source_order" in result.columns assert result.loc[0, "_source_order"] == 0 assert result.loc[1, "_source_order"] == 1 # ----------------------------------------------------------- # Functional tests # ----------------------------------------------------------- class TestLeaderboardDataFunctional: r'''End-to-end tests for leaderboard data building logic.''' def test_build_main_leaderboard_takes_first_row_per_team(self) -> None: r'''Main table keeps the first row for each team.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamA", "score": 0.95, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T10:00:00Z", }, { "team_name": "TeamA", "score": 0.80, "mae_era5_region1": 1.5, "mae_era5_region2": 1.5, "crps_aimip_region1": 1.5, "crps_aimip_region2": 1.5, "submitted_at": "2026-04-24T11:00:00Z", }, { "team_name": "TeamB", "score": 0.85, "mae_era5_region1": 1.2, "mae_era5_region2": 1.2, "crps_aimip_region1": 1.2, "crps_aimip_region2": 1.2, "submitted_at": "2026-04-24T09:00:00Z", }, ] ) # Act main = build_main_leaderboard(submissions) # Assert assert len(main) == 2 team_a_row = main.loc[main["team_name"] == "TeamA"] assert team_a_row.iloc[0]["score"] == 0.95 def test_build_main_leaderboard_filters_by_team_query(self) -> None: r'''Verifies main leaderboard restricts results by team name filter.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamAlpha", "score": 0.95, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T10:00:00Z", }, { "team_name": "TeamBeta", "score": 0.85, "mae_era5_region1": 1.2, "mae_era5_region2": 1.2, "crps_aimip_region1": 1.2, "crps_aimip_region2": 1.2, "submitted_at": "2026-04-24T09:00:00Z", }, ] ) # Act main = build_main_leaderboard(submissions, team_query="Alpha") # Assert assert len(main) == 1 assert main.iloc[0]["team_name"] == "TeamAlpha" def test_build_main_leaderboard_rounds_numerics_to_three_decimals( self, ) -> None: r'''Verifies displayed numeric values are rounded to 3 decimals.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamA", "score": 0.123456789, "mae_era5_region1": 1.987654321, "mae_era5_region2": 2.123456789, "crps_aimip_region1": 3.456789012, "crps_aimip_region2": 4.567890123, "submitted_at": "2026-04-24T10:00:00Z", } ] ) # Act main = build_main_leaderboard(submissions) # Assert assert main.iloc[0]["score"] == 0.123 assert main.iloc[0]["mae_era5_region1"] == 1.988 assert main.iloc[0]["mae_era5_region2"] == 2.123 def test_build_main_leaderboard_labels_non_competitive_teams( self, ) -> None: r'''Non-competitive teams are labelled with (non-competitive).''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "OrgTeam", "score": 0.99, "mae_era5_region1": 0.5, "mae_era5_region2": 0.5, "crps_aimip_region1": 0.5, "crps_aimip_region2": 0.5, "submitted_at": "2026-04-24T10:00:00Z", }, { "team_name": "CompetingTeam", "score": 0.85, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T09:00:00Z", }, ] ) non_competitive = frozenset({"OrgTeam"}) # Act main = build_main_leaderboard( submissions, non_competitive_teams=non_competitive, ) # Assert org_row = main.loc[ main["team_name"].str.contains("OrgTeam") ] assert len(org_row) == 1 assert org_row.iloc[0]["team_name"] == ( "* OrgTeam" ) comp_row = main.loc[ main["team_name"] == "CompetingTeam" ] assert len(comp_row) == 1 def test_build_main_leaderboard_non_competitive_pushed_to_end( self, ) -> None: r'''Non-competitive teams appear after all competitive teams.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "OrgTeam", "score": 0.99, "mae_era5_region1": 0.5, "mae_era5_region2": 0.5, "crps_aimip_region1": 0.5, "crps_aimip_region2": 0.5, "submitted_at": "2026-04-24T10:00:00Z", }, { "team_name": "CompetingTeam", "score": 0.85, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T09:00:00Z", }, ] ) non_competitive = frozenset({"OrgTeam"}) # Act main = build_main_leaderboard( submissions, non_competitive_teams=non_competitive, ) # Assert — competitive team first despite lower score assert main.iloc[0]["team_name"] == "CompetingTeam" assert main.iloc[1]["team_name"] == "* OrgTeam" def test_build_main_leaderboard_non_competitive_sorted_by_score( self, ) -> None: r'''Multiple non-competitive teams are ordered by score desc.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "NC_Low", "score": 0.60, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T08:00:00Z", }, { "team_name": "Competitor", "score": 0.75, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T09:00:00Z", }, { "team_name": "NC_High", "score": 0.90, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T10:00:00Z", }, ] ) non_competitive = frozenset({"NC_Low", "NC_High"}) # Act main = build_main_leaderboard( submissions, non_competitive_teams=non_competitive, ) # Assert assert main.iloc[0]["team_name"] == "Competitor" assert main.iloc[1]["team_name"] == "* NC_High" assert main.iloc[2]["team_name"] == "* NC_Low" def test_build_team_history_returns_all_submissions_for_team(self) -> None: r'''Team history keeps every row for that team.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamA", "score": 0.95, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T10:00:00Z", }, { "team_name": "TeamA", "score": 0.80, "mae_era5_region1": 1.5, "mae_era5_region2": 1.5, "crps_aimip_region1": 1.5, "crps_aimip_region2": 1.5, "submitted_at": "2026-04-24T11:00:00Z", }, { "team_name": "TeamB", "score": 0.85, "mae_era5_region1": 1.2, "mae_era5_region2": 1.2, "crps_aimip_region1": 1.2, "crps_aimip_region2": 1.2, "submitted_at": "2026-04-24T09:00:00Z", }, ] ) # Act history = build_team_history(submissions, "TeamA") # Assert assert len(history) == 2 assert all(history["submitted_at"].str.contains("2026-04-24")) def test_build_team_history_sorted_by_submitted_at_descending( self, ) -> None: r'''Verifies team history is sorted by submitted_at descending.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamA", "score": 0.95, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T08:00:00Z", }, { "team_name": "TeamA", "score": 0.80, "mae_era5_region1": 1.5, "mae_era5_region2": 1.5, "crps_aimip_region1": 1.5, "crps_aimip_region2": 1.5, "submitted_at": "2026-04-24T12:00:00Z", }, { "team_name": "TeamA", "score": 0.85, "mae_era5_region1": 1.2, "mae_era5_region2": 1.2, "crps_aimip_region1": 1.2, "crps_aimip_region2": 1.2, "submitted_at": "2026-04-24T10:00:00Z", }, ] ) # Act history = build_team_history(submissions, "TeamA") # Assert assert len(history) == 3 assert history.iloc[0]["submitted_at"] == "2026-04-24 12:00:00 UTC" assert history.iloc[1]["submitted_at"] == "2026-04-24 10:00:00 UTC" assert history.iloc[2]["submitted_at"] == "2026-04-24 08:00:00 UTC" def test_load_submissions_returns_empty_when_file_missing( self, tmp_path: Path, ) -> None: r'''Missing CSV returns an empty snapshot.''' # Arrange missing_path = tmp_path / "missing.csv" # Act snapshot = load_submissions(missing_path) # Assert assert snapshot.submissions.empty assert "not found" in snapshot.status_text def test_load_submissions_normalizes_valid_csv( self, tmp_path: Path, ) -> None: r'''Verifies load_submissions normalizes a valid CSV correctly.''' # Arrange csv_path = tmp_path / "submissions.csv" csv_content = ( CSV_HEADER + "TeamA,0.95,1.0,1.0,1.0,1.0,2026-04-24T10:00:00Z\n" + "TeamB,0.85,1.2,1.2,1.2,1.2,2026-04-24T11:00:00Z\n" ) csv_path.write_text(csv_content, encoding="utf-8") # Act snapshot = load_submissions(csv_path) # Assert assert len(snapshot.submissions) == 2 assert "Loaded 2 public submissions" in snapshot.status_text # ----------------------------------------------------------- # Edge cases # ----------------------------------------------------------- class TestLeaderboardDataEdgeCases: r'''Boundary condition tests for leaderboard data helpers.''' def test_build_main_leaderboard_handles_empty_submissions(self) -> None: r'''Verifies main leaderboard handles empty submissions frame.''' # Arrange submissions = _empty_submissions() # Act main = build_main_leaderboard(submissions) # Assert assert main.empty assert list(main.columns) == LEADERBOARD_COLUMNS def test_build_main_leaderboard_no_label_when_no_non_competitive( self, ) -> None: r'''Team names are unmodified when non_competitive_teams is empty.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamA", "score": 0.9, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T10:00:00Z", } ] ) # Act main = build_main_leaderboard( submissions, non_competitive_teams=frozenset(), ) # Assert assert main.iloc[0]["team_name"] == "TeamA" def test_build_team_history_handles_empty_team_name(self) -> None: r'''Verifies team history returns empty frame for empty team name.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamA", "score": 0.95, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T10:00:00Z", } ] ) # Act history = build_team_history(submissions, "") # Assert assert history.empty assert list(history.columns) == HISTORY_COLUMNS def test_build_team_history_handles_nonexistent_team(self) -> None: r'''Verifies team history returns empty frame for nonexistent team.''' # Arrange submissions = _create_submissions_frame( [ { "team_name": "TeamA", "score": 0.95, "mae_era5_region1": 1.0, "mae_era5_region2": 1.0, "crps_aimip_region1": 1.0, "crps_aimip_region2": 1.0, "submitted_at": "2026-04-24T10:00:00Z", } ] ) # Act history = build_team_history(submissions, "TeamNonexistent") # Assert assert history.empty def test_load_submissions_handles_malformed_csv( self, tmp_path: Path, ) -> None: r'''Verifies load_submissions handles malformed CSV gracefully.''' # Arrange csv_path = tmp_path / "malformed.csv" csv_path.write_text("invalid,csv,content\n1,2", encoding="utf-8") # Act snapshot = load_submissions(csv_path) # Assert assert snapshot.submissions.empty assert "Could not load" in snapshot.status_text