python - Envuelve un flujo abierto con io.TextIOWrapper

unit-testing python-3.x

Question

¿Cómo puedo envolver un flujo binario abierto - un file Python 2, un io.BufferedReader Python 3, un io.BytesIO - en un io.TextIOWrapper ?

Estoy tratando de escribir código que funcione sin cambios:

• Corriendo en Python 2.
• Corriendo en Python 3.
• Con flujos binarios generados desde la biblioteca estándar (es decir, no puedo controlar qué tipo son)
• Con flujos binarios hechos para ser dobles de prueba (es decir, no se manejan archivos, no se pueden volver a abrir).
• Producir un io.TextIOWrapper que envuelve la secuencia especificada.

El io.TextIOWrapper es necesario porque su API es esperada por otras partes de la biblioteca estándar. Existen otros tipos de archivo, pero no proporcionan la API correcta.

Ejemplo

Envolviendo el flujo binario presentado como el atributo subprocess.Popen.stdout :

import subprocess
import io
gnupg_subprocess = subprocess.Popen(
["gpg", "--version"], stdout=subprocess.PIPE)
gnupg_stdout = io.TextIOWrapper(gnupg_subprocess.stdout, encoding="utf-8")

En pruebas unitarias, la secuencia se reemplaza con una instancia de io.BytesIO para controlar su contenido sin tocar ningún subproceso o sistema de archivos.

gnupg_subprocess.stdout = io.BytesIO("Lorem ipsum".encode("utf-8"))

Eso funciona bien en las secuencias creadas por la biblioteca estándar de Python 3. Sin embargo, el mismo código falla en las secuencias generadas por Python 2:

[Python 2]
>>> type(gnupg_subprocess.stdout)
<type ''file''>
>>> gnupg_stdout = io.TextIOWrapper(gnupg_subprocess.stdout, encoding="utf-8")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
AttributeError: ''file'' object has no attribute ''readable''

No es una solución: Tratamiento especial para file

Una respuesta obvia es tener una rama en el código que compruebe si la secuencia en realidad es un objeto de file Python 2, y manejar eso de manera diferente de los objetos io.* .

Esa no es una opción para el código bien probado, porque hace que una rama que las pruebas de la unidad, que para ejecutar lo más rápido posible, no debe crear ningún objeto de sistema de archivos real , no puede ejercer.

Las pruebas unitarias proporcionarán pruebas de dobles, no objetos de file reales. Así que crear una rama que no se ejerza con esos dobles de prueba es derrotar al conjunto de pruebas.

No es una solución: io.open

Algunos encuestados sugieren volver a abrir (por ejemplo, con io.open ) el identificador de archivo subyacente:

gnupg_stdout = io.open(
gnupg_subprocess.stdout.fileno(), mode=''r'', encoding="utf-8")

Eso funciona tanto en Python 3 como en Python 2:

[Python 3]
>>> type(gnupg_subprocess.stdout)
<class ''_io.BufferedReader''>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode=''r'', encoding="utf-8")
>>> type(gnupg_stdout)
<class ''_io.TextIOWrapper''>

[Python 2]
>>> type(gnupg_subprocess.stdout)
<type ''file''>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode=''r'', encoding="utf-8")
>>> type(gnupg_stdout)
<type ''_io.TextIOWrapper''>

Pero, por supuesto, se basa en volver a abrir un archivo real desde su identificador de archivo. Entonces falla en las pruebas unitarias cuando el doble de prueba es una instancia de io.BytesIO :

>>> gnupg_subprocess.stdout = io.BytesIO("Lorem ipsum".encode("utf-8"))
>>> type(gnupg_subprocess.stdout)
<type ''_io.BytesIO''>
>>> gnupg_stdout = io.open(gnupg_subprocess.stdout.fileno(), mode=''r'', encoding="utf-8")
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
io.UnsupportedOperation: fileno

No es una solución: codecs.getreader

La biblioteca estándar también tiene el módulo de codecs , que proporciona características de envoltorio:

import codecs
gnupg_stdout = codecs.getreader("utf-8")(gnupg_subprocess.stdout)

Eso es bueno porque no intenta volver a abrir la secuencia. Pero no proporciona la API io.TextIOWrapper . Específicamente, no hereda io.IOBase y no tiene el atributo de encoding :

>>> type(gnupg_subprocess.stdout)
<type ''file''>
>>> gnupg_stdout = codecs.getreader("utf-8")(gnupg_subprocess.stdout)
>>> type(gnupg_stdout)
<type ''instance''>
>>> isinstance(gnupg_stdout, io.IOBase)
False
>>> gnupg_stdout.encoding
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/lib/python2.7/codecs.py", line 643, in __getattr__
return getattr(self.stream, name)
AttributeError: ''_io.BytesIO'' object has no attribute ''encoding''

Así que los codecs no proporcionan objetos que sustituyan a io.TextIOWrapper .

¿Qué hacer?

Entonces, ¿cómo puedo escribir código que funcione tanto para Python 2 como para Python 3, con el doble de prueba y los objetos reales, que envuelven un io.TextIOWrapper alrededor del flujo de bytes ya abierto ?

Answer 1

Aquí hay un código que he probado tanto en python 2.7 como en python 3.6.

La clave aquí es que primero debe usar detach () en su transmisión anterior. Esto no cierra el archivo subyacente, simplemente arranca el objeto de flujo sin procesar para que se pueda reutilizar. detach () devolverá un objeto que se puede envolver con TextIOWrapper.

Como ejemplo aquí, abro un archivo en modo de lectura binaria, lo leo así, luego cambio a un flujo de texto decodificado UTF-8 a través de io.TextIOWrapper.

Guardé este ejemplo como este archivo.py

import io
fileName = ''this-file.py''
fp = io.open(fileName,''rb'')
fp.seek(20)
someBytes = fp.read(10)
print(type(someBytes) + len(someBytes))
# now let''s do some wrapping to get a new text (non-binary) stream
pos = fp.tell() # we''re about to lose our position, so let''s save it
newStream = io.TextIOWrapper(fp.detach(),''utf-8'') # FYI -- fp is now unusable
newStream.seek(pos)
theRest = newStream.read()
print(type(theRest), len(theRest))

Esto es lo que obtengo cuando lo ejecuto con python2 y python3.

$ python2.7 this-file.py
(<type ''str''>, 10)
(<type ''unicode''>, 406)
$ python3.6 this-file.py
<class ''bytes''> 10
<class ''str''> 406

Obviamente, la sintaxis de impresión es diferente y, como era de esperar, los tipos de variables difieren entre las versiones de python, pero funcionan como deberían en ambos casos.

Answer 2

Basándome en múltiples sugerencias en diversos foros y experimentando con la biblioteca estándar para cumplir con los criterios, mi conclusión actual es que esto no se puede hacer con la biblioteca y los tipos que tenemos actualmente.

Answer 3

Bien, esto parece ser una solución completa, para todos los casos mencionados en la pregunta, probado con Python 2.7 y Python 3.5. La solución general terminó por volver a abrir el descriptor de archivo, pero en lugar de io.BytesIO, debe usar una tubería para su prueba doble para que tenga un descriptor de archivo.

import io
import subprocess
import os
# Example function, re-opens a file descriptor for UTF-8 decoding,
# reads until EOF and prints what is read.
def read_as_utf8(fileno):
fp = io.open(fileno, mode="r", encoding="utf-8", closefd=False)
print(fp.read())
fp.close()
# Subprocess
gpg = subprocess.Popen(["gpg", "--version"], stdout=subprocess.PIPE)
read_as_utf8(gpg.stdout.fileno())
# Normal file (contains "Lorem ipsum." as UTF-8 bytes)
normal_file = open("loremipsum.txt", "rb")
read_as_utf8(normal_file.fileno()) # prints "Lorem ipsum."
# Pipe (for test harness - write whatever you want into the pipe)
pipe_r, pipe_w = os.pipe()
os.write(pipe_w, "Lorem ipsum.".encode("utf-8"))
os.close(pipe_w)
read_as_utf8(pipe_r) # prints "Lorem ipsum."
os.close(pipe_r)

Answer 4

Resulta que solo necesitas envolver tu io.BytesIO en io.BufferedReader que existe tanto en Python 2 como en Python 3.

import io
reader = io.BufferedReader(io.BytesIO("Lorem ipsum".encode("utf-8")))
wrapper = io.TextIOWrapper(reader)
wrapper.read() # returns Lorem ipsum

Esta respuesta sugirió originalmente el uso de os.pipe, pero el lado de lectura de la tubería tendría que estar envuelto en io.BufferedReader en Python 2 para que funcione, por lo que esta solución es más simple y evita la asignación de una tubería.

Answer 5

También necesitaba esto, pero en base al hilo aquí, determiné que no era posible usar solo el módulo io Python 2. Si bien esto rompe la regla de "Tratamiento especial para el file ", la técnica que usé fue crear una envoltura extremadamente delgada para el file (código a continuación) que luego podría envolverse en un io.BufferedReader , que a su vez se puede pasar al io.TextIOWrapper constructor. Será difícil realizar la prueba de la unidad, ya que, obviamente, la nueva ruta del código no se puede probar en Python 3.

Por cierto, la razón por la que los resultados de un open() se pueden pasar directamente a io.TextIOWrapper en Python 3 es porque un modo binario open() realidad devuelve una instancia de io.BufferedReader para comenzar (al menos en Python 3.4, que es donde estaba probando en ese momento).

import io
import six # for six.PY2
if six.PY2:
class _ReadableWrapper(object):
def __init__(self, raw):
self._raw = raw
def readable(self):
return True
def writable(self):
return False
def seekable(self):
return True
def __getattr__(self, name):
return getattr(self._raw, name)
def wrap_text(stream, *args, **kwargs):
# Note: order important here, as ''file'' doesn''t exist in Python 3
if six.PY2 and isinstance(stream, file):
stream = io.BufferedReader(_ReadableWrapper(stream))
return io.TextIOWrapper(stream)

Al menos esto es pequeño, por lo que esperamos que minimice la exposición de las partes que no se pueden probar fácilmente por unidad.

Answer 6

Use codecs.getreader para producir un objeto contenedor:

text_stream = codecs.getreader("utf-8")(bytes_stream)

Funciona en Python 2 y Python 3.