# Path: featureflow/bytestream.py
# class BytesWithTotalLength(bytes):
#     def __new__(cls, s, total_length):
#         o = bytes.__new__(cls, s)
#         o.total_length = int(total_length)
#         return o
# 
#     def __radd__(self, other):
#         return BytesWithTotalLength(other + bytes(self), self.total_length)
# 
# class ByteStream(Node):
#     def __init__(self, chunksize=4096, needs=None):
#         super(ByteStream, self).__init__(needs=needs)
#         self._chunksize = int(chunksize)
# 
#     def _generator(self, stream, content_length):
#         if not content_length:
#             raise ValueError('content_length should be greater than zero')
#         for chunk in chunked(stream, chunksize=self._chunksize):
#             yield BytesWithTotalLength(chunk, content_length)
# 
#     def _from_http_response(self, resp):
#         resp.raise_for_status()
#         content_length = int(resp.headers['Content-Length'])
#         return self._generator(resp.raw, content_length)
# 
#     def _handle_simple_get(self, data):
#         parsed = urlparse(data)
#         if parsed.scheme and parsed.netloc:
#             resp = requests.get(data, stream=True)
#             return self._from_http_response(resp)
#         else:
#             raise ValueError
# 
#     def _handle_http_request(self, data):
#         s = requests.Session()
#         prepped = data.prepare()
#         resp = s.send(prepped, stream=True)
#         return self._from_http_response(resp)
# 
#     def _handle_file_like_object(self, data):
#         content_length = data.seek(0, 2)
#         data.seek(0)
#         return self._generator(data, content_length)
# 
#     def _handle_zip_file(self, data):
#         return self._generator(data.zipfile, data.file_size)
# 
#     def _handle_file(self, data):
#         with open(data, 'rb') as f:
#             content_length = int(os.path.getsize(data))
#             for chunk in self._generator(f, content_length):
#                 yield chunk
# 
#     def _get_strategy(self, data):
#         if isinstance(data, ZipWrapper):
#             return self._handle_zip_file
#         if isinstance(data, requests.Request):
#             return self._handle_http_request
#         if isinstance(data, str):
#             parsed = urlparse(data)
#             is_url = parsed.netloc and parsed.scheme
#             return self._handle_simple_get if is_url else self._handle_file
#         return self._handle_file_like_object
# 
#     def _process(self, data):
#         try:
#             data = data.uri
#         except AttributeError:
#             pass
#         strategy = self._get_strategy(data)
#         for chunk in strategy(data):
#             yield chunk
# 
# class ZipWrapper(object):
#     def __init__(self, zipfile, zipinfo):
#         self.zipinfo = zipinfo
#         self.zipfile = zipfile
# 
#     def __enter__(self):
#         self.zipfile.__enter__()
#         return self
# 
#     def __exit__(self, exc_type, exc_val, exc_tb):
#         self.zipfile.__exit__(exc_type, exc_val, exc_tb)
# 
#     @property
#     def file_size(self):
#         return self.zipinfo.file_size
# 
#     @property
#     def filename(self):
#         return self.zipinfo.filename
# 
# def iter_zip(fn):
#     with zipfile.ZipFile(fn) as zf:
#         for info in zf.filelist:
#             if not info.file_size:
#                 continue
#             f = zf.open(info.filename)
#             yield ZipWrapper(f, info)
# 
# Path: featureflow/util.py
# def wait_for_http_server(host, port, wait_time_seconds=0.1, max_tries=100):
#     for _ in range(max_tries):
#         try:
#             resp = requests.get(f'http://{host}:{port}')
#             return resp
#         except requests.exceptions.ConnectionError:
#             time.sleep(wait_time_seconds)
#             continue
#     raise RuntimeError(f'Failed to start HTTP server at {host}:{port}')


from .bytestream import BytesWithTotalLength, ByteStream, ZipWrapper, iter_zip
from io import BytesIO
from collections import namedtuple
from uuid import uuid4
from .util import wait_for_http_server
import unittest2
import sys
import tempfile
import subprocess
import requests
import os
import zipfile
        self.process = subprocess.Popen(
            [sys.executable, server, self.port, self.expected],
            stdout=devnull,
            stderr=devnull)
        wait_for_http_server('localhost', '9876')

    def tearDown(self):
        self.process.kill()

    def results(self, inp):
        return b''.join(self.bytestream._process(inp))

    def local_url(self):
        return 'http://localhost:{port}'.format(**self.__dict__)

    def get_request(self):
        return requests.Request(
            method='GET',
            url=self.local_url())

    def test_can_accept_chunksize_implementing_int(self):
        class SemanticChunksize(object):
            THING_SIZE_BYTES = 64

            def __init__(self, n_things=10):
                self.n_things = n_things

            def __int__(self):
                return self.n_things * self.THING_SIZE_BYTES


# Path: featureflow/bytestream.py
# class BytesWithTotalLength(bytes):
#     def __new__(cls, s, total_length):
#         o = bytes.__new__(cls, s)
#         o.total_length = int(total_length)
#         return o
# 
#     def __radd__(self, other):
#         return BytesWithTotalLength(other + bytes(self), self.total_length)
# 
# class ByteStream(Node):
#     def __init__(self, chunksize=4096, needs=None):
#         super(ByteStream, self).__init__(needs=needs)
#         self._chunksize = int(chunksize)
# 
#     def _generator(self, stream, content_length):
#         if not content_length:
#             raise ValueError('content_length should be greater than zero')
#         for chunk in chunked(stream, chunksize=self._chunksize):
#             yield BytesWithTotalLength(chunk, content_length)
# 
#     def _from_http_response(self, resp):
#         resp.raise_for_status()
#         content_length = int(resp.headers['Content-Length'])
#         return self._generator(resp.raw, content_length)
# 
#     def _handle_simple_get(self, data):
#         parsed = urlparse(data)
#         if parsed.scheme and parsed.netloc:
#             resp = requests.get(data, stream=True)
#             return self._from_http_response(resp)
#         else:
#             raise ValueError
# 
#     def _handle_http_request(self, data):
#         s = requests.Session()
#         prepped = data.prepare()
#         resp = s.send(prepped, stream=True)
#         return self._from_http_response(resp)
# 
#     def _handle_file_like_object(self, data):
#         content_length = data.seek(0, 2)
#         data.seek(0)
#         return self._generator(data, content_length)
# 
#     def _handle_zip_file(self, data):
#         return self._generator(data.zipfile, data.file_size)
# 
#     def _handle_file(self, data):
#         with open(data, 'rb') as f:
#             content_length = int(os.path.getsize(data))
#             for chunk in self._generator(f, content_length):
#                 yield chunk
# 
#     def _get_strategy(self, data):
#         if isinstance(data, ZipWrapper):
#             return self._handle_zip_file
#         if isinstance(data, requests.Request):
#             return self._handle_http_request
#         if isinstance(data, str):
#             parsed = urlparse(data)
#             is_url = parsed.netloc and parsed.scheme
#             return self._handle_simple_get if is_url else self._handle_file
#         return self._handle_file_like_object
# 
#     def _process(self, data):
#         try:
#             data = data.uri
#         except AttributeError:
#             pass
#         strategy = self._get_strategy(data)
#         for chunk in strategy(data):
#             yield chunk
# 
# class ZipWrapper(object):
#     def __init__(self, zipfile, zipinfo):
#         self.zipinfo = zipinfo
#         self.zipfile = zipfile
# 
#     def __enter__(self):
#         self.zipfile.__enter__()
#         return self
# 
#     def __exit__(self, exc_type, exc_val, exc_tb):
#         self.zipfile.__exit__(exc_type, exc_val, exc_tb)
# 
#     @property
#     def file_size(self):
#         return self.zipinfo.file_size
# 
#     @property
#     def filename(self):
#         return self.zipinfo.filename
# 
# def iter_zip(fn):
#     with zipfile.ZipFile(fn) as zf:
#         for info in zf.filelist:
#             if not info.file_size:
#                 continue
#             f = zf.open(info.filename)
#             yield ZipWrapper(f, info)
# 
# Path: featureflow/util.py
# def wait_for_http_server(host, port, wait_time_seconds=0.1, max_tries=100):
#     for _ in range(max_tries):
#         try:
#             resp = requests.get(f'http://{host}:{port}')
#             return resp
#         except requests.exceptions.ConnectionError:
#             time.sleep(wait_time_seconds)
#             continue
#     raise RuntimeError(f'Failed to start HTTP server at {host}:{port}')


# Path: featureflow/test_bytestream.py
from .bytestream import BytesWithTotalLength, ByteStream, ZipWrapper, iter_zip
from io import BytesIO
from collections import namedtuple
from uuid import uuid4
from .util import wait_for_http_server
import unittest2
import sys
import tempfile
import subprocess
import requests
import os
import zipfile

        self.process = subprocess.Popen(
            [sys.executable, server, self.port, self.expected],
            stdout=devnull,
            stderr=devnull)
        wait_for_http_server('localhost', '9876')

    def tearDown(self):
        self.process.kill()

    def results(self, inp):
        return b''.join(self.bytestream._process(inp))

    def local_url(self):
        return 'http://localhost:{port}'.format(**self.__dict__)

    def get_request(self):
        return requests.Request(
            method='GET',
            url=self.local_url())

    def test_can_accept_chunksize_implementing_int(self):
        class SemanticChunksize(object):
            THING_SIZE_BYTES = 64

            def __init__(self, n_things=10):
                self.n_things = n_things

            def __int__(self):
                return self.n_things * self.THING_SIZE_BYTES


        bs = ByteStream(