Skip to content

Commit

Permalink
Fixed style issues and complains about exceptions with description.
Browse files Browse the repository at this point in the history
  • Loading branch information
TobiasRzepka committed Oct 3, 2024
1 parent 674af5f commit e75b680
Showing 1 changed file with 62 additions and 70 deletions.
132 changes: 62 additions & 70 deletions tests/test_single_file_watched.py
Original file line number Diff line number Diff line change
@@ -1,28 +1,26 @@
# coding: utf-8
"""Test cases when a single file is subscribed."""

""" Test cases when a single file is subscribed. """

from os import mkdir
import time
from os import mkdir

from .utils import P, StartWatching, TestEventQueue

SLEEP = 0.001


def test_selftest(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
""" Check if all events are catched in SLEEP time. """
"""Check if all events are catched in SLEEP time."""

emitter = start_watching(path=p()) # Pretty sure this should work

with open(p('file1.bak'), 'wt') as fp:
fp.write('test1')
with open(p('file2.bak'), 'wt') as fp:
fp.write('test2')
with open(p("file1.bak"), "w") as fp:
fp.write("test1")

with open(p("file2.bak"), "w") as fp:
fp.write("test2")

time.sleep(SLEEP)

found_files = set()
try:
while event_queue.qsize():
Expand All @@ -33,101 +31,95 @@ def test_selftest(p: P, start_watching: StartWatching, event_queue: TestEventQue
found_files.add(event.src_path)
finally:
emitter.stop()
assert len(found_files) == 2, 'Number of expected files differ. Increase sleep time.'

assert len(found_files) == 2, "Number of expected files differ. Increase sleep time."


def test_file_access(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
""" Check if file fires events. """
"""Check if file fires events."""

file1 = 'file1.bak'
file1 = "file1.bak"
tmpfile = p(file1)

with open(tmpfile, 'wt') as fp:
fp.write('init1')
with open(tmpfile, "w") as fp:
fp.write("init1")

emitter = start_watching(path=tmpfile)

# This is what we want to see
with open(tmpfile, 'wt') as fp:
fp.write('test1')
with open(tmpfile, "w") as fp:
fp.write("test1")

time.sleep(SLEEP)

try:
while event_queue.qsize():
event, _ = event_queue.get()
if event.is_directory:
assert event.src_path == p()
continue
else:
assert event.src_path.endswith(file1)
assert event.src_path.endswith(file1)
break
else:
assert False, 'No event catched.'
raise AssertionError # No event catched
finally:
emitter.stop()


def test_file_access_multiple(
p: P, start_watching: StartWatching, event_queue: TestEventQueue
) -> None:
""" Check if file fires events multiple times. """
def test_file_access_multiple(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
"""Check if file fires events multiple times."""

file1 = 'file1.bak'
file1 = "file1.bak"
tmpfile = p(file1)

with open(tmpfile, 'wt') as fp:
fp.write('init1')
with open(tmpfile, "w") as fp:
fp.write("init1")

emitter = start_watching(path=tmpfile)

try:
for i in range(5):
for _i in range(5):
# This is what we want to see multiple times
with open(tmpfile, 'wt') as fp:
fp.write('test1')
with open(tmpfile, "w") as fp:
fp.write("test1")

time.sleep(SLEEP)

while event_queue.qsize():
event, _ = event_queue.get()
if event.is_directory:
assert event.src_path == p()
continue
else:
assert event.src_path.endswith(file1)
assert event.src_path.endswith(file1)
break
else:
assert False, 'No event catched.'
raise AssertionError # No event catched

finally:
emitter.stop()


def test_file_access_other_file(
p: P, start_watching: StartWatching, event_queue: TestEventQueue
) -> None:
""" Check if other files doesn't fires events. """
def test_file_access_other_file(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
"""Check if other files doesn't fires events."""

file1 = 'file1.bak'
file1 = "file1.bak"
tmpfile = p(file1)

with open(tmpfile, 'wt') as fp:
fp.write('init1')
with open(tmpfile, "w") as fp:
fp.write("init1")

emitter = start_watching(path=tmpfile)

# Don't wanted
with open(p('file2.bak'), 'wt') as fp:
fp.write('test2')
with open(p("file2.bak"), "w") as fp:
fp.write("test2")

# but this
with open(tmpfile, 'wt') as fp:
fp.write('test1')
with open(tmpfile, "w") as fp:
fp.write("test1")

time.sleep(SLEEP)

found_files = set()
try:
while event_queue.qsize():
Expand All @@ -139,32 +131,32 @@ def test_file_access_other_file(
assert event.src_path.endswith(file1)
finally:
emitter.stop()
assert len(found_files) == 1, 'Number of expected files differ. Wrong file catched.'

assert len(found_files) == 1, "Number of expected files differ. Wrong file catched."


def test_create_folder(p: P, start_watching: StartWatching, event_queue: TestEventQueue) -> None:
""" Check if creation of a directory and inside files doesn't fires events. """
"""Check if creation of a directory and inside files doesn't fires events."""

file1 = 'file1.bak'
file1 = "file1.bak"
tmpfile = p(file1)

with open(tmpfile, 'wt') as fp:
fp.write('init1')
with open(tmpfile, "w") as fp:
fp.write("init1")

emitter = start_watching(path=tmpfile)

# Don't wanted
mkdir(p('myfolder'))
with open(p('myfolder/file2.bak'), 'wt') as fp:
fp.write('test2')
mkdir(p("myfolder"))
with open(p("myfolder/file2.bak"), "w") as fp:
fp.write("test2")

# but this
with open(tmpfile, 'wt') as fp:
fp.write('test1')
with open(tmpfile, "w") as fp:
fp.write("test1")

time.sleep(SLEEP)

found_files = set()
try:
while event_queue.qsize():
Expand All @@ -176,5 +168,5 @@ def test_create_folder(p: P, start_watching: StartWatching, event_queue: TestEve
assert event.src_path.endswith(file1)
finally:
emitter.stop()
assert len(found_files) == 1, 'Number of expected files differ. Wrong file catched.'

assert len(found_files) == 1, "Number of expected files differ. Wrong file catched."

0 comments on commit e75b680

Please sign in to comment.