Merge pull request #657 from MDAnalysis/feature-NamedStream-hacks

NamedStream cleanup

package/MDAnalysis/lib/util.py

@@ -472,8 +472,11 @@ class NamedStream(io.IOBase):
     The class can be used as a context manager.
 
     :class:`NamedStream` is derived from :class:`io.IOBase` (to indicate that
-    it is a stream); some operations that normally expect a string will also
-    work with a :class:`NamedStream`.
+    it is a stream). Many operations that normally expect a string will also
+    work with a :class:`NamedStream`; for instance, most of the functions in
+    :mod:`os.path` will work with the exception of :func:`os.path.expandvars`
+    and :func:`os.path.expanduser`, which will return the :class:`NamedStream`
+    itself instead of a string if no substitutions were made.
 
     .. rubric:: Example
 
@@ -527,22 +530,22 @@ def __init__(self, stream, filename, reset=True, close=False):
            :meth:`close` (see there) unless the *close* keyword is set to
            ``True``.
 
-        :Arguments:
-
-           *stream*
-               open stream (e.g. :class:`file` or :func:`cStringIO.StringIO`)
-           *filename*
+        Arguments
+        ---------
+        stream : stream
+               an open stream (e.g. :class:`file` or :func:`cStringIO.StringIO`)
+        filename : str
                the filename that should be associated with the stream
 
-        :Keywords:
-
-           *reset*
+        Keywords
+        --------
+        reset : boolean, default ``True``
                start the stream from the beginning (either :meth:`reset` or :meth:`seek`)
-               when the class instance is constructed [``True``]
-           *close*
+               when the class instance is constructed
+        close : booelan, default ``True``
                close the stream when a :keyword:`with` block exits or when
                :meth:`close` is called; note that the default is **not to close
-               the stream** [``False``]
+               the stream**
 
         .. versionadded:: 0.9.0
         """
@@ -576,7 +579,7 @@ def __iter__(self):
         return iter(self.stream)
 
     def __enter__(self):
-        # do not call the stream __enter__ because the stream is already open
+        # do not call the stream's __enter__ because the stream is already open
         return self
 
     def __exit__(self, *args):
@@ -730,6 +733,17 @@ def __lt__(self, x):
     def __len__(self):
         return len(self.name)
 
+    def __add__(self, x):
+        return self.name + x
+
+    def __radd__(self, x):
+        return x + self.name
+
+    def __mul__(self, x):
+        return self.name * x
+
+    __rmul__ = __mul__
+
     def __format__(self, format_spec):
         return self.name.format(format_spec)
 

testsuite/MDAnalysisTests/test_streamio.py

@@ -16,7 +16,10 @@
 from six.moves import range
 
 import numpy as np
-from numpy.testing import *
+from numpy.testing import (TestCase, dec,
+                           assert_equal, assert_almost_equal,
+                           assert_array_almost_equal,
+                           )
 
 import MDAnalysis
 import MDAnalysis.lib.util as util
@@ -26,9 +29,10 @@
 
 import StringIO
 import cStringIO
-import tempfile
 import os
 
+import tempdir
+
 
 class TestIsstream(TestCase):
     def test_hasmethod(self):
@@ -86,14 +90,14 @@ def setUp(self):
         self.textname = "jabberwock.txt"
         self.numtextlines = len(self.text)
 
-    def testClosing(self):
+    def test_closing(self):
         obj = cStringIO.StringIO("".join(self.text))
         ns = util.NamedStream(obj, self.textname, close=True)
         assert_equal(ns.closed, False)
         ns.close()
         assert_equal(ns.closed, True)
 
-    def testClosingForce(self):
+    def test_closing_force(self):
         obj = cStringIO.StringIO("".join(self.text))
         ns = util.NamedStream(obj, self.textname)
         assert_equal(ns.closed, False)
@@ -102,7 +106,7 @@ def testClosingForce(self):
         ns.close(force=True)
         assert_equal(ns.closed, True)
 
-    def testcStringIO_read(self):
+    def test_cStringIO_read(self):
         obj = cStringIO.StringIO("".join(self.text))
         ns = util.NamedStream(obj, self.textname)
         assert_equal(ns.name, self.textname)
@@ -112,7 +116,7 @@ def testcStringIO_read(self):
         assert_equal(len(ns.readlines()), self.numtextlines)
         ns.close(force=True)
 
-    def testFile_read(self):
+    def test_File_read(self):
         obj = open(self.filename, 'r')
         ns = util.NamedStream(obj, self.filename)
         assert_equal(ns.name, self.filename)
@@ -122,7 +126,7 @@ def testFile_read(self):
         assert_equal(len(ns.readlines()), self.numlines)
         ns.close(force=True)
 
-    def testcStringIO_write(self):
+    def test_cStringIO_write(self):
         obj = cStringIO.StringIO()
         ns = util.NamedStream(obj, self.textname)
         ns.writelines(self.text)
@@ -134,27 +138,114 @@ def testcStringIO_write(self):
         assert_equal(ns.read(20), "".join(self.text)[:20])
         ns.close(force=True)
 
-    def testFile_write(self):
-        fd, outfile = tempfile.mkstemp(suffix=".txt")
-        os.close(fd)
-        try:
-            obj = open(outfile, "w")
-            ns = util.NamedStream(obj, outfile, close=True)
-            ns.writelines(self.text)
-            ns.close()
-            text = open(outfile).readlines()
-
-            assert_equal(ns.name, outfile)
-            assert_equal(str(ns), outfile)
-            assert_equal(len(text), len(self.text))
-            assert_equal("".join(text), "".join(self.text))
-        finally:
-            ns.close()
-            obj.close()
+    def test_File_write(self):
+        with tempdir.in_tempdir():
+            outfile = "lookingglas.txt"
             try:
-                os.unlink(outfile)
-            except OSError:
-                pass
+                obj = open(outfile, "w")
+                ns = util.NamedStream(obj, outfile, close=True)
+                ns.writelines(self.text)
+                ns.close()
+                text = open(outfile).readlines()
+
+                assert_equal(ns.name, outfile)
+                assert_equal(str(ns), outfile)
+                assert_equal(len(text), len(self.text))
+                assert_equal("".join(text), "".join(self.text))
+            finally:
+                ns.close()
+                obj.close()
+
+
+class TestNamedStream_filename_behavior(object):
+    textname = "~/stories/jabberwock.txt"  # with tilde ~ to test regular expanduser()
+    # note: no setUp() because classes with generators would run it
+    #       *for each generated test* and we need it for the generator method
+
+    def create_NamedStream(self, name=None):
+        if name is None:
+            name = self.textname
+        obj = cStringIO.StringIO()
+        return util.NamedStream(obj, name)
+
+    def test_ospath_funcs(self):
+        ns = self.create_NamedStream()
+
+        # - "expandvars" gave Segmentation fault (OS X 10.6, Python 2.7.11 -- orbeckst)
+        # - "expanduser" will either return a string if it carried out interpolation
+        #   or "will do nothing" and return the NamedStream (see extra test below).
+        #   On systems without a user or HOME, it will also do nothing and the test
+        #   below will fail.
+        funcs = ("abspath", "basename", "dirname", "expanduser",
+                 "normpath", "relpath", "split", "splitext")
+        def _test_func(funcname, fn=self.textname, ns=ns):
+            func = getattr(os.path, funcname)
+            reference = func(fn)
+            value = func(ns)
+            assert_equal(value, reference,
+                         err_msg=("os.path.{0}() does not work with "
+                                  "NamedStream").format(funcname))
+        # join not included because of different call signature
+        # but added first argument for the sake of it showing up in the verbose
+        # nose output
+        def _test_join(funcname="join", fn=self.textname, ns=ns, path="/tmp/MDAnalysisTests"):
+            reference = os.path.join(path, fn)
+            value = os.path.join(path, ns)
+            assert_equal(value, reference,
+                         err_msg=("os.path.{0}() does not work with "
+                                  "NamedStream").format(funcname))
+        for func in funcs:
+            yield _test_func, func
+        yield _test_join, "join"
+
+    # Segmentation fault when run as a test on Mac OS X 10.6, Py 2.7.11 [orbeckst]
+    @dec.skipif(True)
+    def test_expanduser_noexpansion_returns_NamedStream(self):
+        ns = self.create_NamedStream("de/zipferlack.txt")  # no tilde ~ in name!
+        reference = ns
+        value = os.path.expanduser(ns)
+        assert_equal(value, reference,
+                     err_msg=("os.path.expanduser() without '~' did not "
+                              "return NamedStream --- weird!!"))
+
+    # expandvars(NamedStream) does not work interactively, so it is a knownfailure
+    # Segmentation fault when run as a test on Mac OS X 10.6, Py 2.7.11 [orbeckst]
+    @dec.skipif(True)
+    @dec.skipif("HOME" not in os.environ)
+    @knownfailure()
+    def test_expandvars(self):
+        name = "${HOME}/stories/jabberwock.txt"
+        ns = self.create_NamedStream(name)
+        reference = os.path.expandvars(name)
+        value = os.path.expandvars(ns)
+        assert_equal(value, reference,
+                     err_msg="os.path.expandvars() did not expand HOME")
+
+    # Segmentation fault when run as a test on Mac OS X 10.6, Py 2.7.11 [orbeckst]
+    @dec.skipif(True)
+    def test_expandvars_noexpansion_returns_NamedStream(self):
+        ns = self.create_NamedStream() # no $VAR constructs
+        reference = ns
+        value = os.path.expandvars(ns)
+        assert_equal(value, reference,
+                     err_msg=("os.path.expandvars() without '$VARS' did not "
+                              "return NamedStream --- weird!!"))
+
+    def test_add(self):
+        ns = self.create_NamedStream()
+        try:
+            assert_equal(ns + "foo", self.textname + "foo")
+        except TypeError:
+            raise AssertionError("NamedStream does not support  "
+                                 "string concatenation, NamedStream + str")
+
+    def test_radd(self):
+        ns = self.create_NamedStream()
+        try:
+            assert_equal("foo" + ns, "foo" + self.textname)
+        except TypeError:
+            raise AssertionError("NamedStream does not support right "
+                                 "string concatenation, str + NamedStream")
 
 
 class _StreamData(object):