aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/tests/test_enrich_audio.py
blob: 62d0a0a1f0f710b5f48d6eb8a81c3b8aec1fa703 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
"""Tests for indexer/enrich_audio.py — tag/cover extraction and the end-to-end pool."""

import asyncio
import shutil
import subprocess
from pathlib import Path

import pytest

from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.enrich_audio import (
    AudioEnricher, _artist_album_from_ancestors, _extract_cover,
)
from meshbay_node.indexer.title_parse import parse_track_filename
from meshbay_node.media_cache import MediaCache

_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe")


# ── pure helpers, no ffmpeg/mutagen file needed ──────────────────────────────

def test_parse_track_filename_splits_leading_track_number():
    parsed = parse_track_filename("01 - Venus As A Boy (Edited Lp Version).mp3")
    assert parsed.track_no == 1
    assert parsed.title == "Venus As A Boy (Edited Lp Version)"


def test_parse_track_filename_handles_dot_separated():
    parsed = parse_track_filename("03. Human Behaviour.mp3")
    assert parsed.track_no == 3
    assert parsed.title == "Human Behaviour"


def test_parse_track_filename_handles_underscore_separated():
    parsed = parse_track_filename("12_Some_Title.mp3")
    assert parsed.track_no == 12
    assert parsed.title == "Some Title"


def test_parse_track_filename_no_prefix_leaves_track_no_none():
    parsed = parse_track_filename("Some Title.mp3")
    assert parsed.track_no is None
    assert parsed.title == "Some Title"


def test_parse_track_filename_does_not_mistake_a_leading_year_for_a_track_number():
    parsed = parse_track_filename("1999 - Some Title.mp3")
    assert parsed.track_no is None, "a 4-digit prefix is capped out, not read as track 199"


def test_artist_album_from_ancestors_reads_artist_album_track_layout(tmp_path):
    folder = tmp_path / "Some Artist" / "Some Album"
    folder.mkdir(parents=True)
    track = folder / "01 - A Track.mp3"
    track.touch()

    artist, album = _artist_album_from_ancestors(track)

    assert artist == "Some Artist"
    assert album == "Some Album"


# ── end-to-end against a real (tiny, synthetic) MP3 file ────────────────────

pytestmark_ffmpeg = pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed")


def _make_clip(path: Path, *, title=None, artist=None, album=None, track=None) -> None:
    subprocess.run(
        ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y",
         "-f", "lavfi", "-i", "sine=frequency=440:duration=1",
         "-c:a", "libmp3lame", "-b:a", "64k",
         *(["-metadata", f"title={title}"] if title else []),
         *(["-metadata", f"artist={artist}"] if artist else []),
         *(["-metadata", f"album={album}"] if album else []),
         *(["-metadata", f"track={track}"] if track else []),
         str(path)],
        check=True, capture_output=True,
    )


@pytest.fixture
async def media_cache(tmp_path):
    c = MediaCache(db_path=tmp_path / "media_cache.db")
    await c.open()
    yield c
    await c.close()


@pytestmark_ffmpeg
@pytest.mark.asyncio
async def test_enricher_prefers_tags_over_filename_parse(tmp_path, media_cache):
    clip = tmp_path / "99 - wrong title.mp3"
    _make_clip(clip, title="Real Title", artist="Real Artist", album="Real Album", track=3)
    entry = IndexEntry(id="fileid1", name=clip.name, path=clip.name,
                        size=clip.stat().st_size, type="audio", added_at=0)

    enricher = AudioEnricher(media_cache)
    done = asyncio.get_event_loop().create_future()

    async def on_done(file_id, fields):
        done.set_result((file_id, fields))

    enricher.spawn(entry, clip, on_done)
    file_id, fields = await asyncio.wait_for(done, timeout=30)

    assert file_id == "fileid1"
    assert fields["display_title"] == "Real Title"
    assert fields["artist"] == "Real Artist"
    assert fields["album"] == "Real Album"
    assert fields["track_no"] == 3
    assert fields["duration"] == 1
    assert fields.get("thumb_hash") is None, "no embedded cover was written in this clip"


@pytestmark_ffmpeg
@pytest.mark.asyncio
async def test_enricher_falls_back_to_filename_and_folder_when_tags_absent(tmp_path, media_cache):
    folder = tmp_path / "Folder Artist" / "Folder Album"
    folder.mkdir(parents=True)
    clip = folder / "05 - Filename Title.mp3"
    _make_clip(clip)   # no metadata tags at all
    entry = IndexEntry(id="fileid2", name=clip.name,
                        path=str(clip.relative_to(tmp_path)),
                        size=clip.stat().st_size, type="audio", added_at=0)

    enricher = AudioEnricher(media_cache)
    done = asyncio.get_event_loop().create_future()

    async def on_done(file_id, fields):
        done.set_result((file_id, fields))

    enricher.spawn(entry, clip, on_done)
    _, fields = await asyncio.wait_for(done, timeout=30)

    assert fields["display_title"] == "Filename Title"
    assert fields["track_no"] == 5
    assert fields["artist"] == "Folder Artist"
    assert fields["album"] == "Folder Album"


@pytestmark_ffmpeg
def test_extract_cover_returns_none_when_no_apic_frame(tmp_path):
    from mutagen import File as MutagenFile

    clip = tmp_path / "plain.mp3"
    _make_clip(clip)

    mf = MutagenFile(str(clip))
    assert _extract_cover(mf) is None