123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170 |
- 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)
|