test_track.py 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176
  1. import datetime as dt
  2. import os
  3. import shutil
  4. import mutagen
  5. import pytest
  6. import symuid
  7. # TODO test aac / m4a itunes tags
  8. TRACKS_DIR_PATH = os.path.join(os.path.dirname(__file__), 'tracks')
  9. def utc_dt(hour=0):
  10. return dt.datetime(2018, 9, 26, hour, tzinfo=dt.timezone.utc)
  11. @pytest.fixture
  12. def empty_id3_track(tmpdir):
  13. path = tmpdir.join('empty.mp3').strpath
  14. shutil.copyfile(
  15. src=os.path.join(TRACKS_DIR_PATH, 'id3v2.4-empty.mp3'),
  16. dst=path,
  17. )
  18. return symuid.Track(path)
  19. @pytest.mark.parametrize('path', [
  20. os.path.join(TRACKS_DIR_PATH, 'id3v2.4-typical.mp3'),
  21. os.path.join(TRACKS_DIR_PATH, 'id3v2.4-empty.mp3'),
  22. ])
  23. def test_init(path):
  24. symuid.Track(path)
  25. @pytest.mark.parametrize(('path', 'expected_comment'), [
  26. (os.path.join(TRACKS_DIR_PATH, 'id3v2.4-typical.mp3'), 'some comment'),
  27. (os.path.join(TRACKS_DIR_PATH, 'id3v2.4-empty.mp3'), None),
  28. ])
  29. def test_get_comment(path, expected_comment):
  30. track = symuid.Track(path)
  31. assert expected_comment == track.comment
  32. def test_set_comment(empty_id3_track):
  33. assert empty_id3_track.comment is None
  34. empty_id3_track.comment = 'note'
  35. assert empty_id3_track.comment == 'note'
  36. empty_id3_track.comment += 's'
  37. assert empty_id3_track.comment == 'notes'
  38. assert symuid.Track(empty_id3_track.path).comment == 'notes'
  39. @pytest.mark.parametrize(('play_count'), [
  40. symuid.PlayCount('pytest', 'lib', utc_dt(), 7),
  41. ])
  42. def test_register_play_count(empty_id3_track, play_count):
  43. empty_id3_track.register_play_count(play_count)
  44. tags = mutagen.File(empty_id3_track.path).tags
  45. assert len(tags) == 1
  46. expected_desc = 'symuid:pcnt:{}:{}:{}'.format(
  47. play_count.player, play_count.library_id, int(
  48. play_count.register_dt.timestamp()),
  49. )
  50. tag = tags['TXXX:' + expected_desc]
  51. assert tag.desc == expected_desc
  52. assert tag.text == [str(play_count.count)]
  53. @pytest.mark.parametrize(('expected_counts'), [
  54. [],
  55. [symuid.PlayCount('player', 'lib', utc_dt(), 3)],
  56. [symuid.PlayCount('a', '0', utc_dt(0), 1),
  57. symuid.PlayCount('b', '1', utc_dt(1), 2)],
  58. [symuid.PlayCount('a', '0', utc_dt(0), 1),
  59. symuid.PlayCount('a', '2', utc_dt(1), 2),
  60. symuid.PlayCount('b', '3', utc_dt(3), 3)],
  61. ])
  62. def test__get_play_counts_all(empty_id3_track, expected_counts):
  63. for play_count in expected_counts:
  64. empty_id3_track.register_play_count(play_count)
  65. assert expected_counts == list(empty_id3_track._get_play_counts())
  66. def test__get_play_counts_filtered(empty_id3_track):
  67. counts = [symuid.PlayCount('a', '0', utc_dt(0), 1),
  68. symuid.PlayCount('a', '0', utc_dt(1), 2),
  69. symuid.PlayCount('a', '1', utc_dt(0), 3),
  70. symuid.PlayCount('b', '2', utc_dt(1), 4)]
  71. for play_count in counts:
  72. empty_id3_track.register_play_count(play_count)
  73. assert set(empty_id3_track._get_play_counts(player='a')) \
  74. == set(filter(lambda pc: pc.player == 'a', counts))
  75. assert set(empty_id3_track._get_play_counts(player='b')) \
  76. == set(filter(lambda pc: pc.player == 'b', counts))
  77. assert set(empty_id3_track._get_play_counts(player='a', library_id='0')) \
  78. == set(filter(lambda pc: pc.library_id == '0', counts))
  79. assert set(empty_id3_track._get_play_counts(player='a', library_id='2')) \
  80. == set()
  81. def test__get_latest_play_counts(empty_id3_track):
  82. counts = [symuid.PlayCount('a', '0', utc_dt(0), 1),
  83. symuid.PlayCount('a', '0', utc_dt(1), 2),
  84. symuid.PlayCount('a', '1', utc_dt(0), 3),
  85. symuid.PlayCount('a', '1', utc_dt(2), 4),
  86. symuid.PlayCount('b', '2', utc_dt(3), 5)]
  87. for play_count in counts:
  88. empty_id3_track.register_play_count(play_count)
  89. assert set(empty_id3_track._get_latest_play_counts()) \
  90. == set([counts[1], counts[3], counts[4]])
  91. assert set(empty_id3_track._get_latest_play_counts(player='a')) \
  92. == set([counts[1], counts[3]])
  93. assert set(empty_id3_track._get_latest_play_counts(player='a', library_id='0')) \
  94. == set([counts[1]])
  95. def test_get_play_count_sum(empty_id3_track):
  96. counts = [symuid.PlayCount('a', '0', utc_dt(0), 1),
  97. symuid.PlayCount('a', '0', utc_dt(1), 2),
  98. symuid.PlayCount('a', '1', utc_dt(0), 3),
  99. symuid.PlayCount('a', '1', utc_dt(2), 4),
  100. symuid.PlayCount('b', '2', utc_dt(3), 5)]
  101. for play_count in counts:
  102. empty_id3_track.register_play_count(play_count)
  103. assert 2 + 4 + 5 == empty_id3_track.get_play_count_sum()
  104. def test_increase_play_count(empty_id3_track):
  105. init_count = symuid.PlayCount('a', '0', utc_dt(0), 3)
  106. empty_id3_track.register_play_count(init_count)
  107. assert empty_id3_track.get_play_count_sum() == 3
  108. empty_id3_track.increase_play_count('a', '0')
  109. dt_now = symuid._utc_dt_now()
  110. assert empty_id3_track.get_play_count_sum() == 4
  111. counts = set(empty_id3_track._get_play_counts())
  112. assert len(counts) == 2
  113. counts.remove(init_count)
  114. new_count = counts.pop()
  115. assert new_count.player == 'a'
  116. assert new_count.library_id == '0'
  117. assert abs(new_count.register_dt - dt_now).total_seconds() < 5
  118. assert new_count.count == 4
  119. def test_increase_play_count_init(empty_id3_track):
  120. empty_id3_track.increase_play_count('a', '0')
  121. assert empty_id3_track.get_play_count_sum() == 1
  122. count, = empty_id3_track._get_play_counts()
  123. assert count.player == 'a'
  124. assert count.library_id == '0'
  125. assert abs(count.register_dt - symuid._utc_dt_now()).total_seconds() < 5
  126. assert count.count == 1
  127. def test_increase_play_count_others(empty_id3_track):
  128. empty_id3_track.register_play_count(
  129. symuid.PlayCount('a', '0', utc_dt(0), 1),
  130. )
  131. empty_id3_track.register_play_count(
  132. symuid.PlayCount('a', '1', utc_dt(0), 2),
  133. )
  134. empty_id3_track.register_play_count(
  135. symuid.PlayCount('b', '0', utc_dt(0), 3),
  136. )
  137. assert empty_id3_track.get_play_count_sum() == 6
  138. empty_id3_track.increase_play_count('a', '1')
  139. assert empty_id3_track.get_play_count_sum() == 7
  140. assert len(list(empty_id3_track._get_play_counts(player='b'))) == 1
  141. assert len(list(
  142. empty_id3_track._get_play_counts(player='a', library_id='0')
  143. )) == 1
  144. assert len(list(
  145. empty_id3_track._get_play_counts(player='a', library_id='1')
  146. )) == 2