symuid/tests/test_sync.py

import os
import shutil
import unittest.mock

import pytest
from symuid import PlayCount, Track
from symuid._datetime import unix_epoch_time_to_datetime_utc
from symuid._uuid import generate_uuid4_bytes
from symuid.sync import _main, _walk_tracks, sync


def test_add_uuid(empty_ogg_opus_path):
    assert Track(empty_ogg_opus_path).get_uuid() is None
    sync(_walk_tracks([os.path.dirname(empty_ogg_opus_path)]))
    assert Track(empty_ogg_opus_path).get_uuid() is not None


def test_add_uuid_idempotent(empty_ogg_opus_path):
    sync(_walk_tracks([os.path.dirname(empty_ogg_opus_path)]))
    uuid = Track(empty_ogg_opus_path).get_uuid()
    sync(_walk_tracks([os.path.dirname(empty_ogg_opus_path)]))
    assert Track(empty_ogg_opus_path).get_uuid() == uuid


def test_add_uuid_main_file(empty_ogg_opus_path):
    assert Track(empty_ogg_opus_path).get_uuid() is None
    with unittest.mock.patch("sys.argv", ["", empty_ogg_opus_path]):
        _main()
    assert Track(empty_ogg_opus_path).get_uuid() is not None


def test_add_uuid_main_multiple_files(empty_ogg_opus_path, empty_ogg_vorbis_path):
    with unittest.mock.patch(
        "sys.argv", ["", empty_ogg_opus_path, empty_ogg_vorbis_path]
    ):
        _main()
    assert Track(empty_ogg_opus_path).get_uuid() is not None
    assert Track(empty_ogg_vorbis_path).get_uuid() is not None


def test_add_uuid_main_dir(empty_ogg_opus_path):
    assert Track(empty_ogg_opus_path).get_uuid() is None
    with unittest.mock.patch("sys.argv", ["", os.path.dirname(empty_ogg_opus_path)]):
        _main()
    assert Track(empty_ogg_opus_path).get_uuid() is not None


def test_sync_play_count(tmpdir, tracks_dir_path):
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "id3v2.4-empty.mp3"),
        dst=tmpdir.join("a1.mp3"),
    )
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "ogg-vorbis-empty.ogg"),
        dst=tmpdir.join("a2.ogg"),
    )
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "ogg-opus-empty.opus"),
        dst=tmpdir.join("a3.opus"),
    )
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "id3v2.4-empty.mp3"),
        dst=tmpdir.join("b.mp3"),
    )
    uuid_a = generate_uuid4_bytes()
    uuid_b = generate_uuid4_bytes()
    track_a1 = Track(tmpdir.join("a1.mp3"))
    track_a1.assign_uuid(uuid_a)
    track_a2 = Track(tmpdir.join("a2.ogg"))
    track_a2.assign_uuid(uuid_a)
    track_a3 = Track(tmpdir.join("a3.opus"))
    track_a3.assign_uuid(uuid_a)
    track_b = Track(tmpdir.join("b.mp3"))
    track_b.assign_uuid(uuid_b)
    track_a1.register_play_count(
        PlayCount(
            player="cmus",
            library_id="lib1",
            register_dt=unix_epoch_time_to_datetime_utc(0),
            count=21,
        )
    )
    track_a1.register_play_count(
        PlayCount(
            player="cmus",
            library_id="lib1",
            register_dt=unix_epoch_time_to_datetime_utc(1),
            count=22,
        )
    )
    track_a2.register_play_count(
        PlayCount(
            player="cmus",
            library_id="lib1",
            register_dt=unix_epoch_time_to_datetime_utc(0),
            count=21,
        )
    )
    track_a2.register_play_count(
        PlayCount(
            player="cmus",
            library_id="lib2",
            register_dt=unix_epoch_time_to_datetime_utc(2),
            count=7,
        )
    )
    track_b.register_play_count(
        PlayCount(
            player="cmus",
            library_id="lib2",
            register_dt=unix_epoch_time_to_datetime_utc(3),
            count=4,
        )
    )
    sync((track_a1, track_a2, track_a3, track_b))
    play_counts_a1 = set(Track(tmpdir.join("a1.mp3")).get_play_counts())
    play_counts_a2 = set(Track(tmpdir.join("a2.ogg")).get_play_counts())
    play_counts_a3 = set(Track(tmpdir.join("a3.opus")).get_play_counts())
    play_counts_b = set(Track(tmpdir.join("b.mp3")).get_play_counts())
    assert len(play_counts_a1) == 3
    assert len(play_counts_a1) == len(play_counts_a2)
    assert len(play_counts_a1) == len(play_counts_a3)
    assert play_counts_a1 == play_counts_a2
    assert {pc.register_dt.timestamp() for pc in play_counts_a1} == pytest.approx(
        {0, 1, 2}
    )
    assert len(play_counts_b) == 1
    assert {pc.register_dt.timestamp() for pc in play_counts_b} == pytest.approx({3})


def test_sync_play_count_callback(tmpdir, tracks_dir_path):
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "id3v2.4-empty.mp3"),
        dst=tmpdir.join("a1.mp3"),
    )
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "ogg-vorbis-empty.ogg"),
        dst=tmpdir.join("a2.ogg"),
    )
    uuid_a = generate_uuid4_bytes()
    track_a1 = Track(tmpdir.join("a1.mp3"))
    track_a1.assign_uuid(uuid_a)
    track_a2 = Track(tmpdir.join("a2.ogg"))
    track_a2.assign_uuid(uuid_a)
    track_a1.register_play_count(
        PlayCount(
            player="cmus",
            library_id="lib1",
            register_dt=unix_epoch_time_to_datetime_utc(0),
            count=21,
        )
    )
    play_count_added_cb = unittest.mock.MagicMock()
    sync(_walk_tracks([tmpdir]), play_count_added_cb=play_count_added_cb)
    play_count_added_cb.assert_called_once_with(
        track_a2, ("symuid:pcnt:cmus:lib1:0", ["21"])
    )


def test_sync_play_count_callback_main(capsys, tmpdir, tracks_dir_path):
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "id3v2.4-empty.mp3"),
        dst=tmpdir.join("a1.mp3"),
    )
    shutil.copyfile(
        src=os.path.join(tracks_dir_path, "ogg-vorbis-empty.ogg"),
        dst=tmpdir.join("a2.ogg"),
    )
    uuid_a = generate_uuid4_bytes()
    track_a1 = Track(tmpdir.join("a1.mp3"))
    track_a1.assign_uuid(uuid_a)
    track_a2 = Track(tmpdir.join("a2.ogg"))
    track_a2.assign_uuid(uuid_a)
    track_a1.register_play_count(
        PlayCount(
            player="cmus",
            library_id="lib1",
            register_dt=unix_epoch_time_to_datetime_utc(0),
            count=21,
        )
    )
    with unittest.mock.patch("sys.argv", ["", tmpdir.strpath]):
        _main()
    out, _ = capsys.readouterr()
    assert (
        out
        == "{!r}: added play count tag ('symuid:pcnt:cmus:lib1:0', ['21'])\n".format(
            track_a2.path
        )
    )