feat(config): warn with the lines of a duplicate key in any config load

docs/guide/cli-reference.md

@@ -339,7 +339,7 @@ unshackle cfg [KEY] [VALUE] [OPTIONS]
 | `--list` | List all set configuration values. |
 
 !!! warning "Comments are stripped"
-    Writing a value through `cfg` rewrites the configuration file and **removes all comments** from it. If your config relies on comments, edit it by hand instead. Setting a value and using `--unset` together is an error. If a config key is set twice in the same block, `cfg` stops and gives the line numbers of the two copies. Other commands use the later copy.
+    Writing a value through `cfg` rewrites the configuration file and **removes all comments** from it. If your config relies on comments, edit it by hand instead. Setting a value and using `--unset` together is an error. If a config key is set twice in the same block, `cfg` stops and gives the line numbers of the two copies. Other commands warn with the same line numbers and use the later copy.
 
 !!! example
     ```shell

tests/commands/test_cfg.py

@@ -45,13 +45,6 @@ def test_a_valid_config_still_reads_a_key(config_file: Path) -> None:
     assert "rainbow" in result.output
 
 
-def test_duplicate_key_lines_follows_aliases_without_looping() -> None:
-    from ruamel.yaml import YAML
-
-    root = YAML().compose("base: &b\n  x: 1\n  x: 2\n  self: *b\nother: *b\nlist:\n  - y: 1\n    y: 2\n")
-    assert cfg_module.duplicate_key_lines(root) == [(2, 3), (7, 8)]
-
-
 def test_a_duplicate_only_ruamel_sees_still_gives_its_line(config_file: Path) -> None:
     config_file.write_text("a: 1\n1: x\n0x1: y\n", encoding="utf-8")
     result = CliRunner().invoke(cfg_module.cfg, ["a"])

tests/core/test_config_duplicate_keys.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import warnings
+from pathlib import Path
+
+import pytest
+import yaml
+from ruamel.yaml import YAML
+
+from unshackle.core.config import Config, duplicate_key_lines, load_yaml
+
+ALIASES = "base: &b\n  x: 1\n  x: 2\n  self: *b\nother: *b\nlist:\n  - y: 1\n    y: 2\n"
+
+
+@pytest.mark.parametrize("compose", [yaml.compose, YAML().compose], ids=["pyyaml", "ruamel"])
+def test_duplicate_key_lines_follows_aliases_without_looping(compose) -> None:
+    assert duplicate_key_lines(compose(ALIASES)) == [(2, 3), (7, 8)]
+
+
+def test_load_yaml_warns_with_lines_and_keeps_the_later_copy(tmp_path: Path) -> None:
+    path = tmp_path / "unshackle.yaml"
+    path.write_text("credentials:\n  EXAMPLE: first-secret\n  EXAMPLE: second-secret\n", encoding="utf-8")
+    with pytest.warns(UserWarning) as caught:
+        data = load_yaml(path)
+    assert data == {"credentials": {"EXAMPLE": "second-secret"}}
+    message = str(caught[0].message)
+    assert message == f"{path}: duplicate key on lines 2 and 3. Unshackle uses the later one."
+    assert "EXAMPLE" not in message and "secret" not in message
+
+
+def test_config_from_yaml_warns_on_a_duplicate_key(tmp_path: Path) -> None:
+    path = tmp_path / "unshackle.yaml"
+    path.write_text("tag: first\ntag: second\n", encoding="utf-8")
+    with pytest.warns(UserWarning, match="duplicate key on lines 1 and 2"):
+        assert Config.from_yaml(path).tag == "second"
+
+
+@pytest.mark.parametrize(
+    "text",
+    ["a: 1\nb: [1, 2]\n", "", "a: &a\n  x: 1\nb: &b\n  y: 2\nc:\n  <<: *a\n  <<: *b\n"],
+    ids=["valid", "empty", "two-merges"],
+)
+def test_load_yaml_matches_safe_load_without_a_warning(tmp_path: Path, text: str) -> None:
+    path = tmp_path / "config.yaml"
+    path.write_text(text, encoding="utf-8")
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        assert load_yaml(path) == yaml.safe_load(text)

unshackle/commands/cfg.py

@@ -5,36 +5,11 @@ import sys
 import click
 from ruamel.yaml import YAML
 from ruamel.yaml.constructor import DuplicateKeyError
-from ruamel.yaml.nodes import MappingNode, Node, ScalarNode, SequenceNode
 
-from unshackle.core.config import config, get_config_path
+from unshackle.core.config import config, duplicate_key_lines, duplicate_key_message, get_config_path
 from unshackle.core.constants import context_settings
 
 
-def duplicate_key_lines(root: Node) -> list[tuple[int, int]]:
-    """Return (first line, repeat line), counted from 1, for each repeated scalar key in the node tree."""
-    pairs, stack, visited = [], [root], set()
-    while stack:
-        node = stack.pop()
-        if node in visited:  # an alias reuses its node, and can point back at itself
-            continue
-        visited.add(node)
-        if isinstance(node, MappingNode):
-            first: dict[tuple[str, str], int] = {}
-            for key_node, value_node in node.value:
-                if isinstance(key_node, ScalarNode):
-                    key = (key_node.tag, key_node.value)
-                    line = key_node.start_mark.line + 1
-                    if key in first:
-                        pairs.append((first[key], line))
-                    else:
-                        first[key] = line
-                stack += [key_node, value_node]
-        elif isinstance(node, SequenceNode):
-            stack += node.value
-    return sorted(pairs)
-
-
 @click.command(
     short_help="Manage configuration values for the program and its services.", context_settings=context_settings
 )
@@ -71,10 +46,8 @@ def cfg(ctx: click.Context, key: str, value: str, unset: bool, list_: bool) -> N
             data = yaml.load(config_path)
         except DuplicateKeyError as e:
             pairs = duplicate_key_lines(yaml.compose(config_path))
-            where = ", ".join(f"lines {first} and {repeat}" for first, repeat in pairs)
-            where = where or f"line {e.problem_mark.line + 1}"
             raise click.ClickException(
-                f"{config_path.name}: duplicate key on {where}. Unshackle uses the later one."
+                duplicate_key_message(config_path.name, pairs, e.problem_mark.line + 1)
             ) from None
 
     if not data:

unshackle/commands/dl.py

@@ -27,7 +27,6 @@ from uuid import UUID
 
 import click
 import mediaexport
-import yaml
 from click.core import ParameterSource
 from langcodes import Language, tag_is_valid
 from pymediainfo import MediaInfo
@@ -43,7 +42,7 @@ from rich.tree import Tree
 from unshackle.core import __version__, binaries, providers
 from unshackle.core.cdm import DecryptLabsRemoteCDM
 from unshackle.core.cdm.detect import cdm_type_stub, is_playready_cdm, is_remote_cdm, is_widevine_cdm
-from unshackle.core.config import cdm_entry_names_device, config, resolve_cdm_name, resolve_decryption
+from unshackle.core.config import cdm_entry_names_device, config, load_yaml, resolve_cdm_name, resolve_decryption
 from unshackle.core.console import GradientPulseBarColumn, SyncLive, console, listing_panel
 from unshackle.core.constants import (
     DEFAULT_SAMPLE,
@@ -1454,7 +1453,7 @@ class dl:
                 try:
                     service_config_path = Services.get_path(self.service) / config.filenames.config
                     if service_config_path.exists():
-                        self.service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8"))
+                        self.service_config = load_yaml(service_config_path)
                         self.log.info("Service Config loaded")
                         # log key names only: the full config carries service certificates,
                         # device fingerprints and endpoints that bloat the log and may be sensitive

unshackle/commands/search.py

@@ -6,13 +6,12 @@ import sys
 from typing import Any, Optional
 
 import click
-import yaml
 from rich.padding import Padding
 from rich.rule import Rule
 from rich.tree import Tree
 
 from unshackle.commands.dl import dl
-from unshackle.core.config import config
+from unshackle.core.config import config, load_yaml
 from unshackle.core.console import console
 from unshackle.core.constants import context_settings
 from unshackle.core.proxies.proxy import Proxy
@@ -91,7 +90,7 @@ def search(
         with console.status("Loading Service Config...", spinner="dots"):
             service_config_path = Services.get_path(service) / config.filenames.config
             if service_config_path.exists():
-                service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8"))
+                service_config = load_yaml(service_config_path)
                 log.info("Service Config loaded")
     merge_dict(ci_get(config.services, service), service_config)
 

unshackle/core/api/download_manager.py

@@ -371,11 +371,10 @@ def perform_download(
 
     # Import dl.py components lazily to avoid circular deps during module import
     import click
-    import yaml
 
     from unshackle.commands.dl import dl
     from unshackle.core.api.errors import APIError, APIErrorCode
-    from unshackle.core.config import config
+    from unshackle.core.config import config, load_yaml
     from unshackle.core.services import Services
     from unshackle.core.tracks import Video
     from unshackle.core.utils.click_types import ContextData
@@ -467,7 +466,7 @@ def perform_download(
     remote = params.get("remote")
     service_config_path = None if remote else Services.get_path(service) / config.filenames.config
     if service_config_path and service_config_path.exists():
-        service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8"))
+        service_config = load_yaml(service_config_path)
     else:
         service_config = {}
     merge_dict(config.services.get(service), service_config)

unshackle/core/api/handlers.py

@@ -28,7 +28,7 @@ from unshackle.core.api.session_log import SessionLogBuffer, SessionLogMirror, c
 from unshackle.core.api.session_store import SessionStore
 from unshackle.core.cacher import Cacher
 from unshackle.core.cdm.detect import cdm_type_stub
-from unshackle.core.config import config
+from unshackle.core.config import config, load_yaml
 from unshackle.core.constants import AUDIO_CODEC_MAP, DYNAMIC_RANGE_MAP, VIDEO_CODEC_MAP
 from unshackle.core.drm import cbcs
 from unshackle.core.providers.anilist import parse_anilist_ref
@@ -190,13 +190,11 @@ def load_full_cdm(service: str, profile: Optional[str], cdm_type: Optional[str]
 
 def load_service_yaml(normalized_service: str) -> dict:
     """Load a service's config.yaml and merge it with the global override block."""
-    import yaml
-
     from unshackle.core.utils.collections import merge_dict
 
     service_config_path = Services.get_path(normalized_service) / config.filenames.config
     if service_config_path.exists():
-        service_config = yaml.safe_load(service_config_path.read_text(encoding="utf8")) or {}
+        service_config = load_yaml(service_config_path) or {}
     else:
         service_config = {}
     merge_dict(config.services.get(normalized_service), service_config)

unshackle/core/config.py

@@ -374,7 +374,55 @@ class Config:
             raise FileNotFoundError(f"Config file path ({path}) was not found")
         if not path.is_file():
             raise FileNotFoundError(f"Config file path ({path}) is not to a file.")
-        return cls(**restore_bool_languages(yaml.safe_load(path.read_text(encoding="utf8")) or {}))
+        return cls(**restore_bool_languages(load_yaml(path) or {}))
+
+
+def duplicate_key_lines(root: Any) -> list[tuple[int, int]]:
+    """
+    Return (first line, repeat line), counted from 1, for each repeated scalar key in a PyYAML or ruamel node tree.
+
+    Keys compare as written, so two spellings that load as the same key (yes and true, 1 and 0x1) are not found.
+    """
+    pairs, stack, visited = [], [root], set()
+    while stack:
+        node = stack.pop()
+        if node in visited:
+            continue
+        visited.add(node)
+        if node.id == "mapping":
+            first: dict[tuple[str, str], int] = {}
+            for key_node, value_node in node.value:
+                if key_node.id == "scalar" and key_node.tag != "tag:yaml.org,2002:merge":
+                    key = (key_node.tag, key_node.value)
+                    line = key_node.start_mark.line + 1
+                    if key in first:
+                        pairs.append((first[key], line))
+                    else:
+                        first[key] = line
+                stack += [key_node, value_node]
+        elif node.id == "sequence":
+            stack += node.value
+    return sorted(pairs)
+
+
+def duplicate_key_message(name: str, pairs: list[tuple[int, int]], line: int = 0) -> str:
+    """Say where a file repeats a key: each (first, repeat) pair, or `line` when no pair is known."""
+    where = ", ".join(f"lines {first} and {repeat}" for first, repeat in pairs) or f"line {line}"
+    return f"{name}: duplicate key on {where}. Unshackle uses the later one."
+
+
+def load_yaml(path: Path) -> Any:
+    """yaml.safe_load a file, and warn with the lines of each duplicate key, which PyYAML takes silently."""
+    loader = yaml.SafeLoader(path.read_text(encoding="utf8"))
+    try:
+        node = loader.get_single_node()
+        if node is None:
+            return None
+        if pairs := duplicate_key_lines(node):
+            warnings.warn(duplicate_key_message(str(path), pairs), stacklevel=2)
+        return loader.construct_document(node)
+    finally:
+        loader.dispose()
 
 
 # noinspection PyProtectedMember