from __future__ import with_statement

from nose.tools import eq_

from fabric.io import OutputLooper
from fabric.context_managers import settings


def test_request_prompts():
    """
    Test valid responses from prompts
    """
    def run(txt, prompts):
        with settings(prompts=prompts):
            # try to fulfil the OutputLooper interface, only want to test
            # _get_prompt_response. (str has a method upper)
            ol = OutputLooper(str, 'upper', None, list(txt), None)
            return ol._get_prompt_response()

    prompts = {"prompt2": "response2",
               "prompt1": "response1",
               "prompt": "response"
               }

    eq_(run("this is a prompt for prompt1", prompts), ("prompt1", "response1"))
    eq_(run("this is a prompt for prompt2", prompts), ("prompt2", "response2"))
    eq_(run("this is a prompt for promptx:", prompts), (None, None))
    eq_(run("prompt for promp", prompts), (None, None))

from __future__ import with_statement

import sys
from io import BytesIO

from nose.tools import eq_

from fabric.io import OutputLooper
from fabric.context_managers import hide, settings
from utils import mock_streams


def test_request_prompts():
    """
    Test valid responses from prompts
    """
    def run(txt, prompts):
        with settings(prompts=prompts):
            # try to fulfil the OutputLooper interface, only want to test
            # _get_prompt_response. (str has a method upper)
            ol = OutputLooper(str, 'upper', None, list(txt), None)
            return ol._get_prompt_response()

    prompts = {"prompt2": "response2",
               "prompt1": "response1",
               "prompt": "response"
               }

    eq_(run("this is a prompt for prompt1", prompts), ("prompt1", "response1"))
    eq_(run("this is a prompt for prompt2", prompts), ("prompt2", "response2"))
    eq_(run("this is a prompt for promptx:", prompts), (None, None))
    eq_(run("prompt for promp", prompts), (None, None))


@mock_streams('stdout')
def test_pip_progressbar_at_4096_byte_boundary_error():
    """
    Test for unicode characters from the pip installation progress bar
    causing a UnicodeDecodeError.
    """
    expect = '█' * 4096

    class Mock(object):
        def __init__(self):
            three_bytes = u'█'.encode('utf-8')
            # 4096 comes from OutputLooper.read_size being hard-coded to 4096
            self.source = BytesIO(three_bytes * 4096)

        def get_unicode_bytes(self, size):
            return self.source.read(size)

    ol = OutputLooper(Mock(), 'get_unicode_bytes', sys.stdout, None, None)

    with settings(hide('everything')):
        ol.loop()
    eq_(expect, sys.stdout.getvalue())

Add test which detects the UnicodeDecodeError from pip
Add test which detects the UnicodeDecodeError from pip

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe2 in position 4095: unexpected end of data
https://github.com/mathiasertl/fabric/issues/5

  from __future__ import with_statement
+ 
+ import sys
+ from io import BytesIO
  
  from nose.tools import eq_
  
  from fabric.io import OutputLooper
- from fabric.context_managers import settings
+ from fabric.context_managers import hide, settings
+ from utils import mock_streams
  
  
  def test_request_prompts():
      """
      Test valid responses from prompts
      """
      def run(txt, prompts):
          with settings(prompts=prompts):
              # try to fulfil the OutputLooper interface, only want to test
              # _get_prompt_response. (str has a method upper)
              ol = OutputLooper(str, 'upper', None, list(txt), None)
              return ol._get_prompt_response()
  
      prompts = {"prompt2": "response2",
                 "prompt1": "response1",
                 "prompt": "response"
                 }
  
      eq_(run("this is a prompt for prompt1", prompts), ("prompt1", "response1"))
      eq_(run("this is a prompt for prompt2", prompts), ("prompt2", "response2"))
      eq_(run("this is a prompt for promptx:", prompts), (None, None))
      eq_(run("prompt for promp", prompts), (None, None))
  
+ 
+ @mock_streams('stdout')
+ def test_pip_progressbar_at_4096_byte_boundary_error():
+     """
+     Test for unicode characters from the pip installation progress bar
+     causing a UnicodeDecodeError.
+     """
+     expect = '█' * 4096
+ 
+     class Mock(object):
+         def __init__(self):
+             three_bytes = u'█'.encode('utf-8')
+             # 4096 comes from OutputLooper.read_size being hard-coded to 4096
+             self.source = BytesIO(three_bytes * 4096)
+ 
+         def get_unicode_bytes(self, size):
+             return self.source.read(size)
+ 
+     ol = OutputLooper(Mock(), 'get_unicode_bytes', sys.stdout, None, None)
+ 
+     with settings(hide('everything')):
+         ol.loop()
+     eq_(expect, sys.stdout.getvalue())
+ 
Add test which detects the UnicodeDecodeError from pip
## Code Before:
from __future__ import with_statement

from nose.tools import eq_

from fabric.io import OutputLooper
from fabric.context_managers import settings


def test_request_prompts():
    """
    Test valid responses from prompts
    """
    def run(txt, prompts):
        with settings(prompts=prompts):
            # try to fulfil the OutputLooper interface, only want to test
            # _get_prompt_response. (str has a method upper)
            ol = OutputLooper(str, 'upper', None, list(txt), None)
            return ol._get_prompt_response()

    prompts = {"prompt2": "response2",
               "prompt1": "response1",
               "prompt": "response"
               }

    eq_(run("this is a prompt for prompt1", prompts), ("prompt1", "response1"))
    eq_(run("this is a prompt for prompt2", prompts), ("prompt2", "response2"))
    eq_(run("this is a prompt for promptx:", prompts), (None, None))
    eq_(run("prompt for promp", prompts), (None, None))

## Instruction:
Add test which detects the UnicodeDecodeError from pip
## Code After:
from __future__ import with_statement

import sys
from io import BytesIO

from nose.tools import eq_

from fabric.io import OutputLooper
from fabric.context_managers import hide, settings
from utils import mock_streams


def test_request_prompts():
    """
    Test valid responses from prompts
    """
    def run(txt, prompts):
        with settings(prompts=prompts):
            # try to fulfil the OutputLooper interface, only want to test
            # _get_prompt_response. (str has a method upper)
            ol = OutputLooper(str, 'upper', None, list(txt), None)
            return ol._get_prompt_response()

    prompts = {"prompt2": "response2",
               "prompt1": "response1",
               "prompt": "response"
               }

    eq_(run("this is a prompt for prompt1", prompts), ("prompt1", "response1"))
    eq_(run("this is a prompt for prompt2", prompts), ("prompt2", "response2"))
    eq_(run("this is a prompt for promptx:", prompts), (None, None))
    eq_(run("prompt for promp", prompts), (None, None))


@mock_streams('stdout')
def test_pip_progressbar_at_4096_byte_boundary_error():
    """
    Test for unicode characters from the pip installation progress bar
    causing a UnicodeDecodeError.
    """
    expect = '█' * 4096

    class Mock(object):
        def __init__(self):
            three_bytes = u'█'.encode('utf-8')
            # 4096 comes from OutputLooper.read_size being hard-coded to 4096
            self.source = BytesIO(three_bytes * 4096)

        def get_unicode_bytes(self, size):
            return self.source.read(size)

    ol = OutputLooper(Mock(), 'get_unicode_bytes', sys.stdout, None, None)

    with settings(hide('everything')):
        ol.loop()
    eq_(expect, sys.stdout.getvalue())
