diff --git a/src/prism/flashlight.py b/src/prism/flashlight.py new file mode 100644 index 00000000..146c306d --- /dev/null +++ b/src/prism/flashlight.py @@ -0,0 +1,142 @@ +import functools +import logging +from json import JSONDecodeError +from typing import Any + +import requests +from requests.exceptions import RequestException + +from prism.errors import APIError +from prism.player import Tags, TagSeverity +from prism.ratelimiting import RateLimiter +from prism.requests import make_prism_requests_session +from prism.retry import ExecutionError, execute_with_retry + +logger = logging.getLogger(__name__) + +FLASHLIGHT_API_URL = "https://flashlight.prismoverlay.com" + + +class FlashlightTagsProvider: + def __init__( + self, + *, + retry_limit: int, + initial_timeout: float, + ) -> None: + self._retry_limit = retry_limit + self._initial_timeout = initial_timeout + self._session = make_prism_requests_session() + self._limiter = RateLimiter(limit=120, window=60) + + @property + def seconds_until_unblocked(self) -> float: + """Return the number of seconds until we are unblocked""" + return self._limiter.block_duration_seconds + + def _make_tags_request( + self, + *, + url: str, + user_id: str, + last_try: bool, + urchin_api_key: str | None, + ) -> requests.Response: # pragma: nocover + headers = {"X-User-Id": user_id} + if urchin_api_key: + headers["X-Urchin-Api-Key"] = urchin_api_key + + try: + # Uphold our prescribed rate-limits + with self._limiter: + response = self._session.get(url, headers=headers, timeout=10) + except RequestException as e: + raise ExecutionError( + "Request to flashlight failed due to an unknown error" + ) from e + + if response.status_code == 429 or response.status_code == 503 and not last_try: + raise ExecutionError( + "Request to flashlight failed due to intermittent error, retrying" + ) + + return response + + def get_tags( + self, + uuid: str, + *, + urchin_api_key: str | None, + user_id: str, + ) -> Tags: # pragma: nocover + """Get the tags for the given player (are they a sniper/cheater)""" + url = f"{FLASHLIGHT_API_URL}/v1/tags/{uuid}" + + try: + response = execute_with_retry( + functools.partial( + self._make_tags_request, + url=url, + user_id=user_id, + urchin_api_key=urchin_api_key, + ), + retry_limit=self._retry_limit, + initial_timeout=self._initial_timeout, + ) + except ExecutionError as e: + raise APIError(f"Request to flashlight failed for {uuid=}.") from e + + if not response: + raise APIError( + f"Request to flashlight failed with status code " + f"{response.status_code} when getting tags for player {uuid}. " + f"Response: {response.text}" + ) + + try: + response_json = response.json() + except JSONDecodeError as e: + raise APIError( + "Failed parsing the response from flashlight. " + f"Raw content: {response.text}" + ) from e + + return parse_flashlight_tags(response_json) + + +def validate_tag_severity(tag_severity: object) -> TagSeverity | None: + """Validate that the string is a valid TagSeverity""" + if not isinstance(tag_severity, str): + return None + + if tag_severity == "none": + return "none" + if tag_severity == "medium": + return "medium" + if tag_severity == "high": + return "high" + + return None + + +def parse_flashlight_tags(response_json: Any) -> Tags: + """Parse the flashlight tags from the response JSON""" + tags_data = response_json.get("tags", {}) + if not isinstance(tags_data, dict): + raise APIError(f"Invalid tags data {tags_data=} {type(tags_data)=}") + + cheating_severity = validate_tag_severity(tags_data.get("cheating", None)) + if cheating_severity is None: + raise APIError( + f"Invalid cheating tag severity " f"{response_json.get('cheating', None)=}" + ) + sniper_severity = validate_tag_severity(tags_data.get("sniping", None)) + if sniper_severity is None: + raise APIError( + f"Invalid sniping tag severity " f"{response_json.get('sniping', None)=}" + ) + + return Tags( + cheating=cheating_severity, + sniping=sniper_severity, + ) diff --git a/src/prism/overlay/__main__.py b/src/prism/overlay/__main__.py index d1e3ceef..7286b951 100644 --- a/src/prism/overlay/__main__.py +++ b/src/prism/overlay/__main__.py @@ -67,6 +67,7 @@ def main() -> None: # pragma: nocover nick_database = NickDatabase.from_disk([], default_database=default_database) # Import late so we can patch ssl certs in requests + from prism.flashlight import FlashlightTagsProvider from prism.mojang import MojangAccountProvider from prism.overlay.antisniper_api import ( StrangePlayerProvider, @@ -82,6 +83,7 @@ def main() -> None: # pragma: nocover retry_limit=5, initial_timeout=2, get_time_ns=time.time_ns ) winstreak_provider = PlaceholderWinstreakProvider() + tags_provider = FlashlightTagsProvider(retry_limit=5, initial_timeout=2) controller = OverlayController( state=OverlayState(), @@ -90,6 +92,7 @@ def main() -> None: # pragma: nocover account_provider=account_provider, player_provider=player_provider, winstreak_provider=winstreak_provider, + tags_provider=tags_provider, ) if options.test_ssl: diff --git a/src/prism/overlay/behaviour.py b/src/prism/overlay/behaviour.py index 62ce6e39..99a67e86 100644 --- a/src/prism/overlay/behaviour.py +++ b/src/prism/overlay/behaviour.py @@ -140,6 +140,22 @@ def get_and_cache_player( completed_queue.put(username) logger.debug(f"Updated missing winstreak for {username}") + # Get tags + if isinstance(player, KnownPlayer): + tags = controller.get_tags(player.uuid) + if tags is ERROR_DURING_PROCESSING: + logger.error(f"Error getting tags for {username}") + else: + for alias in player.aliases: + controller.player_cache.update_cached_player( + alias, + functools.partial(KnownPlayer.set_tags, tags=tags), + ) + + # Tell the main thread that we got the tags + completed_queue.put(username) + logger.debug(f"Set tags for {username}") + def update_settings(new_settings: SettingsDict, controller: OverlayController) -> None: """ diff --git a/src/prism/overlay/controller.py b/src/prism/overlay/controller.py index fc694a0f..5f5daa5e 100644 --- a/src/prism/overlay/controller.py +++ b/src/prism/overlay/controller.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Protocol from prism.errors import APIError, APIKeyError, APIThrottleError, PlayerNotFoundError -from prism.player import MISSING_WINSTREAKS, KnownPlayer, Winstreaks +from prism.player import MISSING_WINSTREAKS, KnownPlayer, Tags, Winstreaks from prism.ssl_errors import MissingLocalIssuerSSLError if TYPE_CHECKING: # pragma: no cover @@ -50,6 +50,19 @@ def get_estimated_winstreaks_for_uuid( def seconds_until_unblocked(self) -> float: ... +class TagsProvider(Protocol): + def get_tags( + self, + uuid: str, + *, + user_id: str, + urchin_api_key: str | None, + ) -> Tags: ... + + @property + def seconds_until_unblocked(self) -> float: ... + + class OverlayController: def __init__( self, @@ -59,6 +72,7 @@ def __init__( account_provider: AccountProvider, player_provider: PlayerProvider, winstreak_provider: WinstreakProvider, + tags_provider: TagsProvider, ) -> None: from prism.overlay.player_cache import PlayerCache @@ -81,6 +95,7 @@ def __init__( self._account_provider = account_provider self._player_provider = player_provider self._winstreak_provider = winstreak_provider + self._tags_provider = tags_provider def get_uuid(self, username: str) -> str | None | ProcessingError: try: @@ -174,3 +189,16 @@ def get_estimated_winstreaks(self, uuid: str) -> tuple[Winstreaks, bool]: self.antisniper_api_key_throttled = False self.missing_local_issuer_certificate = False return winstreaks, accurate + + def get_tags(self, uuid: str) -> Tags | ProcessingError: + try: + tags = self._tags_provider.get_tags( + uuid=uuid, + user_id=self.settings.user_id, + urchin_api_key=None, # TODO: Implement urchin api key setting + ) + except APIError as e: + logger.error(f"Error getting tags for {uuid=}", exc_info=e) + return ERROR_DURING_PROCESSING + else: + return tags diff --git a/src/prism/overlay/output/cell_renderer.py b/src/prism/overlay/output/cell_renderer.py index cf7e31d5..22d7c63e 100644 --- a/src/prism/overlay/output/cell_renderer.py +++ b/src/prism/overlay/output/cell_renderer.py @@ -1,12 +1,20 @@ from collections.abc import Sequence from dataclasses import dataclass, replace from functools import lru_cache -from typing import assert_never +from typing import Literal, assert_never from prism.overlay.output.cells import CellValue, ColorSection, ColumnName from prism.overlay.output.color import GUIColor, MinecraftColor from prism.overlay.output.config import RatingConfig, RatingConfigCollection -from prism.player import KnownPlayer, NickedPlayer, PendingPlayer, Player, UnknownPlayer +from prism.player import ( + KnownPlayer, + NickedPlayer, + PendingPlayer, + Player, + Tags, + TagSeverity, + UnknownPlayer, +) from prism.utils import format_seconds_short, truncate_float GUI_COLORS = ( @@ -38,6 +46,8 @@ class RenderedStats: wins: CellValue sessiontime: CellValue + tags: CellValue + def truncate_float_or_int(value: float | int, decimals: int) -> str: """Truncate the decimals of the float, or keep the int""" @@ -367,6 +377,59 @@ def render_stars( return replace(levels_rating, color_sections=color_sections) +def tag_severity_to_color(severity: Literal["medium", "high"]) -> str: + if severity == "medium": + return GUI_COLORS[3] + elif severity == "high": + return GUI_COLORS[4] + + +def render_tags(tags: Tags | None) -> CellValue: + if tags is None: + # Pending + return CellValue.monochrome(text="-", gui_color=GUI_COLORS[0]) + + def add_tag( + text: str, + color_sections: tuple[ColorSection, ...], + tag_char: str, + severity: TagSeverity, + ) -> tuple[str, tuple[ColorSection, ...]]: + if severity == "none": + return text, color_sections + + if text != "": + # There is already a tag, add space separator + text += " " + color_sections += (ColorSection(GUI_COLORS[0], 1),) + + return ( + text + tag_char, + color_sections + (ColorSection(tag_severity_to_color(severity), 1),), + ) + + color_sections: tuple[ColorSection, ...] = () + text = "" + + text, color_sections = add_tag(text, color_sections, "C", tags.cheating) + text, color_sections = add_tag(text, color_sections, "S", tags.sniping) + + if len(color_sections) == 0: + # The painter needs at least one color + return CellValue.monochrome(text="", gui_color=GUI_COLORS[0]) + + # Build hover text with detailed tag information + hover_parts: list[str] = [] + if tags.cheating != "none": + hover_parts.append(f"C: Cheating ({tags.cheating})") + if tags.sniping != "none": + hover_parts.append(f"S: Sniping ({tags.sniping})") + + hover_text = "\n".join(hover_parts) if hover_parts else None + + return CellValue(text=text, color_sections=color_sections, hover=hover_text) + + @lru_cache(maxsize=100) def render_stats( player: Player, @@ -480,6 +543,7 @@ def render_stats( rating_configs.sessiontime.rate_by_level, rating_configs.sessiontime.sort_ascending, ) + tags_cell = render_tags(player.tags) else: if isinstance(player, NickedPlayer): text = "nick" @@ -496,7 +560,7 @@ def render_stats( cell = CellValue.monochrome(text, gui_color=gui_color) stars_cell = index_cell = fkdr_cell = kdr_cell = bblr_cell = wlr_cell = cell winstreak_cell = kills_cell = finals_cell = beds_cell = wins_cell = cell - sessiontime_cell = cell + sessiontime_cell = tags_cell = cell username_cell = CellValue.monochrome(username_str, gui_color=GUIColor.WHITE) @@ -514,6 +578,7 @@ def render_stats( beds=beds_cell, wins=wins_cell, sessiontime=sessiontime_cell, + tags=tags_cell, ) diff --git a/src/prism/overlay/output/cells.py b/src/prism/overlay/output/cells.py index 0165be62..08a3426b 100644 --- a/src/prism/overlay/output/cells.py +++ b/src/prism/overlay/output/cells.py @@ -15,6 +15,7 @@ "beds", "wins", "sessiontime", + "tags", ] ALL_COLUMN_NAMES_ORDERED: tuple[ColumnName, ...] = ( @@ -31,6 +32,7 @@ "beds", "wins", "sessiontime", + "tags", ) ALL_COLUMN_NAMES: frozenset[ColumnName] = frozenset(ALL_COLUMN_NAMES_ORDERED) @@ -42,6 +44,7 @@ "kdr", "winstreak", "sessiontime", + "tags", ) COLUMN_NAMES: dict[ColumnName, str] = { @@ -58,6 +61,7 @@ "beds": "Beds", "wins": "Wins", "sessiontime": "Time", + "tags": "Tags", } @@ -85,13 +89,15 @@ class CellValue: text: str color_sections: tuple[ColorSection, ...] + hover: str | None = None @classmethod - def monochrome(cls, text: str, gui_color: str) -> Self: + def monochrome(cls, text: str, gui_color: str, hover: str | None = None) -> Self: """Make a monochrome cell value""" return cls( text, color_sections=(ColorSection(gui_color, -1),), + hover=hover, ) diff --git a/src/prism/overlay/output/overlay/main_content.py b/src/prism/overlay/output/overlay/main_content.py index 2571c443..078f0554 100644 --- a/src/prism/overlay/output/overlay/main_content.py +++ b/src/prism/overlay/output/overlay/main_content.py @@ -22,6 +22,52 @@ class Cell: text_widget: tk.Text +class Tooltip: # pragma: nocover + """Simple tooltip implementation for showing hover text""" + + def __init__(self, widget: tk.Widget, text: str) -> None: + self.widget = widget + self.text = text + self.tooltip_window: tk.Toplevel | None = None + self.widget.bind("", self.show_tooltip) + self.widget.bind("", self.hide_tooltip) + + def show_tooltip(self, event: "tk.Event[tk.Widget]") -> None: + """Show the tooltip window""" + if self.tooltip_window or not self.text: + return + + x = self.widget.winfo_rootx() + 20 + y = self.widget.winfo_rooty() + 20 + + self.tooltip_window = tk.Toplevel(self.widget) + self.tooltip_window.wm_overrideredirect(True) + self.tooltip_window.wm_geometry(f"+{x}+{y}") + + label = tk.Label( + self.tooltip_window, + text=self.text, + justify=tk.LEFT, + background="lightyellow", + relief=tk.SOLID, + borderwidth=1, + font=("Consolas", 10), + ) + label.pack() + + def hide_tooltip(self, event: "tk.Event[tk.Widget]") -> None: + """Hide the tooltip window""" + if self.tooltip_window: + self.tooltip_window.destroy() + self.tooltip_window = None + + def destroy(self) -> None: + """Clean up the tooltip""" + self.hide_tooltip(None) # type: ignore[arg-type] + self.widget.unbind("") + self.widget.unbind("") + + OverlayRow = tuple[tk.Button, tuple[Cell, ...]] @@ -44,6 +90,9 @@ def __init__( # Start with zero rows self.rows: list[OverlayRow] = [] + # Store tooltips for each cell widget + self.tooltips: dict[tk.Text, Tooltip] = {} + # Frame at the top to display info to the user self.info_frame = tk.Frame(self.frame, background="black") self.info_frame.pack(side=tk.TOP, expand=True, fill=tk.X) @@ -151,6 +200,10 @@ def pop_row(self) -> None: """Remove a row of cells from the table""" edit_button, cells = self.rows.pop() for cell in cells: + # Clean up tooltip if it exists + if cell.text_widget in self.tooltips: + self.tooltips[cell.text_widget].destroy() + del self.tooltips[cell.text_widget] cell.text_widget.destroy() edit_button.destroy() @@ -273,6 +326,21 @@ def update_content( width=len(new_text), ) + # Handle tooltip + if cell_value.hover: + # Remove old tooltip if it exists + if cell.text_widget in self.tooltips: + self.tooltips[cell.text_widget].destroy() + # Create new tooltip + self.tooltips[cell.text_widget] = Tooltip( + cell.text_widget, cell_value.hover + ) + else: + # Remove tooltip if no hover text + if cell.text_widget in self.tooltips: + self.tooltips[cell.text_widget].destroy() + del self.tooltips[cell.text_widget] + if nickname is None: edit_button.configure(state="disabled", command=lambda: None) else: diff --git a/src/prism/overlay/output/overlay/settings_page.py b/src/prism/overlay/output/overlay/settings_page.py index e543cdbb..451acf54 100644 --- a/src/prism/overlay/output/overlay/settings_page.py +++ b/src/prism/overlay/output/overlay/settings_page.py @@ -743,9 +743,12 @@ def __init__(self, parent: "SettingsPage") -> None: ) sort_order_label.grid(row=0, column=0, sticky=tk.E) + sort_choices = tuple( + filter(lambda column: column != "tags", ALL_COLUMN_NAMES_ORDERED) + ) self.sort_order_variable = tk.StringVar(value="") self.sort_order_menu = tk.OptionMenu( - self.frame, self.sort_order_variable, *ALL_COLUMN_NAMES_ORDERED + self.frame, self.sort_order_variable, *sort_choices ) self.sort_order_menu.grid(row=0, column=1) parent.make_widgets_scrollable(sort_order_label, self.sort_order_menu) diff --git a/src/prism/overlay/player_cache.py b/src/prism/overlay/player_cache.py index 06424c8e..62aa7111 100644 --- a/src/prism/overlay/player_cache.py +++ b/src/prism/overlay/player_cache.py @@ -28,11 +28,13 @@ def set_player_pending(self, username: str) -> PendingPlayer: """Note that this user is pending""" pending_player = PendingPlayer(username) + cache_key = username.lower() + with self._mutex: - if username in self._cache: + if cache_key in self._cache: logger.error(f"Player {username} set to pending, but already exists") - self._cache[username] = self._long_term_cache[username] = pending_player + self._cache[cache_key] = self._long_term_cache[cache_key] = pending_player return pending_player @@ -42,6 +44,8 @@ def set_cached_player( player: KnownPlayer | NickedPlayer | UnknownPlayer, genus: int, ) -> None: + cache_key = username.lower() + with self._mutex: if genus != self.current_genus: logger.warning( @@ -50,34 +54,42 @@ def set_cached_player( ) return - self._cache[username] = self._long_term_cache[username] = player + self._cache[cache_key] = self._long_term_cache[cache_key] = player def get_cached_player( self, username: str, *, long_term: bool = False ) -> Player | None: + cache_key = username.lower() + with self._mutex: return ( - self._cache.get(username, None) + self._cache.get(cache_key, None) if not long_term - else self._long_term_cache.get(username, None) + else self._long_term_cache.get(cache_key, None) ) def update_cached_player( self, username: str, update: Callable[[KnownPlayer], KnownPlayer] ) -> None: """Update the cache for a player""" + cache_key = username.lower() + with self._mutex: - player = self._cache.get(username, None) + player = self._cache.get(cache_key, None) if isinstance(player, KnownPlayer): - self._cache[username] = self._long_term_cache[username] = update(player) + self._cache[cache_key] = self._long_term_cache[cache_key] = update( + player + ) else: logger.warning(f"Player {username} not found during update") def uncache_player(self, username: str) -> None: """Clear the cache entry for `username`""" + cache_key = username.lower() + with self._mutex: - self._cache.pop(username, None) - self._long_term_cache.pop(username, None) + self._cache.pop(cache_key, None) + self._long_term_cache.pop(cache_key, None) def clear_cache(self, *, short_term_only: bool = False) -> None: """Clear the entire player cache""" diff --git a/src/prism/overlay/rating.py b/src/prism/overlay/rating.py index 5381d01e..588c9dcb 100644 --- a/src/prism/overlay/rating.py +++ b/src/prism/overlay/rating.py @@ -3,7 +3,7 @@ from collections.abc import Set from typing import TYPE_CHECKING, Literal, assert_never -from prism.player import KnownPlayer, Player +from prism.player import KnownPlayer, Player, Tags if TYPE_CHECKING: # pragma: no coverage from prism.overlay.output.cells import ColumnName @@ -11,9 +11,26 @@ GamemodeName = Literal["overall", "solo", "doubles", "threes", "fours"] +def rate_tags(tags: Tags) -> int: + """Convert Tags to a numeric rating for sorting purposes""" + # Uses magic weights picked to ensure the correct order (see test) + rating = 0 + if tags.sniping == "high": + rating += 22 + elif tags.sniping == "medium": + rating += 10 + + if tags.cheating == "high": + rating += 13 + elif tags.cheating == "medium": + rating += 5 + + return rating + + def rate_player( player: Player, party_members: Set[str], column: "ColumnName", sort_ascending: bool -) -> tuple[bool, bool, str | int | float]: +) -> tuple[bool, bool, int, str | int | float]: """Used as a key function for sorting""" is_enemy = player.username not in party_members @@ -23,7 +40,12 @@ def rate_player( # If we added the username here we would get reverse alphabetical stat: int | float | None + tags_rating = 0 + if isinstance(player, KnownPlayer): + if player.tags is not None: + tags_rating = rate_tags(player.tags) + if column == "username": stat = 0 elif column == "stars": @@ -50,6 +72,9 @@ def rate_player( stat = player.stats.winstreak elif column == "sessiontime": stat = player.sessiontime_seconds + elif column == "tags": # pragma: no coverage + # NOTE: Should not be needed, and should be disabled in the UI + stat = tags_rating else: # pragma: no coverage assert_never(column) @@ -63,7 +88,7 @@ def rate_player( # Unknown players are always sorted last stat = 0 if column == "username" else float("-inf") - return (is_enemy, player.stats_unknown, stat) + return (is_enemy, player.stats_unknown, tags_rating, stat) def sort_players( diff --git a/src/prism/overlay/settings.py b/src/prism/overlay/settings.py index 42ce4003..dd27599a 100644 --- a/src/prism/overlay/settings.py +++ b/src/prism/overlay/settings.py @@ -117,7 +117,7 @@ class Settings: def sort_ascending(self) -> bool: """Return True if the sort order is ascending for the current sort column""" sort_order = self.sort_order # The column we sort on - if sort_order == "username": + if sort_order == "username" or sort_order == "tags": return False config = self.rating_configs.to_dict()[sort_order] diff --git a/src/prism/player.py b/src/prism/player.py index 4f508406..e8065d3e 100644 --- a/src/prism/player.py +++ b/src/prism/player.py @@ -3,6 +3,16 @@ GamemodeName = Literal["overall", "solo", "doubles", "threes", "fours"] +TagSeverity = Literal["none", "medium", "high"] + + +@dataclass(frozen=True, slots=True) +class Tags: + """Dataclass holding tags for a player""" + + sniping: TagSeverity + cheating: TagSeverity + class Winstreaks(TypedDict): """Dict holding winstreaks for each core gamemode""" @@ -60,6 +70,8 @@ class KnownPlayer: lastLogoutMs: int | None = field(default=None) nick: str | None = field(default=None) + tags: Tags | None = field(default=None) + @property def stats_unknown(self) -> bool: return False @@ -107,6 +119,10 @@ def update_winstreaks( ), ) + def set_tags(self, tags: Tags) -> Self: + """Set the tags for a player""" + return replace(self, tags=tags) + @dataclass(frozen=True, slots=True) class NickedPlayer: diff --git a/tests/prism/overlay/output/overlay/test_overlay.py b/tests/prism/overlay/output/overlay/test_overlay.py index f9b2e8bf..694efe5e 100644 --- a/tests/prism/overlay/output/overlay/test_overlay.py +++ b/tests/prism/overlay/output/overlay/test_overlay.py @@ -18,6 +18,7 @@ PendingPlayer, Player, Stats, + Tags, UnknownPlayer, ) @@ -76,6 +77,7 @@ beds=2, wins=1, ), + tags=Tags(sniping="medium", cheating="high"), ), ( None, @@ -93,6 +95,15 @@ beds=CellValue.monochrome("2", rating0), wins=CellValue.monochrome("1", rating0), sessiontime=CellValue.monochrome("<1m", rating4), + tags=CellValue( + text="C S", + color_sections=( + ColorSection(rating4, 1), # Cheating: high + ColorSection(GUI_COLORS[0], 1), # Space + ColorSection(rating3, 1), # Sniping: medium + ), + hover="C: Cheating (high)\nS: Sniping (medium)", + ), ), ), ), @@ -118,6 +129,7 @@ beds=8_000, wins=6_000, ), + tags=Tags(sniping="high", cheating="none"), ), ( "the_amazing_nick", @@ -135,6 +147,11 @@ beds=CellValue.monochrome("8000", rating2), wins=CellValue.monochrome("6000", rating3), sessiontime=CellValue.monochrome("-", rating4), + tags=CellValue( + text="S", + color_sections=(ColorSection(rating4, 1),), + hover="S: Sniping (high)", + ), ), ), ), @@ -156,6 +173,7 @@ beds=CellValue.monochrome("nick", rating4), wins=CellValue.monochrome("nick", rating4), sessiontime=CellValue.monochrome("nick", rating4), + tags=CellValue.monochrome("nick", rating4), ), ), ), @@ -177,6 +195,7 @@ beds=CellValue.monochrome("error", rating4), wins=CellValue.monochrome("error", rating4), sessiontime=CellValue.monochrome("error", rating4), + tags=CellValue.monochrome("error", rating4), ), ), ), @@ -198,10 +217,11 @@ beds=CellValue.monochrome("-", rating0), wins=CellValue.monochrome("-", rating0), sessiontime=CellValue.monochrome("-", rating0), + tags=CellValue.monochrome("-", rating0), ), ), ), - # Missing winstreak and sessiontime + # Missing winstreak, sessiontime, and tags ( KnownPlayer( dataReceivedAtMs=CURRENT_TIME_MS, @@ -223,6 +243,7 @@ beds=12_000, wins=15_000, ), + tags=None, # Request failed or still in flight ), ( None, @@ -240,6 +261,7 @@ beds=CellValue.monochrome("12000", rating3), wins=CellValue.monochrome("15000", rating4), sessiontime=CellValue.monochrome("-", rating4), + tags=CellValue.monochrome("-", rating0), ), ), ), @@ -265,6 +287,7 @@ beds=1900, wins=2000, ), + tags=Tags(sniping="none", cheating="high"), ), ( None, @@ -282,6 +305,11 @@ beds=CellValue.monochrome("1900", rating0), wins=CellValue.monochrome("2000", rating1), sessiontime=CellValue.monochrome("-", rating4), + tags=CellValue( + text="C", + color_sections=(ColorSection(rating4, 1),), + hover="C: Cheating (high)", + ), ), ), ), @@ -307,6 +335,7 @@ beds=3000, wins=4000, ), + tags=Tags(sniping="none", cheating="none"), ), ( None, @@ -324,6 +353,7 @@ beds=CellValue.monochrome("3000", rating1), wins=CellValue.monochrome("4000", rating2), sessiontime=CellValue.monochrome("-", rating4), + tags=CellValue.monochrome("", rating0), ), ), ), @@ -355,6 +385,7 @@ beds=CellValue.monochrome("0", rating0), wins=CellValue.monochrome("0", rating0), sessiontime=CellValue.monochrome("7m", rating3), + tags=CellValue.monochrome("-", rating0), ), ), ), @@ -386,6 +417,7 @@ beds=CellValue.monochrome("0", rating0), wins=CellValue.monochrome("0", rating0), sessiontime=CellValue.monochrome("-", rating4), + tags=CellValue.monochrome("-", rating0), ), ), ), diff --git a/tests/prism/overlay/output/test_cell_renderer.py b/tests/prism/overlay/output/test_cell_renderer.py index 5c8d0f8b..ec96ecc8 100644 --- a/tests/prism/overlay/output/test_cell_renderer.py +++ b/tests/prism/overlay/output/test_cell_renderer.py @@ -391,6 +391,7 @@ def test_render_stars( BEDS_VALUE = CellValue.monochrome("beds", gui_color=MinecraftColor.RED) WINS_VALUE = CellValue.monochrome("wins", gui_color=MinecraftColor.AQUA) SESSIONTIME_VALUE = CellValue.monochrome("sessiontime", gui_color=MinecraftColor.RED) +TAGS_VALUE = CellValue.monochrome("tags", gui_color=MinecraftColor.LIGHT_PURPLE) RENDERED_STATS = RenderedStats( username=USERNAME_VALUE, stars=STARS_VALUE, @@ -405,6 +406,7 @@ def test_render_stars( beds=BEDS_VALUE, wins=WINS_VALUE, sessiontime=SESSIONTIME_VALUE, + tags=TAGS_VALUE, ) @@ -435,6 +437,7 @@ def test_render_stars( BEDS_VALUE, WINS_VALUE, SESSIONTIME_VALUE, + TAGS_VALUE, ), ), ) @@ -570,3 +573,76 @@ def test_render_based_on_level_too_many_levels() -> None: assert render_based_on_level( "a", 100, (1, 2, 3, 4, 5, 6, 7, 8), True, sort_ascending=False ) == CellValue.monochrome("a", GUI_COLORS[4]) + + +def test_render_tags_none() -> None: + """Test render_tags with None (pending tags)""" + from prism.overlay.output.cell_renderer import render_tags + + result = render_tags(None) + assert result.text == "-" + assert result.hover is None + + +def test_render_tags_both_high() -> None: + """Test render_tags with both cheating and sniping high""" + from prism.overlay.output.cell_renderer import render_tags + from prism.player import Tags + + tags = Tags(sniping="high", cheating="high") + result = render_tags(tags) + assert result.text == "C S" + assert result.hover == "C: Cheating (high)\nS: Sniping (high)" + + +def test_render_tags_only_cheating() -> None: + """Test render_tags with only cheating tag""" + from prism.overlay.output.cell_renderer import render_tags + from prism.player import Tags + + tags = Tags(sniping="none", cheating="medium") + result = render_tags(tags) + assert result.text == "C" + assert result.hover == "C: Cheating (medium)" + + +def test_render_tags_only_sniping() -> None: + """Test render_tags with only sniping tag""" + from prism.overlay.output.cell_renderer import render_tags + from prism.player import Tags + + tags = Tags(sniping="high", cheating="none") + result = render_tags(tags) + assert result.text == "S" + assert result.hover == "S: Sniping (high)" + + +def test_render_tags_none_severity() -> None: + """Test render_tags with both tags having none severity""" + from prism.overlay.output.cell_renderer import render_tags + from prism.player import Tags + + tags = Tags(sniping="none", cheating="none") + result = render_tags(tags) + assert result.text == "" + assert result.hover is None + + +def test_cellvalue_with_hover() -> None: + """Test CellValue with hover text""" + cell = CellValue( + text="Test", + color_sections=(ColorSection("red", 4),), + hover="Hover text", + ) + assert cell.text == "Test" + assert cell.hover == "Hover text" + + +def test_cellvalue_monochrome_with_hover() -> None: + """Test CellValue.monochrome with hover text""" + cell = CellValue.monochrome("Test", "red", hover="Hover text") + assert cell.text == "Test" + assert cell.hover == "Hover text" + assert len(cell.color_sections) == 1 + assert cell.color_sections[0].color == "red" diff --git a/tests/prism/overlay/test_behaviour.py b/tests/prism/overlay/test_behaviour.py index 55457dc1..6a5e45c6 100644 --- a/tests/prism/overlay/test_behaviour.py +++ b/tests/prism/overlay/test_behaviour.py @@ -6,7 +6,7 @@ import pytest -from prism.errors import PlayerNotFoundError +from prism.errors import APIError, PlayerNotFoundError from prism.overlay.behaviour import ( autodenick_teammate, bedwars_game_ended, @@ -19,7 +19,7 @@ from prism.overlay.keybinds import AlphanumericKeyDict from prism.overlay.nick_database import NickDatabase from prism.overlay.settings import NickValue, Settings, SettingsDict, get_settings -from prism.player import MISSING_WINSTREAKS, KnownPlayer, Winstreaks +from prism.player import MISSING_WINSTREAKS, KnownPlayer, Tags, Winstreaks from tests.prism.overlay import test_get_stats from tests.prism.overlay.test_settings import ( DEFAULT_STATS_THREAD_COUNT, @@ -29,6 +29,7 @@ CUSTOM_RATING_CONFIG_COLLECTION_DICT, MockedAccountProvider, MockedPlayerProvider, + MockedTagsProvider, MockedWinstreakProvider, create_controller, create_state, @@ -266,12 +267,22 @@ def get_player(uuid: str, user_id: str) -> KnownPlayer: return user.player + def get_tags(uuid: str, user_id: str, urchin_api_key: str | None) -> Tags: + assert uuid == user.uuid + assert urchin_api_key is None + + return Tags( + cheating="none", + sniping="none", + ) + controller = create_controller( account_provider=MockedAccountProvider(get_uuid_for_username=get_uuid), player_provider=MockedPlayerProvider(get_player=get_player), winstreak_provider=MockedWinstreakProvider( get_estimated_winstreaks_for_uuid=get_estimated_winstreaks ), + tags_provider=MockedTagsProvider(get_tags=get_tags), nick_database=NickDatabase([{user.nick: user.uuid}]), settings=make_settings( use_antisniper_api=True, # Always enable antisniper API for testing @@ -286,6 +297,9 @@ def get_player(uuid: str, user_id: str) -> KnownPlayer: # One update for getting the stats assert completed_queue.get_nowait() == user.nick + # One update for getting the tags + assert completed_queue.get_nowait() == user.nick + # One update for getting estimated winstreaks - only when missing + gotten if not player_has_winstreaks and estimated_winstreaks_from_provider: assert completed_queue.get_nowait() == user.nick @@ -294,6 +308,62 @@ def get_player(uuid: str, user_id: str) -> KnownPlayer: completed_queue.get_nowait() +def test_get_and_cache_stats_tags_error() -> None: + user = test_get_stats.users["NickedPlayer"] + assert user.nick is not None # For typing + + def get_estimated_winstreaks( + uuid: str, antisniper_api_key: str + ) -> tuple[Winstreaks, bool]: + assert uuid == user.uuid + return (MISSING_WINSTREAKS, False) + + def get_uuid(username: str) -> str: + if username == user.nick: + raise PlayerNotFoundError + + assert username == user.username + return user.uuid + + def get_player(uuid: str, user_id: str) -> KnownPlayer: + assert uuid == user.uuid + assert user.player is not None + + return user.player + + def get_tags(uuid: str, user_id: str, urchin_api_key: str | None) -> Tags: + assert uuid == user.uuid + assert urchin_api_key is None + + raise APIError("API failure") + + controller = create_controller( + account_provider=MockedAccountProvider(get_uuid_for_username=get_uuid), + player_provider=MockedPlayerProvider(get_player=get_player), + winstreak_provider=MockedWinstreakProvider( + get_estimated_winstreaks_for_uuid=get_estimated_winstreaks + ), + tags_provider=MockedTagsProvider(get_tags=get_tags), + nick_database=NickDatabase([{user.nick: user.uuid}]), + settings=make_settings( + use_antisniper_api=True, # Always enable antisniper API for testing + antisniper_api_key="test_key", + ), + ) + + completed_queue = queue.Queue[str]() + + get_and_cache_player(user.nick, completed_queue, controller) + + # One update for getting the stats + assert completed_queue.get_nowait() == user.nick + + # No update for winstreak + # No update since we failed getting tags + with pytest.raises(queue.Empty): + completed_queue.get_nowait() + + def test_update_settings_nothing() -> None: settings_file = no_close(io.StringIO()) controller = create_controller( diff --git a/tests/prism/overlay/test_controller.py b/tests/prism/overlay/test_controller.py index b5902c0e..a1e411a3 100644 --- a/tests/prism/overlay/test_controller.py +++ b/tests/prism/overlay/test_controller.py @@ -4,11 +4,12 @@ from prism.errors import APIError, APIKeyError, APIThrottleError, PlayerNotFoundError from prism.overlay.controller import ERROR_DURING_PROCESSING -from prism.player import MISSING_WINSTREAKS, KnownPlayer, Stats, Winstreaks +from prism.player import MISSING_WINSTREAKS, KnownPlayer, Stats, Tags, Winstreaks from prism.ssl_errors import MissingLocalIssuerSSLError from tests.prism.overlay.utils import ( MockedAccountProvider, MockedPlayerProvider, + MockedTagsProvider, MockedWinstreakProvider, assert_not_called, create_controller, @@ -125,6 +126,35 @@ def mock_get_player(uuid: str, user_id: str) -> KnownPlayer: assert not controller.missing_local_issuer_certificate +def test_overlay_controller_get_tags() -> None: + error: Exception | None = None + returned_tags = Tags(sniping="medium", cheating="none") + + def get_tags_mock(username: str, user_id: str, urchin_api_key: str | None) -> Tags: + assert username == "username" + assert user_id == "1234" + assert urchin_api_key is None + if error: + raise error + + return returned_tags + + controller = create_controller( + settings=make_settings( + user_id="1234", + ), + tags_provider=MockedTagsProvider(get_tags=get_tags_mock), + ) + + error = APIError() + uuid = controller.get_tags("username") + assert uuid is ERROR_DURING_PROCESSING + + error = None + uuid = controller.get_tags("username") + assert uuid == returned_tags + + def test_overlay_controller_get_estimated_winstreaks_success() -> None: """Test that OverlayController uses injected get_estimated_winstreaks function""" diff --git a/tests/prism/overlay/test_player.py b/tests/prism/overlay/test_player.py index c8269fdf..8095330f 100644 --- a/tests/prism/overlay/test_player.py +++ b/tests/prism/overlay/test_player.py @@ -6,7 +6,7 @@ from prism.hypixel import create_known_player from prism.overlay.output.cells import ColumnName from prism.overlay.rating import sort_players -from prism.player import KnownPlayer, Player, Stats, Winstreaks +from prism.player import KnownPlayer, Player, Stats, Tags, Winstreaks from tests.prism.overlay.utils import make_player CURRENT_TIME_MS = 1234567890123 @@ -99,6 +99,31 @@ def make_winstreaks( lastLogoutMs=0, lastLoginMs=3000, ), + "untagged_5_star": make_player( + username="untagged", + stars=5, + tags=None, + ), + "untagged_4_star": make_player( + username="untagged", + stars=4, + tags=None, + ), + "h_snipe_m_cheat": make_player( + username="h_snipe_m_cheat", + stars=3, + tags=Tags(sniping="high", cheating="medium"), + ), + "m_snipe_m_cheat": make_player( + username="m_snipe_m_cheat", + stars=2, + tags=Tags(sniping="medium", cheating="medium"), + ), + "m_snipe_h_cheat": make_player( + username="m_snipe_h_cheat", + stars=1, + tags=Tags(sniping="medium", cheating="high"), + ), } @@ -550,6 +575,32 @@ def test_update_winstreaks( players["jonathan"], ], ), + ( + [ + players["untagged_5_star"], + players["untagged_4_star"], + players["h_snipe_m_cheat"], + players["m_snipe_m_cheat"], + players["m_snipe_h_cheat"], + ], + set(), + "stars", + [ + players["h_snipe_m_cheat"], + players["m_snipe_h_cheat"], + players["m_snipe_m_cheat"], + players["untagged_5_star"], + players["untagged_4_star"], + ], + [ + # Tags always sorted descending, so only the stars are ascending + players["h_snipe_m_cheat"], + players["m_snipe_h_cheat"], + players["m_snipe_m_cheat"], + players["untagged_4_star"], + players["untagged_5_star"], + ], + ), ) diff --git a/tests/prism/overlay/test_rating.py b/tests/prism/overlay/test_rating.py new file mode 100644 index 00000000..e53a1e0f --- /dev/null +++ b/tests/prism/overlay/test_rating.py @@ -0,0 +1,19 @@ +from prism.overlay.rating import rate_tags +from prism.player import Tags + + +def test_rate_tags() -> None: + # Assert the possible tags are ordered in the specified order + order = ( + Tags(sniping="high", cheating="high"), + Tags(sniping="high", cheating="medium"), + Tags(sniping="medium", cheating="high"), + Tags(sniping="high", cheating="none"), + Tags(sniping="medium", cheating="medium"), + Tags(sniping="none", cheating="high"), + Tags(sniping="medium", cheating="none"), + Tags(sniping="none", cheating="medium"), + Tags(sniping="none", cheating="none"), + ) + + assert sorted(reversed(order), key=rate_tags, reverse=True) == list(order) diff --git a/tests/prism/overlay/test_settings.py b/tests/prism/overlay/test_settings.py index 1a539c39..9a090913 100644 --- a/tests/prism/overlay/test_settings.py +++ b/tests/prism/overlay/test_settings.py @@ -89,7 +89,15 @@ def make_settings_dict( "sort_order": value_or_default(sort_order, default="index"), "column_order": value_or_default( column_order, - default=("username", "stars", "fkdr", "kdr", "winstreak", "sessiontime"), + default=( + "username", + "stars", + "fkdr", + "kdr", + "winstreak", + "sessiontime", + "tags", + ), ), "rating_configs": value_or_default( rating_configs, default=DEFAULT_RATING_CONFIG_COLLECTION_DICT diff --git a/tests/prism/overlay/utils.py b/tests/prism/overlay/utils.py index 8b8ec21d..76b0b676 100644 --- a/tests/prism/overlay/utils.py +++ b/tests/prism/overlay/utils.py @@ -7,6 +7,7 @@ AccountProvider, OverlayController, PlayerProvider, + TagsProvider, WinstreakProvider, ) from prism.overlay.keybinds import Key @@ -25,6 +26,7 @@ PendingPlayer, Player, Stats, + Tags, UnknownPlayer, Winstreaks, ) @@ -176,6 +178,7 @@ def make_player( lastLoginMs: int | None = ..., lastLogoutMs: int | None = ..., dataReceivedAtMs: int = ..., + tags: Tags | None = None, ) -> KnownPlayer: ... @@ -198,6 +201,7 @@ def make_player( lastLoginMs: int | None = None, lastLogoutMs: int | None = None, dataReceivedAtMs: int = 1234567890, + tags: Tags | None = Tags(sniping="none", cheating="none"), ) -> Player: if variant == "player": return KnownPlayer( @@ -221,6 +225,7 @@ def make_player( uuid=uuid, lastLoginMs=lastLoginMs, lastLogoutMs=lastLogoutMs, + tags=tags, ) elif variant == "unknown": return UnknownPlayer(username) @@ -369,6 +374,31 @@ def seconds_until_unblocked(self) -> float: return self._seconds_until_unblocked +class MockedTagsProvider: + def __init__( + self, + get_tags: Callable[ + [str, str, str | None], + Tags, + ], + seconds_until_unblocked: float = 0.0, + ) -> None: + self._get_tags = get_tags + self._seconds_until_unblocked = seconds_until_unblocked + + def get_tags( + self, + uuid: str, + user_id: str, + urchin_api_key: str | None = None, + ) -> Tags: + return self._get_tags(uuid, user_id, urchin_api_key) + + @property + def seconds_until_unblocked(self) -> float: + return self._seconds_until_unblocked + + def create_controller( state: OverlayState | None = None, settings: Settings | None = None, @@ -386,6 +416,7 @@ def create_controller( account_provider: AccountProvider = MockedAccountProvider(assert_not_called), player_provider: PlayerProvider = MockedPlayerProvider(assert_not_called), winstreak_provider: WinstreakProvider = MockedWinstreakProvider(assert_not_called), + tags_provider: TagsProvider = MockedTagsProvider(assert_not_called), ) -> OverlayController: controller = OverlayController( state=state or create_state(), @@ -394,6 +425,7 @@ def create_controller( account_provider=account_provider, player_provider=player_provider, winstreak_provider=winstreak_provider, + tags_provider=tags_provider, ) controller.wants_shown = wants_shown diff --git a/tests/prism/test_flashlight.py b/tests/prism/test_flashlight.py new file mode 100644 index 00000000..8b77e1c0 --- /dev/null +++ b/tests/prism/test_flashlight.py @@ -0,0 +1,122 @@ +import pytest + +from prism.errors import APIError +from prism.flashlight import ( + FlashlightTagsProvider, + parse_flashlight_tags, + validate_tag_severity, +) +from prism.player import Tags, TagSeverity + + +def test_make_tags_provider() -> None: + provider = FlashlightTagsProvider(retry_limit=3, initial_timeout=1.0) + assert provider.seconds_until_unblocked == 0.0 + + +@pytest.mark.parametrize( + "raw_severity, validated", + ( + ("none", "none"), + ("medium", "medium"), + ("high", "high"), + ("invalid", None), + (None, None), + (123, None), + ), +) +def test_validate_tag_severity( + raw_severity: object, validated: TagSeverity | None +) -> None: + assert validate_tag_severity(raw_severity) == validated + + +@pytest.mark.parametrize( + "response_json, tags", + # Real responses from Flashlight API with anonymized UUIDs + ( + ( + { + "uuid": "01234567-89ab-cdef-0123-456789abcdef", + "tags": {"cheating": "medium", "sniping": "none"}, + }, + Tags( + cheating="medium", + sniping="none", + ), + ), + ( + { + "uuid": "01234567-89ab-cdef-0123-456789abcdef", + "tags": {"cheating": "none", "sniping": "none"}, + }, + Tags( + cheating="none", + sniping="none", + ), + ), + ( + # Made up response - all high + { + "uuid": "01234567-89ab-cdef-0123-456789abcdef", + "tags": {"cheating": "high", "sniping": "high"}, + }, + Tags( + cheating="high", + sniping="high", + ), + ), + ), +) +def test_parse_flashlight_tags(response_json: str, tags: Tags) -> None: + assert parse_flashlight_tags(response_json) == tags + + +@pytest.mark.parametrize( + "response_json, error", + ( + ( + # Made up response - empty tags + { + "uuid": "some-uuid", + "tags": {}, + }, + APIError, + ), + ( + # Made up response - missing cheating + { + "uuid": "some-uuid", + "tags": { + "sniping": "high", + }, + }, + APIError, + ), + ( + # Made up response - missing sniping + { + "uuid": "some-uuid", + "tags": { + "cheating": "high", + }, + }, + APIError, + ), + ( + # Made up response - bad type on tags + {"uuid": "some-uuid", "tags": "123"}, + APIError, + ), + ( + # Made up response - bad type on tags + {"uuid": "some-uuid", "tags": None}, + APIError, + ), + ), +) +def test_parse_flashlight_tags_error( + response_json: str, error: type[Exception] +) -> None: + with pytest.raises(error): + parse_flashlight_tags(response_json)