Skip to content

Commit

Permalink
add flac tag test
Browse files Browse the repository at this point in the history
  • Loading branch information
vm86 committed Nov 16, 2024
1 parent f341cb7 commit 7ba59e0
Show file tree
Hide file tree
Showing 4 changed files with 79 additions and 74 deletions.
1 change: 1 addition & 0 deletions .gitattributes
Original file line number Diff line number Diff line change
@@ -1,2 +1,3 @@
tests/contrib/test.m4a filter=lfs diff=lfs merge=lfs -text
tests/contrib/test.mp3 filter=lfs diff=lfs merge=lfs -text
tests/contrib/test.flac filter=lfs diff=lfs merge=lfs -text
19 changes: 15 additions & 4 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,22 @@
import pytest


def _load_audio(codec: str) -> BytesIO:
return BytesIO(
initial_bytes=Path(f"tests/contrib/test.{codec}").read_bytes()
)


@pytest.fixture
def m4a_audio() -> BytesIO:
return _load_audio("m4a")


@pytest.fixture
def m4a_wtihout_tag() -> BytesIO:
return BytesIO(initial_bytes=Path("tests/contrib/test.m4a").read_bytes())
def mp3_audio() -> BytesIO:
return _load_audio("mp3")


@pytest.fixture
def mp3_wtihout_tag() -> BytesIO:
return BytesIO(initial_bytes=Path("tests/contrib/test.mp3").read_bytes())
def flac_audio() -> BytesIO:
return _load_audio("flac")
3 changes: 3 additions & 0 deletions tests/contrib/test.flac
Git LFS file not shown
130 changes: 60 additions & 70 deletions tests/test_tag.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
# ruff: noqa: S101
from __future__ import annotations

import json
from io import BytesIO
from subprocess import check_output
Expand All @@ -14,6 +16,26 @@


class TestTag:
def ffprobe_tag(self, data: bytes) -> dict[str, str]:
with NamedTemporaryFile("wb") as tmp_fd:
tmp_fd.write(data)
tmp_fd.flush()
json_output = json.loads(
check_output(
[
"ffprobe",
"-loglevel",
"panic",
"-print_format",
"json",
"-show_format",
"-show_streams",
tmp_fd.name,
]
)
)
return {k.lower(): v for k, v in json_output["format"]["tags"].items()}

@pytest.fixture
def tag(self) -> TrackTag:
return TrackTag(
Expand All @@ -25,95 +47,63 @@ def tag(self) -> TrackTag:
duration_ms=100.0,
)

def test_tag_mp4(self, tag: TrackTag, m4a_wtihout_tag: BytesIO) -> None:
after_tag = tag.to_bytes(m4a_wtihout_tag)
def tagging(self, track_tag: TrackTag, audio: BytesIO, codec: str) -> None:
after_tag = track_tag.to_bytes(audio, codec)
assert after_tag is not None

with NamedTemporaryFile("wb") as tmp_fd:
tmp_fd.write(after_tag)
tmp_fd.flush()
ffprobe_output = check_output(
[
"ffprobe",
"-loglevel",
"panic",
"-print_format",
"json",
"-show_format",
"-show_streams",
tmp_fd.name,
]
)
json_output = json.loads(ffprobe_output)
audio_tags = self.ffprobe_tag(after_tag)

tag_json = tag.to_dict()
tag_json = track_tag.to_dict()

for key in ("artist", "title", "album", "genre"):
assert json_output["format"]["tags"][key] == tag_json[key]
assert audio_tags.get(key, "") == tag_json[key], audio_tags

assert json_output["format"]["tags"]["date"] == tag_json["year"]
assert audio_tags["date"] == tag_json["year"]

def test_tag_partial_data_mp4(
self, tag: TrackTag, m4a_wtihout_tag: BytesIO
def partial_tagging(
self,
track_tag: TrackTag,
audio: BytesIO,
codec: str,
read_size: int,
min_tag_size: int,
) -> None:
m4a_parital = BytesIO()
while data := m4a_wtihout_tag.read(49):
m4a_parital.write(data)
last_seek = m4a_parital.tell()
audio_parital = BytesIO()
while data := audio.read(read_size):
audio_parital.write(data)
last_seek = audio_parital.tell()

after_tag = tag.to_bytes(m4a_parital)
if len(m4a_parital.getbuffer()) < MP4_HEADER_MIN_SIZE:
after_tag = track_tag.to_bytes(audio_parital, codec)
if len(audio_parital.getbuffer()) < min_tag_size:
assert after_tag is None
else:
assert after_tag is not None
assert m4a_parital.tell() == last_seek

m4a_parital.seek(0)

assert m4a_wtihout_tag.getvalue() == m4a_parital.getvalue()
assert audio_parital.tell() == last_seek

def test_tag_mp3(self, tag: TrackTag, mp3_wtihout_tag: BytesIO) -> None:
after_tag = tag.to_bytes(mp3_wtihout_tag)
assert after_tag is not None
audio_parital.seek(0)

with NamedTemporaryFile("wb") as tmp_fd:
tmp_fd.write(after_tag)
tmp_fd.flush()
ffprobe_output = check_output(
[
"ffprobe",
"-loglevel",
"panic",
"-print_format",
"json",
"-show_format",
"-show_streams",
tmp_fd.name,
]
)
json_output = json.loads(ffprobe_output)
assert audio.getvalue() == audio_parital.getvalue()

tag_json = tag.to_dict()
def test_tag_mp4(self, tag: TrackTag, m4a_audio: BytesIO) -> None:
self.tagging(tag, m4a_audio, "aac")

for key in ("artist", "title", "album", "genre"):
assert json_output["format"]["tags"][key] == tag_json[key]
def test_tag_partial_data_mp4(
self, tag: TrackTag, m4a_audio: BytesIO
) -> None:
self.partial_tagging(tag, m4a_audio, "aac", 49, MP4_HEADER_MIN_SIZE)

assert json_output["format"]["tags"]["date"] == tag_json["year"]
def test_tag_mp3(self, tag: TrackTag, mp3_audio: BytesIO) -> None:
self.tagging(tag, mp3_audio, "mp3")

def test_tag_partial_data_mp3(
self, tag: TrackTag, mp3_wtihout_tag: BytesIO
self, tag: TrackTag, mp3_audio: BytesIO
) -> None:
mp3_parital = BytesIO()
while data := mp3_wtihout_tag.read(1024):
mp3_parital.write(data)
last_seek = mp3_parital.tell()
self.partial_tagging(tag, mp3_audio, "mp3", 1024, MP3_HEADER_MIN_SIZE)

after_tag = tag.to_bytes(mp3_parital)
if len(mp3_parital.getbuffer()) < MP3_HEADER_MIN_SIZE:
assert after_tag is None
else:
assert after_tag is not None
assert mp3_parital.tell() == last_seek
def test_tag_flac(self, tag: TrackTag, flac_audio: BytesIO) -> None:
self.tagging(tag, flac_audio, "flac")

mp3_parital.seek(0)
assert mp3_wtihout_tag.getvalue() == mp3_parital.getvalue()
def test_tag_partial_data_flac(
self, tag: TrackTag, flac_audio: BytesIO
) -> None:
self.partial_tagging(tag, flac_audio, "flac", 1024, 9098)

0 comments on commit 7ba59e0

Please sign in to comment.