[py3k] compat: add str_to_bytes and str_encode

In py3k, things are different: we will sometimes need to convert str to bytes
if we want to write data into a file;
and on other times we don't need to encode str to something

4 files changed, 21 insertions, 10 deletions

diff --git a/compat.py b/compat.py
index 22f9fa2..cc55532 100644
--- a/compat.py
+++ b/compat.py
@@ -30,6 +30,20 @@ from warnings import warn
 
 import __builtin__ as builtins # 2to3 will tranform '__builtin__' to 'builtins'
 
+if sys.version_info < (3, 0):
+    str_to_bytes = str
+    def str_encode(string, encoding):
+        if isinstance(string, unicode):
+            return string.encode(encoding)
+        return str(string)
+else:
+    def str_to_bytes(string):
+        return str.encode(string)
+    # we have to ignore the encoding in py3k to be able to write a string into a
+    # TextIOWrapper or like object (which expect an unicode string)
+    def str_encode(string, encoding):
+        return str(string)
+
 try:
     callable = callable
 except NameError:# callable removed from py3k
diff --git a/configuration.py b/configuration.py
index fefe519..4357fcf 100644
--- a/configuration.py
+++ b/configuration.py
@@ -114,6 +114,7 @@ from ConfigParser import ConfigParser, NoOptionError, NoSectionError, \
 from warnings import warn
 
 from logilab.common.compat import set, reversed, callable, raw_input
+from logilab.common.compat import str_encode as _encode
 
 from logilab.common.textutils import normalize_text, unquote
 from logilab.common.deprecation import deprecated
@@ -134,11 +135,6 @@ def _get_encoding(encoding, stream):
         encoding = locale.getpreferredencoding()
     return encoding
 
-def _encode(string, encoding):
-    if isinstance(string, unicode):
-        return string.encode(encoding)
-    return str(string)
-
 
 # validation functions ########################################################
 
diff --git a/shellutils.py b/shellutils.py
index 2f6e85a..7d971e1 100644
--- a/shellutils.py
+++ b/shellutils.py
@@ -33,6 +33,7 @@ from os.path import exists, isdir, islink, basename, join
 
 from logilab.common import STD_BLACKLIST, _handle_blacklist
 from logilab.common.compat import raw_input
+from logilab.common.compat import str_to_bytes
 
 try:
     from logilab.common.proc import ProcInfo, NoSuchProcess
@@ -247,7 +248,7 @@ def acquire_lock(lock_file, max_try=10, delay=10, max_delay=3600):
     while count:
         try:
             fd = os.open(lock_file, os.O_EXCL | os.O_RDWR | os.O_CREAT)
-            os.write(fd, str(os.getpid()))
+            os.write(fd, str_to_bytes(str(os.getpid())) )
             os.close(fd)
             return True
         except OSError, e:
diff --git a/test/unittest_shellutils.py b/test/unittest_shellutils.py
index 2d0507c..d748205 100644
--- a/test/unittest_shellutils.py
+++ b/test/unittest_shellutils.py
@@ -20,15 +20,15 @@
 import sys, os, tempfile, shutil
 from os.path import join
 import datetime, time
+from StringIO import StringIO
 
 from logilab.common.testlib import TestCase, unittest_main
 
 from logilab.common.shellutils import (globfind, find, ProgressBar,
                                        acquire_lock, release_lock,
                                        RawInput, confirm)
-
+from logilab.common.compat import str_to_bytes
 from logilab.common.proc import NoSuchProcess
-from StringIO import StringIO
 
 DATA_DIR = join('data','find_test')
 
@@ -165,14 +165,14 @@ class AcquireLockTC(TestCase):
 
     def test_wrong_process(self):
         fd = os.open(self.lock, os.O_EXCL | os.O_RDWR | os.O_CREAT)
-        os.write(fd, '1111111111')
+        os.write(fd, str_to_bytes('1111111111'))
         os.close(fd)
         self.assertTrue(os.path.exists(self.lock))
         self.assertRaises(Exception, acquire_lock, self.lock, 1, 1)
 
     def test_wrong_process_and_continue(self):
         fd = os.open(self.lock, os.O_EXCL | os.O_RDWR | os.O_CREAT)
-        os.write(fd, '1111111111')
+        os.write(fd, str_to_bytes('1111111111'))
         os.close(fd)
         self.assertTrue(os.path.exists(self.lock))
         self.assertTrue(acquire_lock(self.lock))