<commit_before>#!/usr/bin/env py.test

# This testing script can be run either from python 3 or python 2. Run with
# `py.test test.py` or `py.test-2.7 test.py`.
#
# This script will try to import the inotify module from the build directory in
# ../build/lib.linux-{platform}-{pyversion}/inotify relative to its own
# location. If that directory cannot be found it will import the inotify module
# from the default path.


from __future__ import print_function

import sys, os, shutil, tempfile, inspect
import pytest

if not sys.platform.startswith('linux'): raise Exception("This module will only work on Linux")

# find the build dir
un = os.uname()
ver = '.'.join(str(x) for x in sys.version_info[:2])
testdir = os.path.dirname(os.path.abspath(__file__))
inotify_dir = os.path.normpath(testdir + '/../build/lib.{sys}-{plat}-{ver}/'.format(
    sys=un[0].lower(), plat=un[4], ver=ver))
idx = None
if os.path.exists(inotify_dir+'/inotify') and not inotify_dir in sys.path:
  # Insert at the beginning of sys.path, but not before the current directory
  # as we do not want to override an explicit inotify package in the current
  # directory.
  try:
    idx = next(i for i, p in enumerate(sys.path) if p and os.path.samefile(p, '.'))
  except StopIteration:
    # In interactive mode, there is no entry for the current directory, but the
    # first entry of sys.path is the empty string which is interpreted as
    # current directory. So if a path to the current directory is not found,
    # insert after this first empty string.
    idx = 0
  sys.path.insert(idx + 1, inotify_dir)
del un, ver, testdir, idx


import inotify
from inotify import watcher

print("\nTesting inotify module from", inotify.__file__)



@pytest.fixture(autouse=True)
def preparedir(request):
  # global tempdir
  tempdir = tempfile.mkdtemp(prefix='inotify-test-tmpdir-')
  request.addfinalizer(lambda tempdir=tempdir: shutil.rmtree(tempdir))
  os.chdir(tempdir)
  open('testfile', 'w').close()
  os.mkdir('testdir')


@pytest.fixture
def w():
  return watcher.Watcher()


def test_open(w):
  w.add('testfile', inotify.IN_OPEN | inotify.IN_CLOSE)
  open('testfile').close()
  ev1, ev2 = w.read(block=False)
  assert ev1.open
  assert ev2.close
  assert ev2.close_nowrite
  w.close()


def test_move(w):
  w.add('.', inotify.IN_MOVE)
  assert w.read(block=False) == []
  os.rename('testfile', 'targetfile')
  ev = w.read(block=False)
  for e in ev:
    if e.name == 'testfile':
      assert e.moved_from
    if e.name == 'targetfile':
      assert e.moved_to
  assert ev[0].cookie and ev[0].cookie == ev[1].cookie


def test_alias(w):
  '''The inotify system maps watch requests to aliases (e.g. symlinks) to the
  same watch descriptor, so we need to be sure that a watch is only really
  removed if all paths it is watching are dismissed.'''

  os.symlink('testfile', 'testlink')
  w1 = w.add('testfile', inotify.IN_OPEN)
  w2 = w.add('testlink', inotify.IN_CLOSE)
  assert w1 == w2
  assert set(w.paths()) == {'testfile', 'testlink'}
  assert w.get_watch('testfile') == w.get_watch('testlink')
  assert len(w.watches()) == 1
  open('testlink').close()
  ev1, ev2 = w.read(block=False)
  assert ev1.open and ev2.close
  w.remove_path('testfile')
  open('testlink').close()
  ev = w.read(block=False)
  assert any(e.close for e in ev)


def test_delete(w):
    w.add('testfile', inotify.IN_DELETE_SELF)
    os.remove('testfile')
    ev1, ev2 = w.read(block=False)
    assert ev1.delete_self
    assert ev2.ignored
    assert w.num_watches() == 0

def test_noent(w):
    with pytest.raises(OSError) as excinfo:
        w.add('nonexistant', inotify.IN_OPEN)
    assert excinfo.value.errno == os.errno.ENOENT
    with pytest.raises(OSError) as excinfo:
        w.add_all('nonexistant', inotify.IN_OPEN)
    assert excinfo.value.errno == os.errno.ENOENT

def test_removewatch(w):
  'test Watcher.remove_path and Watcher.remove_watch functionality'
  open('testfile2', 'w').close()
  open('testfile3', 'w').close()
  watch1 = w.add('testfile', inotify.IN_OPEN)
  watch2 = w.add('testfile2', inotify.IN_OPEN)
  watch3 = w.add('testfile3', inotify.IN_OPEN)
  open('testfile').close()
  open('testfile2').close()
  open('testfile3').close()
  evts = w.read()
  assert [e.fullpath for e in evts] == ['testfile', 'testfile2', 'testfile3']
  assert all(e.mask & inotify.IN_OPEN for e in evts)

  w.remove_path('testfile')
  w.read()
  open('testfile').close()
  open('testfile2').close()
  open('testfile3').close()
  evts = w.read()
  assert [e.fullpath for e in evts] == ['testfile2', 'testfile3']
  assert all(e.mask & inotify.IN_OPEN for e in evts)

  w.remove_watch(watch2)
  w.read()
  open('testfile').close()
  open('testfile2').close()
  open('testfile3').close()
  evts = w.read()
  assert [e.fullpath for e in evts] == ['testfile3']
  assert all(e.mask & inotify.IN_OPEN for e in evts)


def test_repr(w):
  w.add_all('.', inotify.IN_ALL_EVENTS)
  print(w.read(block=False))


def test_read_args(w):
  with pytest.raises(watcher.NoFilesException):
    w.read()
  w.add('.', inotify.IN_ALL_EVENTS)
  evts = inotify.inotify.read(w.fileno(), block=False)
  assert evts == []
  readfd, writefd = os.pipe()
  os.write(writefd, b'a'*100)
  with pytest.raises(TypeError):
    inotify.inotify.read(readfd, block=False)

  ## In the current implementation this succeeds, but it's not
  ## guaranteed or anything
  # os.write(writefd, b'\0'*32)
  # evts = inotify.inotify.read(readfd, block=False)
  # for e in evts:
  #   assert e.wd == e.mask == 0
  #   assert e.name == e.cookie == None

  os.close(writefd)
  os.close(readfd)
  

def test_kwarg(w):
  with pytest.raises(TypeError):
    inotify.inotify.read(w.fileno(), False)
<commit_msg>fix tests when a path doesn't exist<commit_after>#!/usr/bin/env py.test

# This testing script can be run either from python 3 or python 2. Run with
# `py.test test.py` or `py.test-2.7 test.py`.
#
# This script will try to import the inotify module from the build directory in
# ../build/lib.linux-{platform}-{pyversion}/inotify relative to its own
# location. If that directory cannot be found it will import the inotify module
# from the default path.


from __future__ import print_function

import sys, os, shutil, tempfile, inspect
import pytest

if not sys.platform.startswith('linux'): raise Exception("This module will only work on Linux")

# find the build dir
un = os.uname()
ver = '.'.join(str(x) for x in sys.version_info[:2])
testdir = os.path.dirname(os.path.abspath(__file__))
inotify_dir = os.path.normpath(testdir + '/../build/lib.{sys}-{plat}-{ver}/'.format(
    sys=un[0].lower(), plat=un[4], ver=ver))
idx = None
if os.path.exists(inotify_dir+'/inotify') and not inotify_dir in sys.path:
  # Insert at the beginning of sys.path, but not before the current directory
  # as we do not want to override an explicit inotify package in the current
  # directory.
  try:
    idx = next(i for i, p in enumerate(sys.path) if os.path.exists(p) and os.path.samefile(p, '.'))
  except StopIteration:
    # In interactive mode, there is no entry for the current directory, but the
    # first entry of sys.path is the empty string which is interpreted as
    # current directory. So if a path to the current directory is not found,
    # insert after this first empty string.
    idx = 0
  sys.path.insert(idx + 1, inotify_dir)
del un, ver, testdir, idx


import inotify
from inotify import watcher

print("\nTesting inotify module from", inotify.__file__)



@pytest.fixture(autouse=True)
def preparedir(request):
  # global tempdir
  tempdir = tempfile.mkdtemp(prefix='inotify-test-tmpdir-')
  request.addfinalizer(lambda tempdir=tempdir: shutil.rmtree(tempdir))
  os.chdir(tempdir)
  open('testfile', 'w').close()
  os.mkdir('testdir')


@pytest.fixture
def w():
  return watcher.Watcher()


def test_open(w):
  w.add('testfile', inotify.IN_OPEN | inotify.IN_CLOSE)
  open('testfile').close()
  ev1, ev2 = w.read(block=False)
  assert ev1.open
  assert ev2.close
  assert ev2.close_nowrite
  w.close()


def test_move(w):
  w.add('.', inotify.IN_MOVE)
  assert w.read(block=False) == []
  os.rename('testfile', 'targetfile')
  ev = w.read(block=False)
  for e in ev:
    if e.name == 'testfile':
      assert e.moved_from
    if e.name == 'targetfile':
      assert e.moved_to
  assert ev[0].cookie and ev[0].cookie == ev[1].cookie


def test_alias(w):
  '''The inotify system maps watch requests to aliases (e.g. symlinks) to the
  same watch descriptor, so we need to be sure that a watch is only really
  removed if all paths it is watching are dismissed.'''

  os.symlink('testfile', 'testlink')
  w1 = w.add('testfile', inotify.IN_OPEN)
  w2 = w.add('testlink', inotify.IN_CLOSE)
  assert w1 == w2
  assert set(w.paths()) == {'testfile', 'testlink'}
  assert w.get_watch('testfile') == w.get_watch('testlink')
  assert len(w.watches()) == 1
  open('testlink').close()
  ev1, ev2 = w.read(block=False)
  assert ev1.open and ev2.close
  w.remove_path('testfile')
  open('testlink').close()
  ev = w.read(block=False)
  assert any(e.close for e in ev)


def test_delete(w):
    w.add('testfile', inotify.IN_DELETE_SELF)
    os.remove('testfile')
    ev1, ev2 = w.read(block=False)
    assert ev1.delete_self
    assert ev2.ignored
    assert w.num_watches() == 0

def test_noent(w):
    with pytest.raises(OSError) as excinfo:
        w.add('nonexistant', inotify.IN_OPEN)
    assert excinfo.value.errno == os.errno.ENOENT
    with pytest.raises(OSError) as excinfo:
        w.add_all('nonexistant', inotify.IN_OPEN)
    assert excinfo.value.errno == os.errno.ENOENT

def test_removewatch(w):
  'test Watcher.remove_path and Watcher.remove_watch functionality'
  open('testfile2', 'w').close()
  open('testfile3', 'w').close()
  watch1 = w.add('testfile', inotify.IN_OPEN)
  watch2 = w.add('testfile2', inotify.IN_OPEN)
  watch3 = w.add('testfile3', inotify.IN_OPEN)
  open('testfile').close()
  open('testfile2').close()
  open('testfile3').close()
  evts = w.read()
  assert [e.fullpath for e in evts] == ['testfile', 'testfile2', 'testfile3']
  assert all(e.mask & inotify.IN_OPEN for e in evts)

  w.remove_path('testfile')
  w.read()
  open('testfile').close()
  open('testfile2').close()
  open('testfile3').close()
  evts = w.read()
  assert [e.fullpath for e in evts] == ['testfile2', 'testfile3']
  assert all(e.mask & inotify.IN_OPEN for e in evts)

  w.remove_watch(watch2)
  w.read()
  open('testfile').close()
  open('testfile2').close()
  open('testfile3').close()
  evts = w.read()
  assert [e.fullpath for e in evts] == ['testfile3']
  assert all(e.mask & inotify.IN_OPEN for e in evts)


def test_repr(w):
  w.add_all('.', inotify.IN_ALL_EVENTS)
  print(w.read(block=False))


def test_read_args(w):
  with pytest.raises(watcher.NoFilesException):
    w.read()
  w.add('.', inotify.IN_ALL_EVENTS)
  evts = inotify.inotify.read(w.fileno(), block=False)
  assert evts == []
  readfd, writefd = os.pipe()
  os.write(writefd, b'a'*100)
  with pytest.raises(TypeError):
    inotify.inotify.read(readfd, block=False)

  ## In the current implementation this succeeds, but it's not
  ## guaranteed or anything
  # os.write(writefd, b'\0'*32)
  # evts = inotify.inotify.read(readfd, block=False)
  # for e in evts:
  #   assert e.wd == e.mask == 0
  #   assert e.name == e.cookie == None

  os.close(writefd)
  os.close(readfd)


def test_kwarg(w):
  with pytest.raises(TypeError):
    inotify.inotify.read(w.fileno(), False)

{'fraction_non_alphanumeric': 0.14389179755671902, 'fraction_numerical': 0.009947643979057591, 'mean_word_length': 3.9464825205006475, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1943411', 'n_tokens_mistral': 4275, 'n_tokens_neox': 4222, 'n_words': 1086}