Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def wait():
"""
Wait long enough for file/folder mtime to change. This is needed
to be able to detected modifications.
"""
if platform.is_darwin() or platform.is_windows():
# on macOS resolution of stat.mtime is only 1 second
time.sleep(1.5)
else:
time.sleep(0.5)
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pytest
from watchdog.utils import platform
if not platform.is_windows(): # noqa
pytest.skip("Windows only.", allow_module_level=True)
import os
import os.path
from time import sleep
from watchdog.events import (
DirCreatedEvent,
DirMovedEvent,
)
from watchdog.observers.api import ObservedWatch
from watchdog.observers.read_directory_changes import WindowsApiEmitter
from . import Empty, Queue
from .shell import (
mkdir,
def start_watching(path=None, use_full_emitter=False, recursive=True):
path = p('') if path is None else path
global emitter
if platform.is_linux() and use_full_emitter:
emitter = InotifyFullEmitter(event_queue, ObservedWatch(path, recursive=recursive))
else:
emitter = Emitter(event_queue, ObservedWatch(path, recursive=recursive))
if platform.is_darwin():
# FSEvents will report old events (like create for mkdtemp in test
# setup. Waiting for a considerable time seems to 'flush' the events.
time.sleep(10)
emitter.start()
'watchdog.observers.read_directory_changes_async',
]
cover_packages_bsd = [
'watchdog.observers.kqueue',
]
cover_packages_darwin = [
'watchdog.observers.fsevents',
'watchdog.observers.kqueue',
]
cover_packages_linux = [
'watchdog.observers.inotify',
]
if platform.is_windows():
cover_packages.extend(cover_packages_windows)
elif platform.is_darwin():
cover_packages.extend(cover_packages_darwin)
elif platform.is_bsd():
cover_packages.extend(cover_packages_bsd)
elif platform.is_linux():
cover_packages.extend(cover_packages_linux)
if __name__ == "__main__":
config_path = os.path.join(parent_dir_path, 'nose.cfg')
argv = [__file__]
argv.append('--detailed-errors')
argv.append('--with-coverage')
# Coverage by itself generates more usable reports.
#argv.append('--cover-erase')
#argv.append('--cover-html')
FileDeletedEvent,
FileCreatedEvent,
FileModifiedEvent,
EVENT_TYPE_MOVED,
EVENT_TYPE_DELETED,
EVENT_TYPE_CREATED
)
# Maximum number of events to process.
MAX_EVENTS = 4096
# O_EVTONLY value from the header files for OS X only.
O_EVTONLY = 0x8000
# Pre-calculated values for the kevent filter, flags, and fflags attributes.
if platform.is_darwin():
WATCHDOG_OS_OPEN_FLAGS = O_EVTONLY
else:
WATCHDOG_OS_OPEN_FLAGS = os.O_RDONLY | os.O_NONBLOCK
WATCHDOG_KQ_FILTER = select.KQ_FILTER_VNODE
WATCHDOG_KQ_EV_FLAGS = select.KQ_EV_ADD | select.KQ_EV_ENABLE | select.KQ_EV_CLEAR
WATCHDOG_KQ_FFLAGS = (
select.KQ_NOTE_DELETE
| select.KQ_NOTE_WRITE
| select.KQ_NOTE_EXTEND
| select.KQ_NOTE_ATTRIB
| select.KQ_NOTE_LINK
| select.KQ_NOTE_RENAME
| select.KQ_NOTE_REVOKE
)
# Flag tests.
from django.core.exceptions import ObjectDoesNotExist
from django.core.management.base import BaseCommand
from django.db import transaction
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler
from watchdog.observers.polling import PollingObserverVFS
from watchdog.utils import platform
from jane.exceptions import JaneWaveformTaskException
from ... import models
from ...tasks import process_file
# monkey - https://github.com/gorakhargosh/watchdog/issues/123
if platform.is_windows():
import ctypes
from watchdog.observers import winapi
_HANDLE = ctypes.c_void_p
_INVALID_HANDLE_VALUE = _HANDLE(-1).value
winapi.INVALID_HANDLE_VALUE = _INVALID_HANDLE_VALUE
class EventHandler(LoggingEventHandler):
def on_any_event(self, event):
"""
Catch-all event handler.
"""
# Do not deal with modified directories! Those are fired every time
elif platform.is_darwin():
try:
from .fsevents import FSEventsObserver as Observer
except Exception:
try:
from .kqueue import KqueueObserver as Observer
warnings.warn("Failed to import fsevents. Fall back to kqueue")
except Exception:
from .polling import PollingObserver as Observer
warnings.warn("Failed to import fsevents and kqueue. Fall back to polling.")
elif platform.is_bsd():
from .kqueue import KqueueObserver as Observer
elif platform.is_windows():
# TODO: find a reliable way of checking Windows version and import
# polling explicitly for Windows XP
try:
from .read_directory_changes import WindowsApiObserver as Observer
except Exception:
from .polling import PollingObserver as Observer
warnings.warn("Failed to import read_directory_changes. Fall back to polling.")
else:
from .polling import PollingObserver as Observer
__all__ = ["Observer"]
FileDeletedEvent,
FileCreatedEvent,
FileModifiedEvent,
EVENT_TYPE_MOVED,
EVENT_TYPE_DELETED,
EVENT_TYPE_CREATED
)
# Maximum number of events to process.
MAX_EVENTS = 4096
# O_EVTONLY value from the header files for OS X only.
O_EVTONLY = 0x8000
# Pre-calculated values for the kevent filter, flags, and fflags attributes.
if platform.is_darwin():
WATCHDOG_OS_OPEN_FLAGS = O_EVTONLY
else:
WATCHDOG_OS_OPEN_FLAGS = os.O_RDONLY | os.O_NONBLOCK
WATCHDOG_KQ_FILTER = select.KQ_FILTER_VNODE
WATCHDOG_KQ_EV_FLAGS = select.KQ_EV_ADD | select.KQ_EV_ENABLE | select.KQ_EV_CLEAR
WATCHDOG_KQ_FFLAGS = (
select.KQ_NOTE_DELETE |
select.KQ_NOTE_WRITE |
select.KQ_NOTE_EXTEND |
select.KQ_NOTE_ATTRIB |
select.KQ_NOTE_LINK |
select.KQ_NOTE_RENAME |
select.KQ_NOTE_REVOKE
)
# Flag tests.
subdirectories under a directory, additional watches must be created.
This emitter implementation therefore automatically adds watches for
sub-directories if running in recursive mode.
Some extremely useful articles and documentation:
.. _inotify FAQ: http://inotify.aiken.cz/?section=inotify&page=faq&lang=en
.. _intro to inotify: http://www.linuxjournal.com/article/8478
"""
from __future__ import with_statement
from watchdog.utils import platform
if platform.is_linux():
import os
import errno
import struct
import threading
import ctypes
from ctypes import\
c_int,\
c_char_p,\
c_uint32
from pathtools.path import absolute_path
from watchdog.utils import\
has_attribute,\
ctypes_find_library
from watchdog.observers.api import\
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
from __future__ import with_statement
from watchdog.utils import platform
if platform.is_bsd() or platform.is_darwin():
import os
import stat
import sys
import errno
import os.path
import threading
from watchdog.utils import has_attribute
import select
if not has_attribute(select, 'kqueue') or sys.version_info < (2, 7, 0):
import select_backport as select
from watchdog.utils import echo
from watchdog.utils import absolute_path, real_absolute_path
from watchdog.utils.dirsnapshot import DirectorySnapshot