curds: Add save() and load() functions
These functions pickle out the entire tag map onto disk. I also add a stress test for the entire tags module that scans tracks using the threadqueue, since that's how I intend everything to be used. Signed-off-by: Anna Schumaker <Anna@NoWheyCreamery.com>
This commit is contained in:
parent
c28e06730b
commit
5908a8af50
|
@ -1,4 +1,5 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
import data
|
||||
import mutagen
|
||||
import os
|
||||
import re
|
||||
|
@ -61,3 +62,12 @@ class Track(Tag):
|
|||
def add(path):
|
||||
track = Track(path)
|
||||
return track if Tag.lookup(track) == track else None
|
||||
|
||||
|
||||
def save():
|
||||
with data.DataFile("tags.pickle", data.WRITE) as f:
|
||||
f.pickle(tag_map)
|
||||
|
||||
def load():
|
||||
with data.DataFile("tags.pickle", data.READ) as f:
|
||||
tag_map.update(f.unpickle())
|
||||
|
|
|
@ -0,0 +1,65 @@
|
|||
# Copyright 2019 (c) Anna Schumaker.
|
||||
import data
|
||||
import tags
|
||||
import threadqueue
|
||||
import unittest
|
||||
import os
|
||||
|
||||
test_library = os.path.abspath("./trier/Test Library")
|
||||
test_tracks = os.path.abspath("./trier/Test Album")
|
||||
|
||||
class TestTags(unittest.TestCase):
|
||||
def setUp(self):
|
||||
data.DataFile("tags.pickle", data.WRITE).remove()
|
||||
tags.tag_map.clear()
|
||||
|
||||
def test_tags_save_load(self):
|
||||
dfile = data.DataFile("tags.pickle", data.READ)
|
||||
|
||||
self.assertFalse(dfile.exists())
|
||||
tags.load()
|
||||
self.assertEqual(len(tags.tag_map), 0)
|
||||
|
||||
t = tags.Track.add(os.path.join(test_tracks, "01 - Test Track.ogg"))
|
||||
self.assertFalse(dfile.exists())
|
||||
tags.save()
|
||||
self.assertTrue(dfile.exists())
|
||||
|
||||
tags.tag_map.clear()
|
||||
self.assertFalse(hash(t) in tags.tag_map)
|
||||
self.assertFalse(hash(t.album) in tags.tag_map)
|
||||
|
||||
tags.load()
|
||||
self.assertTrue(hash(t) in tags.tag_map)
|
||||
self.assertTrue(hash(t.album) in tags.tag_map)
|
||||
|
||||
def test_tags_stress(self):
|
||||
dfile = data.DataFile("tags.pickle", data.READ)
|
||||
tq = threadqueue.ThreadQueue()
|
||||
tracks = 0
|
||||
albums = 0
|
||||
tag_list = list(tags.tag_map)
|
||||
|
||||
self.assertFalse(dfile.exists())
|
||||
|
||||
for dirname, subdirs, files in os.walk(test_library):
|
||||
save = False
|
||||
for f in files:
|
||||
tq.push(tags.Track.add, os.path.join(dirname, f))
|
||||
tracks += 1
|
||||
save = True
|
||||
if save == True:
|
||||
tq.push(tags.save)
|
||||
albums += 1
|
||||
|
||||
self.assertLess(len(tags.tag_map), tracks + albums)
|
||||
tq.join()
|
||||
tq.stop()
|
||||
self.assertEqual(len(tags.tag_map), tracks + albums)
|
||||
self.assertTrue(dfile.exists())
|
||||
|
||||
tags.tag_map.clear()
|
||||
tags.load()
|
||||
self.assertEqual(len(tags.tag_map), tracks + albums)
|
||||
for tag in tag_list:
|
||||
self.assertIn(hash(tag), tags.tag_map)
|
Loading…
Reference in New Issue