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)