+++ /dev/null
-import pipes\r
-import os\r
-import string\r
-import unittest\r
-from test.test_support import TESTFN, run_unittest, unlink, reap_children\r
-\r
-if os.name != 'posix':\r
- raise unittest.SkipTest('pipes module only works on posix')\r
-\r
-TESTFN2 = TESTFN + "2"\r
-\r
-# tr a-z A-Z is not portable, so make the ranges explicit\r
-s_command = 'tr %s %s' % (string.ascii_lowercase, string.ascii_uppercase)\r
-\r
-class SimplePipeTests(unittest.TestCase):\r
- def tearDown(self):\r
- for f in (TESTFN, TESTFN2):\r
- unlink(f)\r
-\r
- def testSimplePipe1(self):\r
- t = pipes.Template()\r
- t.append(s_command, pipes.STDIN_STDOUT)\r
- f = t.open(TESTFN, 'w')\r
- f.write('hello world #1')\r
- f.close()\r
- with open(TESTFN) as f:\r
- self.assertEqual(f.read(), 'HELLO WORLD #1')\r
-\r
- def testSimplePipe2(self):\r
- with open(TESTFN, 'w') as f:\r
- f.write('hello world #2')\r
- t = pipes.Template()\r
- t.append(s_command + ' < $IN > $OUT', pipes.FILEIN_FILEOUT)\r
- t.copy(TESTFN, TESTFN2)\r
- with open(TESTFN2) as f:\r
- self.assertEqual(f.read(), 'HELLO WORLD #2')\r
-\r
- def testSimplePipe3(self):\r
- with open(TESTFN, 'w') as f:\r
- f.write('hello world #2')\r
- t = pipes.Template()\r
- t.append(s_command + ' < $IN', pipes.FILEIN_STDOUT)\r
- with t.open(TESTFN, 'r') as f:\r
- self.assertEqual(f.read(), 'HELLO WORLD #2')\r
-\r
- def testEmptyPipeline1(self):\r
- # copy through empty pipe\r
- d = 'empty pipeline test COPY'\r
- with open(TESTFN, 'w') as f:\r
- f.write(d)\r
- with open(TESTFN2, 'w') as f:\r
- f.write('')\r
- t=pipes.Template()\r
- t.copy(TESTFN, TESTFN2)\r
- with open(TESTFN2) as f:\r
- self.assertEqual(f.read(), d)\r
-\r
- def testEmptyPipeline2(self):\r
- # read through empty pipe\r
- d = 'empty pipeline test READ'\r
- with open(TESTFN, 'w') as f:\r
- f.write(d)\r
- t=pipes.Template()\r
- with t.open(TESTFN, 'r') as f:\r
- self.assertEqual(f.read(), d)\r
-\r
- def testEmptyPipeline3(self):\r
- # write through empty pipe\r
- d = 'empty pipeline test WRITE'\r
- t = pipes.Template()\r
- with t.open(TESTFN, 'w') as f:\r
- f.write(d)\r
- with open(TESTFN) as f:\r
- self.assertEqual(f.read(), d)\r
-\r
- def testQuoting(self):\r
- safeunquoted = string.ascii_letters + string.digits + '@%_-+=:,./'\r
- unsafe = '"`$\\!'\r
-\r
- self.assertEqual(pipes.quote(''), "''")\r
- self.assertEqual(pipes.quote(safeunquoted), safeunquoted)\r
- self.assertEqual(pipes.quote('test file name'), "'test file name'")\r
- for u in unsafe:\r
- self.assertEqual(pipes.quote('test%sname' % u),\r
- "'test%sname'" % u)\r
- for u in unsafe:\r
- self.assertEqual(pipes.quote("test%s'name'" % u),\r
- "'test%s'\"'\"'name'\"'\"''" % u)\r
-\r
- def testRepr(self):\r
- t = pipes.Template()\r
- self.assertEqual(repr(t), "<Template instance, steps=[]>")\r
- t.append('tr a-z A-Z', pipes.STDIN_STDOUT)\r
- self.assertEqual(repr(t),\r
- "<Template instance, steps=[('tr a-z A-Z', '--')]>")\r
-\r
- def testSetDebug(self):\r
- t = pipes.Template()\r
- t.debug(False)\r
- self.assertEqual(t.debugging, False)\r
- t.debug(True)\r
- self.assertEqual(t.debugging, True)\r
-\r
- def testReadOpenSink(self):\r
- # check calling open('r') on a pipe ending with\r
- # a sink raises ValueError\r
- t = pipes.Template()\r
- t.append('boguscmd', pipes.SINK)\r
- self.assertRaises(ValueError, t.open, 'bogusfile', 'r')\r
-\r
- def testWriteOpenSource(self):\r
- # check calling open('w') on a pipe ending with\r
- # a source raises ValueError\r
- t = pipes.Template()\r
- t.prepend('boguscmd', pipes.SOURCE)\r
- self.assertRaises(ValueError, t.open, 'bogusfile', 'w')\r
-\r
- def testBadAppendOptions(self):\r
- t = pipes.Template()\r
-\r
- # try a non-string command\r
- self.assertRaises(TypeError, t.append, 7, pipes.STDIN_STDOUT)\r
-\r
- # try a type that isn't recognized\r
- self.assertRaises(ValueError, t.append, 'boguscmd', 'xx')\r
-\r
- # shouldn't be able to append a source\r
- self.assertRaises(ValueError, t.append, 'boguscmd', pipes.SOURCE)\r
-\r
- # check appending two sinks\r
- t = pipes.Template()\r
- t.append('boguscmd', pipes.SINK)\r
- self.assertRaises(ValueError, t.append, 'boguscmd', pipes.SINK)\r
-\r
- # command needing file input but with no $IN\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.append, 'boguscmd $OUT',\r
- pipes.FILEIN_FILEOUT)\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.append, 'boguscmd',\r
- pipes.FILEIN_STDOUT)\r
-\r
- # command needing file output but with no $OUT\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.append, 'boguscmd $IN',\r
- pipes.FILEIN_FILEOUT)\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.append, 'boguscmd',\r
- pipes.STDIN_FILEOUT)\r
-\r
-\r
- def testBadPrependOptions(self):\r
- t = pipes.Template()\r
-\r
- # try a non-string command\r
- self.assertRaises(TypeError, t.prepend, 7, pipes.STDIN_STDOUT)\r
-\r
- # try a type that isn't recognized\r
- self.assertRaises(ValueError, t.prepend, 'tr a-z A-Z', 'xx')\r
-\r
- # shouldn't be able to prepend a sink\r
- self.assertRaises(ValueError, t.prepend, 'boguscmd', pipes.SINK)\r
-\r
- # check prepending two sources\r
- t = pipes.Template()\r
- t.prepend('boguscmd', pipes.SOURCE)\r
- self.assertRaises(ValueError, t.prepend, 'boguscmd', pipes.SOURCE)\r
-\r
- # command needing file input but with no $IN\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.prepend, 'boguscmd $OUT',\r
- pipes.FILEIN_FILEOUT)\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.prepend, 'boguscmd',\r
- pipes.FILEIN_STDOUT)\r
-\r
- # command needing file output but with no $OUT\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.prepend, 'boguscmd $IN',\r
- pipes.FILEIN_FILEOUT)\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.prepend, 'boguscmd',\r
- pipes.STDIN_FILEOUT)\r
-\r
- def testBadOpenMode(self):\r
- t = pipes.Template()\r
- self.assertRaises(ValueError, t.open, 'bogusfile', 'x')\r
-\r
- def testClone(self):\r
- t = pipes.Template()\r
- t.append('tr a-z A-Z', pipes.STDIN_STDOUT)\r
-\r
- u = t.clone()\r
- self.assertNotEqual(id(t), id(u))\r
- self.assertEqual(t.steps, u.steps)\r
- self.assertNotEqual(id(t.steps), id(u.steps))\r
- self.assertEqual(t.debugging, u.debugging)\r
-\r
-def test_main():\r
- run_unittest(SimplePipeTests)\r
- reap_children()\r
-\r
-if __name__ == "__main__":\r
- test_main()\r