rattail-project/rattail Files · tests/filemon/test

Files @ 2c7d6c16e14f

Branch filter:

Location: rattail-project/rattail/tests/filemon/test_util.py

2c7d6c16e14f 1.7 KiB text/x-python Show Annotation Show as Raw Download as Raw

lance

File Monitor overhaul!

* New configuration syntax (old syntax still supported but deprecated).
* Class-based actions.
* Configure keyword arguments to action callables.
* Configure retry for actions.
* Add (some) tests, docs.

# -*- coding: utf-8 -*-

import Queue
from unittest import TestCase

from fixture import TempIO

from edbob.configuration import AppConfigParser

from rattail.filemon import util
from rattail.filemon.config import Profile


class TestQueueExisting(TestCase):

    def setUp(self):
        self.tmp = TempIO()
        self.config = AppConfigParser(u'rattail')
        self.config.set(u'rattail.filemon', u'monitor', u'foo')
        self.config.set(u'rattail.filemon', u'foo.dirs', self.tmp)
        self.config.set(u'rattail.filemon', u'foo.actions', u'noop')
        self.config.set(u'rattail.filemon', u'foo.action.noop.func', u'rattail.filemon.actions:noop')
        self.profile = Profile(self.config, u'foo')
        self.profile.queue = Queue.Queue()

    def test_nothing_queued_if_no_files_exist(self):
        util.queue_existing(self.profile, self.tmp)
        self.assertTrue(self.profile.queue.empty())

    def test_normal_files_are_queued_but_not_folders(self):
        self.tmp.putfile(u'file', u'')
        self.tmp.mkdir(u'folder')
        util.queue_existing(self.profile, self.tmp)
        self.assertEqual(self.profile.queue.qsize(), 1)
        self.assertEqual(self.profile.queue.get_nowait(), self.tmp.join(u'file'))
        self.assertTrue(self.profile.queue.empty())

    def test_if_profile_watches_locks_then_normal_files_are_queued_but_not_lock_files(self):
        self.profile.watch_locks = True
        self.tmp.putfile(u'file1.lock', u'')
        self.tmp.putfile(u'file2', u'')
        util.queue_existing(self.profile, self.tmp)
        self.assertEqual(self.profile.queue.qsize(), 1)
        self.assertEqual(self.profile.queue.get_nowait(), self.tmp.join(u'file2'))
        self.assertTrue(self.profile.queue.empty())