emmental/trackdb/test_allocator.py

# Copyright 2021 (c) Anna Schumaker.
from lib import publisher
from . import allocator
from . import library
import pathlib
import threading
import unittest

test_tracks = pathlib.Path("./trier/Test Album")

class TestTrackAllocator(unittest.TestCase):
    def setUp(self):
        self.added = None
        self.removed = None

    def on_track_added(self, track):
        self.added = track

    def on_track_removed(self, track):
        self.removed = track

    def test_allocator_init(self):
        alloc = allocator.TrackAllocator()
        self.assertEqual(alloc.tracks, { })
        self.assertEqual(alloc.nextid, 0)
        self.assertIsInstance(alloc.lock, type(threading.Lock()))
        self.assertIsInstance(alloc.Added, publisher.Publisher)
        self.assertIsInstance(alloc.Removed, publisher.Publisher)

    def test_allocator(self):
        lib = library.LibraryPath(test_tracks)
        alloc = allocator.TrackAllocator()
        alloc.Added.register(self.on_track_added)
        alloc.Removed.register(self.on_track_removed)

        track = alloc.allocate(lib, test_tracks / "01 - Test Track.ogg")
        self.assertEqual(alloc.tracks[0], track)
        self.assertEqual(alloc[0], track)
        self.assertEqual(len(alloc), 1)
        self.assertEqual(alloc.nextid, 1)
        self.assertEqual(track.trackid, 0)
        self.assertEqual(self.added, track)

        track2 = alloc.allocate(lib, test_tracks / "02 - Test {Disc 2}.ogg")
        self.assertEqual(alloc.tracks[1], track2)
        self.assertEqual(alloc[1], track2)
        self.assertEqual(len(alloc), 2)

        self.assertIsNone(alloc.allocate(lib, test_tracks / "No Such File"))
        self.assertEqual(self.added, track2)
        self.assertEqual([ t for t in alloc.list_tracks(lib) ], [ track, track2 ])
        track.library = None
        self.assertEqual([ t for t in alloc.list_tracks(lib) ], [ track2 ])

        alloc.remove(track)
        self.assertEqual(alloc[0], None)
        self.assertEqual(alloc.tracks, { 1 : track2 })
        self.assertEqual(alloc.nextid, 2)
        self.assertEqual(self.removed, track)

        alloc.reset()
        self.assertEqual(alloc.nextid, 0)
        self.assertEqual(alloc.tracks, { })
        self.assertEqual(alloc.Added.subscribers, set())
        self.assertEqual(alloc.Removed.subscribers, set())

    def test_allocator_autoremove(self):
        lib = library.LibraryPath(test_tracks)
        alloc = allocator.TrackAllocator()
        alloc.Removed.register(self.on_track_removed)

        track = alloc.allocate(lib, test_tracks / "01 - Test Track.ogg")
        track2 = alloc.allocate(lib, test_tracks / "02 - Test {Disc 2}.ogg")
        track2.path = pathlib.Path("No Such File")

        self.assertEqual(alloc.autoremove(lib), [ track2 ])
        self.assertEqual([ t for t in alloc.list_tracks(lib) ], [ track ])

    def test_allocator_state(self):
        lib = library.LibraryPath(test_tracks)
        alloc = allocator.TrackAllocator()
        track = alloc.allocate(lib, test_tracks / "01 - Test Track.ogg")

        state = alloc.__getstate__()
        self.assertEqual(state, { "tracks" : { 0 : track }, 
                                  "nextid" : 1 })

        alloc.__dict__.clear()
        alloc.__setstate__(state)
        self.assertEqual(alloc.tracks, { 0 : track })
        self.assertEqual(alloc.nextid, 1)
        self.assertIsInstance(alloc.lock, type(threading.Lock()))
        self.assertIsInstance(alloc.Added, publisher.Publisher)
        self.assertIsInstance(alloc.Removed, publisher.Publisher)