Skip to content
Snippets Groups Projects
Commit ffbe921a authored by Piotr Maślanka's avatar Piotr Maślanka
Browse files

DirectorySource will raise an exception if directory does not exist and on_fail is set to RAISE

parent f4491ea6
No related branches found
No related tags found
No related merge requests found
......@@ -2,6 +2,7 @@
* added safe_listdir
* fixed a bug occurring in Python 3.10 with whereis
* DirectorySource will raise an exception if directory does not exist and on_fail is set to RAISE
Build system
============
......
__version__ = '2.25.0a2'
__version__ = '2.25.0a3'
......@@ -30,9 +30,9 @@ class FileSource(BaseSource):
"""
super().__init__()
from .. import sources
self.source_classes = [ # type: tp.List[tp.Type[FormatSource]]
self.source_classes = [
(p if not isinstance(p, str) else getattr(sources, p)) for p in
interpret_as]
interpret_as] # type: tp.List[tp.Type[FormatSource]]
self.path = path # type: str
self.encoding = encoding # type: str
......@@ -93,10 +93,13 @@ class DirectorySource(FileSource):
try:
files = self.filter(os.path.join(directory, x) for x in os.listdir(directory))
except OSError as e:
logger.warning(
'OSError %s while accessing configuration directory %s, skipping files' % (
e, directory))
return []
if self.on_fail == DirectorySource.SILENT:
logger.warning(
'OSError %s while accessing configuration directory %s, skipping files' % (
e, directory))
return []
else:
raise ConfigurationError(f'Directory {directory} does not exist')
for file_name in files:
......
......@@ -5,6 +5,7 @@ import unittest
from satella.coding import silence_excs
from satella.configuration.sources import FileSource, FORMAT_SOURCES, \
DirectorySource
from satella.exceptions import ConfigurationError
from .utils import SourceTestCase
......@@ -54,6 +55,12 @@ class TestDirectorySource(SourceTestCase):
self.assertEqual(self.ds.provide(),
{'amqp': 'amqp', 'logstash': {'host': 'localhost', 'port': 9600}})
def test_invalid_directory(self):
ds = DirectorySource('not-existing', on_fail=DirectorySource.SILENT)
self.assertEqual(ds.provide(), {})
ds = DirectorySource('not-existing', on_fail=DirectorySource.RAISE)
self.assertRaises(ConfigurationError, ds.provide)
def test_directory_source(self):
with tempfile.TemporaryDirectory() as outdir:
self.filename = outdir
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment