Skip to content

Commit

Permalink
Browse files Browse the repository at this point in the history
Auto add in notify#239 (#240)
* Added auto_add to notify.add_path()

* Added auto_add with default to True to notify.add_path().
  Fixes #239.

* Added various notify tests (#239)
  • Loading branch information
kromg authored and prologic committed Jun 13, 2018
1 parent f0cb656 commit 5dae195
Show file tree
Hide file tree
Showing 2 changed files with 129 additions and 15 deletions.
4 changes: 2 additions & 2 deletions circuits/io/notify.py
Expand Up @@ -61,9 +61,9 @@ def _on_process_events(self, event):
if mask & k:
self.fire(v(name, path, pathname, dir))

def add_path(self, path, mask=None, recursive=False):
def add_path(self, path, mask=None, recursive=False, auto_add=True):
mask = mask or MASK
self._wm.add_watch(path, mask, rec=recursive)
self._wm.add_watch(path, mask, rec=recursive, auto_add=auto_add)

def remove_path(self, path, recursive=False):
wd = self._wm.get_wd(path)
Expand Down
140 changes: 127 additions & 13 deletions tests/io/test_notify.py
@@ -1,5 +1,6 @@
#!/usr/bin/env python
import pytest
import os

from circuits import Component, handler

Expand All @@ -12,31 +13,144 @@
class App(Component):

def init(self, *args, **kwargs):
self.created = False
self.created_status = False

@handler('created', channel='notify')
def created(self, *args, **kwargs):
self.created = True
self.created_status = True


def test_notify(manager, watcher, tmpdir):
class Creator:

def __init__(self, app, watcher, tmpdir, timeout=0.5):
self.app = app
self.watcher = watcher
self.tmpdir = tmpdir
self.timeout = timeout

def create(self, *targets, assert_created=True, dir=False):
target = os.path.join(*targets)
self.tmpdir.ensure(target, dir=dir)
self.watcher.wait("created", timeout=self.timeout)
assert self.app.created_status == assert_created
# Reset for next call
self.watcher.clear()
self.app.created_status = False


@pytest.fixture()
def app(manager, watcher):
app = App().register(manager)
notify = Notify().register(app)
yield app
# Unregister application on test end
app.unregister()
watcher.wait("unregistered")


@pytest.fixture()
def notify(app, watcher):
notify = Notify().register(app)
watcher.wait("registered")
return notify


@pytest.fixture()
def creator(app, watcher, tmpdir):
return Creator(app, watcher, tmpdir)


# TESTS


def test_notify_file(notify, tmpdir, creator):

# Add a path to the watch
notify.add_path(str(tmpdir))

tmpdir.ensure("helloworld.txt")
watcher.wait("created")
assert app.created
app.created = False
# Test creation and detection of a file
creator.create("helloworld.txt")

# Remove the path from the watch
notify.remove_path(str(tmpdir))

tmpdir.ensure("helloworld2.txt")
watcher.wait("created")
assert not app.created
# Test creation and NON detection of a file
creator.create("helloworld2.txt", assert_created=False)


def test_notify_dir(notify, tmpdir, creator):

# Add a path to the watch
notify.add_path(str(tmpdir))

# Test creation and detection of a file
creator.create("hellodir", dir=True)

# Remove the path from the watch
notify.remove_path(str(tmpdir))

# Test creation and NON detection of a file
creator.create("hellodir2", dir=True, assert_created=False)


def test_notify_subdir_recursive(notify, tmpdir, creator):

# Add a subdir
subdir = "sub"
tmpdir.ensure(subdir, dir=True)

# Add a path to the watch
notify.add_path(str(tmpdir), recursive=True)

# Test creation and detection of a file in subdir
creator.create(subdir, "helloworld.txt", assert_created=True)


@pytest.mark.xfail(reason="pyinotify issue #133")
def test_notify_subdir_recursive_remove_path(notify, tmpdir, creator):
# This is logically the second part of the above test,
# but pyinotify fails on rm_watch(...., rec=True)

# Add a subdir
subdir = "sub"
tmpdir.ensure(subdir, dir=True)

# Add a path to the watch
notify.add_path(str(tmpdir), recursive=True)

# Remove the path from the watch
notify.remove_path(str(tmpdir), recursive=True)

# Test creation and NON detection of a file in subdir
creator.create(subdir, "helloworld2.txt", assert_created=False)


def test_notify_subdir_recursive_auto_add(notify, tmpdir, creator):

# Add a path to the watch
notify.add_path(str(tmpdir), recursive=True)

# Create/detect subdirectory
subdir = "sub"
creator.create(subdir, dir=True, assert_created=True)

# Create/detect file in subdirectory
creator.create(subdir, "helloworld.txt", assert_created=True)

# Skip notify.remove_path() because pyinotify is broken


def test_notify_subdir_recursive_no_auto_add(notify, tmpdir, creator):

# Add a path to the watch
notify.add_path(str(tmpdir), recursive=True, auto_add=False)

# Create/detect subdirectory
subdir = "sub"
creator.create(subdir, dir=True, assert_created=True)

# Create, not detect file in subdirectory
creator.create(subdir, "helloworld.txt", assert_created=False)

# Skip notify.remove_path() because pyinotify is broken


app.unregister()
watcher.wait("unregistered")

0 comments on commit 5dae195

Please sign in to comment.