Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_watcher_auto_root() -> None:
watcher = Watcher()
assert watcher.root == [os.path.realpath(sys.argv[0].rsplit('/', 1)[0])]
def test_watcher_configurable_ignored_directory() -> None:
root_path = '{}/tests/watcher_root/configurable_ignored'.format(os.path.realpath(os.getcwd()))
watcher = Watcher(root=[root_path])
assert len(watcher.root) == 1
assert isinstance(watcher.watched_files, dict)
assert len(watcher.watched_files) == 1
watcher = Watcher(root=[root_path], configuration={'options': {'watcher': {'ignored_dirs': ['configurable_ignored']}}})
assert len(watcher.root) == 1
assert isinstance(watcher.watched_files, dict)
assert len(watcher.watched_files) == 0
def test_watcher_configurable_ignored_directory() -> None:
root_path = '{}/tests/watcher_root/configurable_ignored'.format(os.path.realpath(os.getcwd()))
watcher = Watcher(root=[root_path])
assert len(watcher.root) == 1
assert isinstance(watcher.watched_files, dict)
assert len(watcher.watched_files) == 1
watcher = Watcher(root=[root_path], configuration={'options': {'watcher': {'ignored_dirs': ['configurable_ignored']}}})
assert len(watcher.root) == 1
assert isinstance(watcher.watched_files, dict)
assert len(watcher.watched_files) == 0
def test_watcher_empty_directory() -> None:
root_path = '{}/tests/watcher_root/empty'.format(os.path.realpath(os.getcwd()))
watcher = Watcher(root=[root_path])
assert len(watcher.root) == 1
assert isinstance(watcher.watched_files, dict)
assert len(watcher.watched_files) == 0
def test_watcher_default_ignored_directory() -> None:
root_path = '{}/tests/watcher_root/__tmp__'.format(os.path.realpath(os.getcwd()))
watcher = Watcher(root=[root_path])
assert len(watcher.root) == 1
assert isinstance(watcher.watched_files, dict)
assert len(watcher.watched_files) == 0
def test_watcher_callback(loop: Any) -> None:
root_path = '{}/tests/watcher_root'.format(os.path.realpath(os.getcwd()))
watcher = Watcher(root=[root_path])
assert len(watcher.root) == 1
assert isinstance(watcher.watched_files, dict)
assert len(watcher.watched_files) == 2
result = watcher.update_watched_files()
assert result == {}
watcher.watched_files = {'_test': 0}
watcher.watched_files_crc = {'_test': ''}
result = watcher.update_watched_files(reindex=True)
assert len(result.get('added', 0)) == 2
assert len(result.get('removed', 0)) == 1
assert len(result.get('updated', 0)) == 0
class Test():
callbacks_run = {} # type: Dict[int, bool]
print('Invalid config file: {}'.format(str(e)))
sys.exit(2)
except ValueError as e:
print('Invalid config file, invalid JSON format: {}'.format(str(e)))
sys.exit(2)
if '--production' in args:
index = args.index('--production')
args.pop(index)
watcher = None
else:
cwd = os.getcwd()
root_directories = [os.getcwd()]
for arg in set(args):
root_directories.append(os.path.dirname('{}/{}'.format(os.path.realpath(cwd), arg)))
watcher = Watcher(root=root_directories, configuration=configuration)
if '-l' in args or '--log' in args:
index = args.index('-l') if '-l' in args else args.index('--log')
args.pop(index)
if len(args) > index:
log_level = getattr(logging, args.pop(index).upper(), None) or log_level
logging.basicConfig(format='%(asctime)s (%(name)s): %(message)s', level=log_level)
logging.Formatter(fmt='%(asctime)s.%(msecs).03d', datefmt='%Y-%m-%d %H:%M:%S')
self.test_dependencies()
ServiceLauncher.run_until_complete(set(args), configuration, watcher)
sys.exit(0)