# Path: datapackage_pipelines/wrapper/wrapper.py
# def spew(dp, resources_iterator, stats=None, finalizer=None):
#     files = [stdout]
# 
#     cache_filename = ''
#     if len(cache) > 0:
#         if not os.path.exists('.cache'):
#             os.mkdir('.cache')
#         cache_filename = os.path.join('.cache', cache)
#         files.append(gzip.open(cache_filename+'.ongoing', 'wt'))
# 
#     expected_resources = \
#         len(list(filter(streaming, dp.get('resources', []))))
#     row_count = 0
#     try:
#         for f in files:
#             f.write(json.dumps(dp, sort_keys=True, ensure_ascii=True)+'\n')
#             f.flush()
#         num_resources = 0
#         for res in resources_iterator:
#             if hasattr(res, 'it') and res.it is None:
#                 continue
#             num_resources += 1
#             for f in files:
#                 f.write('\n')
#             try:
#                 for rec in res:
#                     try:
#                         line = json.dumpl(rec,
#                                         sort_keys=True,
#                                         ensure_ascii=True)
#                     except TypeError as e:
#                         logging.error('Failed to encode row to JSON: %s\nOffending row: %r', e, rec)
#                         raise
#                     # logging.error('SPEWING: {}'.format(line))
#                     for f in files:
#                         f.write(line+'\n')
#                     # logging.error('WROTE')
#                     row_count += 1
#             except CastError as e:
#                 for err in e.errors:
#                     logging.error('Failed to cast row: %s', err)
#                 raise
#         if num_resources != expected_resources:
#             logging.error('Expected to see %d resource(s) but spewed %d',
#                           expected_resources, num_resources)
#             assert num_resources == expected_resources
# 
#         aggregated_stats = {}
#         if not first:
#             stats_line = sys.stdin.readline().strip()
#             if len(stats_line) > 0:
#                 try:
#                     aggregated_stats = json.loads(stats_line)
#                     assert aggregated_stats is None or \
#                         isinstance(aggregated_stats, dict)
#                 except json.JSONDecodeError:
#                     logging.error('Failed to parse stats: %r', stats_line)
#         if stats is not None:
#             aggregated_stats.update(stats)
#         stats_json = json.dumps(aggregated_stats,
#                                 sort_keys=True,
#                                 ensure_ascii=True)
#         for f in files:
#             f.write('\n'+stats_json)
# 
#     except BrokenPipeError:
#         logging.error('Output pipe disappeared!')
#         sys.stderr.flush()
#         sys.exit(1)
# 
#     stdout.flush()
#     if row_count > 0:
#         logging.info('Processed %d rows', row_count)
# 
#     if finalizer is not None:
#         finalizer()
# 
#     for f in files:
#         f.write('\n')  # Signal to other processors that we're done
#         if f == stdout:
#             # Can't close sys.stdout, otherwise any subsequent
#             # call to print() will throw an exception
#             f.flush()
#         else:
#             f.close()
# 
#     if len(cache) > 0:
#         os.rename(cache_filename+'.ongoing', cache_filename)


import unittest.mock as mock
from datapackage_pipelines.wrapper import spew


class TestWrapper(object):
    def test_spew_finalizer_runs_before_we_signal_that_were_done(self):
        '''Assert that the finalizer param is executed before spew is finished.

        We signal to other processors that we're done by writing an empty line
        to STDOUT. The finalizer parameter to spew() must be executed before that,
        as there can be processors that depend on us finishing our processing
        before they're able to run. For example, a processor that depends on
        `dump_to_zip` must wait until it has finished writing to the local
        filesystem.
        '''
        datapackage = {}
        resources_iterator = iter([])

        with mock.patch('datapackage_pipelines.wrapper.wrapper.stdout') as stdout_mock:
            def finalizer():
                last_call_args = stdout_mock.write.call_args_list[-1]
                assert last_call_args != mock.call('\n')


# Path: datapackage_pipelines/wrapper/wrapper.py
# def spew(dp, resources_iterator, stats=None, finalizer=None):
#     files = [stdout]
# 
#     cache_filename = ''
#     if len(cache) > 0:
#         if not os.path.exists('.cache'):
#             os.mkdir('.cache')
#         cache_filename = os.path.join('.cache', cache)
#         files.append(gzip.open(cache_filename+'.ongoing', 'wt'))
# 
#     expected_resources = \
#         len(list(filter(streaming, dp.get('resources', []))))
#     row_count = 0
#     try:
#         for f in files:
#             f.write(json.dumps(dp, sort_keys=True, ensure_ascii=True)+'\n')
#             f.flush()
#         num_resources = 0
#         for res in resources_iterator:
#             if hasattr(res, 'it') and res.it is None:
#                 continue
#             num_resources += 1
#             for f in files:
#                 f.write('\n')
#             try:
#                 for rec in res:
#                     try:
#                         line = json.dumpl(rec,
#                                         sort_keys=True,
#                                         ensure_ascii=True)
#                     except TypeError as e:
#                         logging.error('Failed to encode row to JSON: %s\nOffending row: %r', e, rec)
#                         raise
#                     # logging.error('SPEWING: {}'.format(line))
#                     for f in files:
#                         f.write(line+'\n')
#                     # logging.error('WROTE')
#                     row_count += 1
#             except CastError as e:
#                 for err in e.errors:
#                     logging.error('Failed to cast row: %s', err)
#                 raise
#         if num_resources != expected_resources:
#             logging.error('Expected to see %d resource(s) but spewed %d',
#                           expected_resources, num_resources)
#             assert num_resources == expected_resources
# 
#         aggregated_stats = {}
#         if not first:
#             stats_line = sys.stdin.readline().strip()
#             if len(stats_line) > 0:
#                 try:
#                     aggregated_stats = json.loads(stats_line)
#                     assert aggregated_stats is None or \
#                         isinstance(aggregated_stats, dict)
#                 except json.JSONDecodeError:
#                     logging.error('Failed to parse stats: %r', stats_line)
#         if stats is not None:
#             aggregated_stats.update(stats)
#         stats_json = json.dumps(aggregated_stats,
#                                 sort_keys=True,
#                                 ensure_ascii=True)
#         for f in files:
#             f.write('\n'+stats_json)
# 
#     except BrokenPipeError:
#         logging.error('Output pipe disappeared!')
#         sys.stderr.flush()
#         sys.exit(1)
# 
#     stdout.flush()
#     if row_count > 0:
#         logging.info('Processed %d rows', row_count)
# 
#     if finalizer is not None:
#         finalizer()
# 
#     for f in files:
#         f.write('\n')  # Signal to other processors that we're done
#         if f == stdout:
#             # Can't close sys.stdout, otherwise any subsequent
#             # call to print() will throw an exception
#             f.flush()
#         else:
#             f.close()
# 
#     if len(cache) > 0:
#         os.rename(cache_filename+'.ongoing', cache_filename)


# Path: tests/wrapper/test_wrapper.py
import unittest.mock as mock
from datapackage_pipelines.wrapper import spew



class TestWrapper(object):
    def test_spew_finalizer_runs_before_we_signal_that_were_done(self):
        '''Assert that the finalizer param is executed before spew is finished.

        We signal to other processors that we're done by writing an empty line
        to STDOUT. The finalizer parameter to spew() must be executed before that,
        as there can be processors that depend on us finishing our processing
        before they're able to run. For example, a processor that depends on
        `dump_to_zip` must wait until it has finished writing to the local
        filesystem.
        '''
        datapackage = {}
        resources_iterator = iter([])

        with mock.patch('datapackage_pipelines.wrapper.wrapper.stdout') as stdout_mock:
            def finalizer():
                last_call_args = stdout_mock.write.call_args_list[-1]
                assert last_call_args != mock.call('\n')


            spew(datapackage, resources_iterator, finalizer=finalizer)