emmental/curds/test_trak.py

112 lines
3.3 KiB
Python

# Copyright 2019 (c) Anna Schumaker.
from . import data
from . import notify
from . import trak
from . 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):
notify.clear()
trak.reset()
self.new_track = None
def on_new_track(self, track):
self.new_track = track
def test_trak_init(self):
self.assertEqual(trak.trak_file, "tracks.pickle")
self.assertFalse(trak.trak_lock.locked())
self.assertEqual(trak.trak_db, { })
def test_trak_clear(self):
trak.trak_db = { 1 : "a", 2 : "b", 3 : "c" }
trak.clear()
self.assertEqual(trak.trak_db, { })
def test_trak_exists_remove(self):
self.assertFalse(trak.exists())
self.assertFalse(trak.remove())
with data.DataFile("tracks.pickle", data.WRITE) as f:
f.pickle([ 1, 2, 3, 4, 5 ])
self.assertTrue(trak.exists())
self.assertTrue(trak.remove())
self.assertFalse(trak.exists())
def test_trak_lookup(self):
notify.register("new-track", self.on_new_track)
path = os.path.join(test_tracks, "01 - Test Track.ogg" )
track = trak.lookup(path)
self.assertEqual(track.path, path)
self.assertEqual(self.new_track, track)
self.new_track = None
self.assertEqual(trak.lookup(path), track)
self.assertEqual(self.new_track, None)
self.assertIsNone(trak.lookup(os.path.join(test_tracks, "test.txt")))
def test_trak_notify_save(self):
self.assertFalse(trak.exists())
notify.notify("save-data")
self.assertTrue(trak.exists())
def test_trak_save_load(self):
trak.load()
self.assertEqual(trak.trak_db, { })
trak.trak_db = { 1 : "a", 2 : "b", 3 : "c" }
self.assertFalse(trak.exists())
trak.save()
self.assertTrue(trak.exists())
trak.clear()
trak.load()
self.assertEqual(trak.trak_db, { 1 : "a", 2 : "b", 3 : "c" })
def test_trak_reset(self):
trak.trak_db = { 1 : "a", 2 : "b", 3 : "c" }
with data.DataFile("tracks.pickle", data.WRITE) as f:
f.pickle(trak.trak_db)
self.assertTrue(trak.exists())
trak.reset()
self.assertEqual(trak.trak_db, { })
self.assertFalse(trak.exists())
def test_trak_stress(self):
dfile = data.DataFile("tracks.pickle", data.READ)
tq = threadqueue.ThreadQueue()
tracks = 0
self.assertFalse(dfile.exists())
self.assertEqual(len(trak.trak_db), 0)
for dirname, subdirs, files in os.walk(test_library):
save = False
for f in files:
tq.push(trak.lookup, os.path.join(dirname, f))
tracks += 1
save = True
if save == True:
tq.push(trak.save)
self.assertLess(len(trak.trak_db), tracks)
tq.join()
tq.stop()
self.assertEqual(len(trak.trak_db), tracks)
self.assertTrue(dfile.exists())
tag_list = list(trak.trak_db.values())
trak.clear()
trak.load()
self.assertEqual(len(trak.trak_db), tracks)
for tag in tag_list:
self.assertIn(tag.path, trak.trak_db)