return _impls[dialect.name]
def static_output(self, text):
- self.output_buffer.write(text + "\n\n")
+ text_ = getattr(builtins, 'unicode', str)(text + '\n\n')
+ self.output_buffer.write(text_)
+ if callable(getattr(self.output_buffer, 'flush', None)):
+ self.output_buffer.flush()
@property
def bind(self):
-import codecs
+import io
import logging
import sys
self._migrations_fn = opts.get('fn')
self.as_sql = as_sql
self.output_buffer = opts.get("output_buffer", sys.stdout)
- if opts.get('output_encoding'):
- self.output_buffer = codecs.getwriter(
+ if (opts.get('output_encoding') and
+ not isinstance(self.output_buffer, io.TextIOBase)):
+ self.output_buffer = io.TextIOWrapper(
+ self.output_buffer,
opts['output_encoding']
- )(self.output_buffer)
+ )
self._user_compare_type = opts.get('compare_type', False)
self._user_compare_server_default = opts.get(
import os
import re
import shutil
-import StringIO
import textwrap
from nose import SkipTest
)
def capture_context_buffer(**kw):
- buf = StringIO.StringIO()
+ if kw.pop('bytes_io', False):
+ raw = io.BytesIO()
+ encoding = kw.get('output_encoding', 'utf-8')
+ buf = io.TextIOWrapper(raw, encoding)
+ else:
+ raw = buf = io.StringIO()
class capture(object):
def __enter__(self):
'output_buffer':buf
}
EnvironmentContext._default_opts.update(kw)
- return buf
+ return raw
def __exit__(self, *arg, **kwarg):
#print(buf.getvalue())
clear_staging_env()
def test_encode(self):
- with capture_context_buffer(output_encoding='utf-8') as buf:
+ with capture_context_buffer(
+ bytes_io=True,
+ output_encoding='utf-8'
+ ) as buf:
command.upgrade(cfg, a, sql=True)
assert "« S’il vous plaît…".encode("utf-8") in buf.getvalue()