Dependencies:
- libuuid (sometimes part of e2fsprogs)
- sqlite3
- - Python (2.7 or later, or 3.2 or later)
- - Python six, a Python 2/3 compatibility library
- https://pypi.python.org/pypi/six
+ - Python 3.2 or later
- boto, the python interface to Amazon's Web Services (for S3 storage)
http://code.google.com/p/boto
- paramiko, SSH2 protocol for python (for sftp storage)
import os
import posixpath
import re
-import six
import sqlite3
import subprocess
import sys
store may either be a Store object or URL.
"""
- if isinstance(backend, six.string_types):
+ if isinstance(backend, str):
self._backend = cumulus.store.open(backend)
else:
self._backend = backend
from __future__ import division, print_function, unicode_literals
import re
-import six
# The encoding assumed when interpreting path names.
ENCODING="utf-8"
-# In both Python 2 and Python 3 pathnames are represented using the str type.
-# For Python 2, this means that the converting from a bytestring to a pathname
-# is a no-op. For Python 3, the conversion assumes a utf-8 encoding, but the
-# surrogateescape encoding error handler is used to allow other byte sequences
-# to be passed through.
-if six.PY2:
- def bytes_to_pathname(b): return b
- def pathname_to_bytes(p):
- if isinstance(p, unicode):
- return p.encode(encoding=ENCODING, errors="replace")
- else:
- return p
-elif six.PY3:
- def bytes_to_pathname(b):
- """Decodes a byte string to a pathname.
+# Pathnames are represented using the str type. The conversion assumes a utf-8
+# encoding, but the surrogateescape encoding error handler is used to allow
+# other byte sequences to be passed through.
+def bytes_to_pathname(b):
+ """Decodes a byte string to a pathname.
- The input is assumed to be encoded using ENCODING (defaults to
- utf-8)."""
- return b.decode(encoding=ENCODING, errors="surrogateescape")
+ The input is assumed to be encoded using ENCODING (defaults to
+ utf-8)."""
+ return b.decode(encoding=ENCODING, errors="surrogateescape")
- def pathname_to_bytes(p):
- """Converts a pathname to encoded bytes.
+def pathname_to_bytes(p):
+ """Converts a pathname to encoded bytes.
- The input is encoded to ENCODING (defaults to utf-8)."""
- return p.encode(encoding=ENCODING, errors="surrogateescape")
-else:
- raise AssertionError("Unsupported Python version")
+ The input is encoded to ENCODING (defaults to utf-8)."""
+ return p.encode(encoding=ENCODING, errors="surrogateescape")
def uri_decode_raw(s):
"""Decode a URI-encoded (%xx escapes) string.
The input should be a string, preferably only using ASCII characters. The
output will be of type bytes."""
- def hex_decode(m): return six.int2byte(int(m.group(1), 16))
+ def hex_decode(m): return bytes((int(m.group(1), 16),))
return re.sub(br"%([0-9a-fA-F]{2})", hex_decode, pathname_to_bytes(s))
def uri_encode_raw(s):
else:
return "%%%02x" % c
- return "".join(hex_encode(c) for c in six.iterbytes(s))
+ return "".join(hex_encode(c) for c in s)
def uri_decode_pathname(s):
"""Decodes a URI-encoded string to a pathname."""
from __future__ import division, print_function, unicode_literals
-import six
import unittest
from cumulus import util
class UtilCodecs(unittest.TestCase):
def test_pathnames(self):
self.assertEqual(util.ENCODING, "utf-8")
- if six.PY2:
- self.assertEqual(util.bytes_to_pathname(b"ext\xc3\xa9nsion"),
- b"ext\xc3\xa9nsion")
- self.assertEqual(util.pathname_to_bytes(b"ext\xc3\xa9nsion"),
- b"ext\xc3\xa9nsion")
- self.assertEqual(util.pathname_to_bytes(u"exténsion"),
- b"ext\xc3\xa9nsion")
- elif six.PY3:
- self.assertEqual(util.bytes_to_pathname(b"ext\xc3\xa9nsion"),
- "exténsion")
- self.assertEqual(util.pathname_to_bytes("exténsion"),
- b"ext\xc3\xa9nsion")
- self.assertEqual(util.bytes_to_pathname(b"inv\xe1lid"),
- "inv\udce1lid")
- self.assertEqual(util.pathname_to_bytes("inv\udce1lid"),
- b"inv\xe1lid")
+ self.assertEqual(util.bytes_to_pathname(b"ext\xc3\xa9nsion"),
+ "exténsion")
+ self.assertEqual(util.pathname_to_bytes("exténsion"),
+ b"ext\xc3\xa9nsion")
+ self.assertEqual(util.bytes_to_pathname(b"inv\xe1lid"),
+ "inv\udce1lid")
+ self.assertEqual(util.pathname_to_bytes("inv\udce1lid"),
+ b"inv\xe1lid")
def test_uri_encode_raw(self):
self.assertEqual(util.uri_encode_raw(b"sample ASCII"), "sample%20ASCII")
b"sample ext\xc3\xa9nded")
def test_uri_decode_pathname(self):
- if six.PY2:
- self.assertEqual(util.uri_decode_pathname("sample%20ext%c3%a9nded"),
- b"sample ext\xc3\xa9nded")
- self.assertEqual(util.uri_decode_pathname("sample%20exténded"),
- b"sample ext\xc3\xa9nded")
- # In Python 2, non-UTF-8 sequences are just passed through as
- # bytestrings.
- self.assertEqual(util.uri_decode_pathname(b"inv%e1lid"),
- b"inv\xe1lid")
- self.assertEqual(util.uri_decode_pathname(b"inv\xe1lid"),
- b"inv\xe1lid")
- elif six.PY3:
- self.assertEqual(util.uri_decode_pathname("sample%20ext%c3%a9nded"),
- "sample exténded")
- self.assertEqual(util.uri_decode_pathname("sample%20exténded"),
- "sample exténded")
- # In Python 3, non-UTF-8 sequences are represented using surrogate
- # escapes to allow lossless conversion back to the appropriate
- # bytestring.
- self.assertEqual(util.uri_decode_pathname("inv%e1lid"),
- "inv\udce1lid")
- self.assertEqual(
- util.pathname_to_bytes(util.uri_decode_pathname("inv%e1lid")),
- b"inv\xe1lid")
+ self.assertEqual(util.uri_decode_pathname("sample%20ext%c3%a9nded"),
+ "sample exténded")
+ self.assertEqual(util.uri_decode_pathname("sample%20exténded"),
+ "sample exténded")
+ # In Python 3, non-UTF-8 sequences are represented using surrogate
+ # escapes to allow lossless conversion back to the appropriate
+ # bytestring.
+ self.assertEqual(util.uri_decode_pathname("inv%e1lid"),
+ "inv\udce1lid")
+ self.assertEqual(
+ util.pathname_to_bytes(util.uri_decode_pathname("inv%e1lid")),
+ b"inv\xe1lid")
if __name__ == "__main__":
projects / cumulus.git / commitdiff