tests/test_exporter.py

   1 # Copyright (C) 2010 Canonical Ltd
   2 #
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU General Public License as published by
   5 # the Free Software Foundation; either version 2 of the License, or
   6 # (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program; if not, write to the Free Software
  15 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  16
  17 """Test the exporter."""
  18
  19 import os
  20 import tempfile
  21 import gzip
  22
  23 from bzrlib import tests
  24
  25 from bzrlib.plugins.fastimport.exporter import (
  26     _get_output_stream,
  27     )
  28
  29 from bzrlib.plugins.fastimport.tests import (
  30     FastimportFeature,
  31     )
  32
  33
  34 class TestOutputStream(tests.TestCase):
  35
  36     _test_needs_features = [FastimportFeature]
  37
  38     def test_get_output_stream_stdout(self):
  39         # - returns standard out
  40         self.assertIsNot(None, _get_output_stream("-"))
  41
  42     def test_get_source_gz(self):
  43         fd, filename = tempfile.mkstemp(suffix=".gz")
  44         os.close(fd)
  45         stream = _get_output_stream(filename)
  46         stream.write("bla")
  47         stream.close()
  48         # files ending in .gz are automatically decompressed.
  49         f = gzip.GzipFile(filename)
  50         self.assertEquals("bla", f.read())
  51         f.close()
  52
  53     def test_get_source_file(self):
  54         # other files are opened as regular files.
  55         fd, filename = tempfile.mkstemp()
  56         os.close(fd)
  57         stream = _get_output_stream(filename)
  58         stream.write("foo")
  59         stream.close()
  60         f = open(filename, 'r')
  61         self.assertEquals("foo", f.read())
  62         f.close()