music-assistant-server

1"""Tests for the sendspin server identity persistence."""
2
3from __future__ import annotations
4
5import os
6import stat
7from typing import TYPE_CHECKING
8
9import pytest
10from aiosendspin.noise.keys import Identity
11
12from music_assistant.providers.sendspin.security import (
13    IDENTITY_FILENAME,
14    get_or_create_server_identity,
15)
16
17if TYPE_CHECKING:
18    from pathlib import Path
19
20
21def test_identity_created_with_restrictive_permissions(tmp_path: Path) -> None:
22    """A fresh identity is generated with a 0600 key file."""
23    storage_dir = tmp_path / "sendspin"
24    identity = get_or_create_server_identity(storage_dir)
25    assert len(identity.peer_id) == 43
26    assert stat.S_IMODE((storage_dir / IDENTITY_FILENAME).stat().st_mode) == 0o600
27
28
29def test_identity_is_persistent(tmp_path: Path) -> None:
30    """Repeated calls return the same identity."""
31    storage_dir = tmp_path / "sendspin"
32    first = get_or_create_server_identity(storage_dir)
33    second = get_or_create_server_identity(storage_dir)
34    assert first == second
35
36
37def test_identity_concurrent_creation_returns_existing(
38    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
39) -> None:
40    """A key file appearing between the read and the create is loaded, not clobbered."""
41    storage_dir = tmp_path / "sendspin"
42    winner = Identity.generate()
43    real_open = os.open
44
45    def racing_open(path: str, flags: int, mode: int = 0o777) -> int:
46        (storage_dir / IDENTITY_FILENAME).write_text(winner.private_b64u)
47        return real_open(path, flags, mode)
48
49    monkeypatch.setattr(os, "open", racing_open)
50    assert get_or_create_server_identity(storage_dir) == winner
51
52
53def test_identity_corrupt_file_raises(tmp_path: Path) -> None:
54    """A corrupt key file raises instead of silently minting a new identity."""
55    storage_dir = tmp_path / "sendspin"
56    get_or_create_server_identity(storage_dir)
57    key_path = storage_dir / IDENTITY_FILENAME
58    key_path.write_text("garbage")
59    with pytest.raises(ValueError, match="32 bytes"):
60        get_or_create_server_identity(storage_dir)
61    assert key_path.read_text() == "garbage"
62