# Path: streamcorpus_pipeline/_local_storage.py
# class from_local_chunks(Configured):
#     '''
#     may use config['max_retries'] (default 1)
#     may use config['max_backoff'] (seconds, default 300)
#     may use config['streamcorpus_version'] (default 'v0_3_0')
#     '''
#     config_name = 'from_local_chunks'
#     default_config = {
#         'max_backoff': 300,
#         'streamcorpus_version': 'v0_3_0',
#     }
# 
#     @staticmethod
#     def check_config(config, name):
#         _check_version(config, name)
# 
#     def __call__(self, i_str):
#         backoff = 0.1
#         start_time = time.time()
#         tries = 0
#         max_retries = int(self.config.get('max_retries', 1))
#         last_exc = None
#         while tries < max_retries:
#             try:
#                 message = _message_versions[self.config['streamcorpus_version']]
#                 logger.debug('reading from %r' % i_str)
#                 chunk = streamcorpus.Chunk(path=i_str, mode='rb', message=message)
#                 return chunk
#             except IOError, exc:
#                 if exc.errno == errno.ENOENT:
#                     logger.critical('File is missing?  Assume is slow NFS, try %d more times',
#                                     max_retries - tries)
#                     backoff *= 2
#                     tries += 1
#                     elapsed = time.time() - start_time
#                     if elapsed > self.config['max_backoff']:
#                         ## give up after five minutes of retries
#                         logger.critical('File %r not found after %d retries', i_str, tries)
#                         raise
#                     time.sleep(backoff)
#                     last_exc = exc
#                 else:
#                     logger.critical('failed loading %r', i_str, exc_info=True)
#                     raise
# 
#         ## exceeded max_retries
#         logger.critical('File %r not found after %d retries', i_str, tries)
#         if last_exc:
#             raise last_exc
#         raise IOError('File %r not found after %d retries' % (i_str, tries))
# 
# class from_local_files(Configured):
#     '''Read plain text or other raw content from local files.
# 
#     Each input filename is taken as an actual filename.  This produces
#     one stream item per file.  This can be configured as follows:
# 
#     .. code-block:: yaml
# 
#         abs_url: null
#         url_prefix: file://
#         absolute_filename: true
#         epoch_ticks: file
#         encoding: null
# 
#     If `abs_url` is provided in configuration, this value is used
#     as-is as the corresponding field in the stream item.  Otherwise,
#     the path is converted to an absolute path if `absolute_filename`
#     is true, and then `url_prefix` is prepended to it.
# 
#     `epoch_ticks` indicates the corresponding time stamp in the stream
#     item.  This may be an integer for specific seconds since the Unix
#     epoch, the string ``file`` indicating to use the last-modified
#     time of the file, or the string ``now`` for the current time.
# 
#     ``encoding`` should be set if you know the encoding of the files
#     you're loading. e.g., ``utf-8``.
# 
#     The values shown above are the defaults.  If reading in files
#     in a directory structure matching a URL hierarchy, an alternate
#     configuration could be
# 
#     .. code-block:: yaml
# 
#         url_prefix: "http://home.example.com/"
#         absolute_path: false
# 
#     '''
#     config_name = 'from_local_files'
#     default_config = {
#         'abs_url': None,
#         'url_prefix': 'file://',
#         'absolute_filename': True,
#         'epoch_ticks': 'file',
#         'encoding': None,
#     }
# 
#     def __init__(self, config):
#         super(from_local_files, self).__init__(config)
#         self.abs_url = self.config.get('abs_url', None)
#         self.url_prefix = self.config.get('url_prefix', '')
#         self.absolute_filename = self.config.get('absolute_filename', False)
#         self.epoch_ticks = self.config.get('epoch_ticks', 'file')
#         self.encoding = self.config.get('encoding')
# 
#         if ((not isinstance(self.epoch_ticks, int) and
#              self.epoch_ticks not in ['file', 'now'])):
#             self.epoch_ticks = 'file'
# 
#     def __call__(self, i_str):
#         if self.epoch_ticks == 'file':
#             st = os.stat(i_str)
#             epoch_ticks = st.st_mtime
#         elif self.epoch_ticks == 'now':
#             epoch_ticks = time.time()
#         else:
#             epoch_ticks = self.epoch_ticks
# 
#         if self.abs_url:
#             abs_url = self.abs_url
#         else:
#             abs_url = i_str
#             if self.absolute_filename:
#                 abs_url = os.path.abspath(abs_url)
#             if self.url_prefix:
#                 abs_url = self.url_prefix + abs_url
# 
#         si = streamcorpus.make_stream_item(epoch_ticks, abs_url)
#         with open(i_str, 'rb') as f:
#             si.body.raw = f.read()
#         if self.encoding is not None:
#             si.body.encoding = self.encoding
# 
#         yield si


import errno
import os
import time
import pytest
from streamcorpus_pipeline._local_storage import from_local_chunks, \
    from_local_files
from __future__ import absolute_import




def test_max_retries():
    flc = from_local_chunks(config={
        'max_retries': 5,
        'max_backoff': 100,
        'streamcorpus_version': 'v0_2_0',
    })
    start_time = time.time()
    with pytest.raises(IOError) as excinfo:  # pylint: disable=E1101
        flc('no-such-path')
    elapsed = time.time() - start_time
    assert elapsed > 0.1 * 2**4
    assert excinfo.value.errno == errno.ENOENT


def test_max_backoff():

# Path: streamcorpus_pipeline/_local_storage.py
# class from_local_chunks(Configured):
#     '''
#     may use config['max_retries'] (default 1)
#     may use config['max_backoff'] (seconds, default 300)
#     may use config['streamcorpus_version'] (default 'v0_3_0')
#     '''
#     config_name = 'from_local_chunks'
#     default_config = {
#         'max_backoff': 300,
#         'streamcorpus_version': 'v0_3_0',
#     }
# 
#     @staticmethod
#     def check_config(config, name):
#         _check_version(config, name)
# 
#     def __call__(self, i_str):
#         backoff = 0.1
#         start_time = time.time()
#         tries = 0
#         max_retries = int(self.config.get('max_retries', 1))
#         last_exc = None
#         while tries < max_retries:
#             try:
#                 message = _message_versions[self.config['streamcorpus_version']]
#                 logger.debug('reading from %r' % i_str)
#                 chunk = streamcorpus.Chunk(path=i_str, mode='rb', message=message)
#                 return chunk
#             except IOError, exc:
#                 if exc.errno == errno.ENOENT:
#                     logger.critical('File is missing?  Assume is slow NFS, try %d more times',
#                                     max_retries - tries)
#                     backoff *= 2
#                     tries += 1
#                     elapsed = time.time() - start_time
#                     if elapsed > self.config['max_backoff']:
#                         ## give up after five minutes of retries
#                         logger.critical('File %r not found after %d retries', i_str, tries)
#                         raise
#                     time.sleep(backoff)
#                     last_exc = exc
#                 else:
#                     logger.critical('failed loading %r', i_str, exc_info=True)
#                     raise
# 
#         ## exceeded max_retries
#         logger.critical('File %r not found after %d retries', i_str, tries)
#         if last_exc:
#             raise last_exc
#         raise IOError('File %r not found after %d retries' % (i_str, tries))
# 
# class from_local_files(Configured):
#     '''Read plain text or other raw content from local files.
# 
#     Each input filename is taken as an actual filename.  This produces
#     one stream item per file.  This can be configured as follows:
# 
#     .. code-block:: yaml
# 
#         abs_url: null
#         url_prefix: file://
#         absolute_filename: true
#         epoch_ticks: file
#         encoding: null
# 
#     If `abs_url` is provided in configuration, this value is used
#     as-is as the corresponding field in the stream item.  Otherwise,
#     the path is converted to an absolute path if `absolute_filename`
#     is true, and then `url_prefix` is prepended to it.
# 
#     `epoch_ticks` indicates the corresponding time stamp in the stream
#     item.  This may be an integer for specific seconds since the Unix
#     epoch, the string ``file`` indicating to use the last-modified
#     time of the file, or the string ``now`` for the current time.
# 
#     ``encoding`` should be set if you know the encoding of the files
#     you're loading. e.g., ``utf-8``.
# 
#     The values shown above are the defaults.  If reading in files
#     in a directory structure matching a URL hierarchy, an alternate
#     configuration could be
# 
#     .. code-block:: yaml
# 
#         url_prefix: "http://home.example.com/"
#         absolute_path: false
# 
#     '''
#     config_name = 'from_local_files'
#     default_config = {
#         'abs_url': None,
#         'url_prefix': 'file://',
#         'absolute_filename': True,
#         'epoch_ticks': 'file',
#         'encoding': None,
#     }
# 
#     def __init__(self, config):
#         super(from_local_files, self).__init__(config)
#         self.abs_url = self.config.get('abs_url', None)
#         self.url_prefix = self.config.get('url_prefix', '')
#         self.absolute_filename = self.config.get('absolute_filename', False)
#         self.epoch_ticks = self.config.get('epoch_ticks', 'file')
#         self.encoding = self.config.get('encoding')
# 
#         if ((not isinstance(self.epoch_ticks, int) and
#              self.epoch_ticks not in ['file', 'now'])):
#             self.epoch_ticks = 'file'
# 
#     def __call__(self, i_str):
#         if self.epoch_ticks == 'file':
#             st = os.stat(i_str)
#             epoch_ticks = st.st_mtime
#         elif self.epoch_ticks == 'now':
#             epoch_ticks = time.time()
#         else:
#             epoch_ticks = self.epoch_ticks
# 
#         if self.abs_url:
#             abs_url = self.abs_url
#         else:
#             abs_url = i_str
#             if self.absolute_filename:
#                 abs_url = os.path.abspath(abs_url)
#             if self.url_prefix:
#                 abs_url = self.url_prefix + abs_url
# 
#         si = streamcorpus.make_stream_item(epoch_ticks, abs_url)
#         with open(i_str, 'rb') as f:
#             si.body.raw = f.read()
#         if self.encoding is not None:
#             si.body.encoding = self.encoding
# 
#         yield si


# Path: streamcorpus_pipeline/tests/test_local_storage.py
import errno
import os
import time
import pytest
from streamcorpus_pipeline._local_storage import from_local_chunks, \
    from_local_files

from __future__ import absolute_import




def test_max_retries():
    flc = from_local_chunks(config={
        'max_retries': 5,
        'max_backoff': 100,
        'streamcorpus_version': 'v0_2_0',
    })
    start_time = time.time()
    with pytest.raises(IOError) as excinfo:  # pylint: disable=E1101
        flc('no-such-path')
    elapsed = time.time() - start_time
    assert elapsed > 0.1 * 2**4
    assert excinfo.value.errno == errno.ENOENT


def test_max_backoff():

    flc = from_local_chunks(config={