*: use open() as context manager to ensure files are closed

cloudooo/file.py

@@ -68,7 +68,8 @@ class File(object):
     file_path = tempfile.mktemp(suffix=".%s" % self.source_format,
                                 dir=self.directory_name)
     # stores the data in temporary file
-    open(file_path, 'wb').write(self.original_data)
+    with open(file_path, 'wb') as f:
+      f.write(self.original_data)
     # If is a zipfile is need extract all files from whitin the compressed file
     if is_zipfile(file_path):
       zipfile = ZipFile(file_path)
@@ -106,12 +107,14 @@ class File(object):
           zipfile.write(file)
       finally:
         zipfile.close()
-      opened_zip = open(zip_path, 'r').read()
+      with open(zip_path, 'rb') as f:
+        opened_zip = f.read()
       remove(zip_path)
       chdir(current_dir_url)
       return opened_zip
     else:
-      return open(self.url, 'r').read()
+      with open(self.url, 'rb') as f:
+        return f.read()
 
   def getUrl(self):
     """Returns full path."""

cloudooo/handler/ffmpeg/tests/testFfmpegServer.py

@@ -44,14 +44,17 @@ class TestServer(TestCase):
     self.runConversionList(self.ConversionScenarioList())
 
   def FaultConversionScenarioList(self):
-    return [
-            # Test to verify if server fail when a empty string is sent
-            ('', '', ''),
+    scenario_list = [
+      # Test to verify if server fail when a empty file is sent
+      (b'', '', ''),
+    ]
     # Try convert one video for a invalid format
-            (open(join('data', 'test.ogv')).read(), 'ogv', 'xyz'),
+    with open(join('data', 'test.ogv'), 'rb') as f:
+      scenario_list.append((f.read(), 'ogv', 'xyz'))
     # Try convert one video to format not possible
-            (open(join('data', 'test.ogv')).read(), 'ogv', 'moov'),
-            ]
+    with open(join('data', 'test.ogv'), 'rb') as f:
+      scenario_list.append((f.read(), 'ogv', 'moov'))
+    return scenario_list
 
   def testFaultConversion(self):
     """Test fail convertion of Invalid video files"""

cloudooo/handler/imagemagick/tests/testImagemagickHandler.py

@@ -41,7 +41,8 @@ class TestHandler(HandlerTestCase):
 
   def testConvertPNGtoJPG(self):
     """Test conversion of png to jpg"""
-    png_file = open("data/test.png").read()
+    with open("data/test.png", "rb") as f:
+      png_file = f.read()
     handler = Handler(self.tmp_url, png_file, "png", **self.kw)
     jpg_file = handler.convert("jpg")
     mime = magic.Magic(mime=True)
@@ -50,7 +51,8 @@ class TestHandler(HandlerTestCase):
 
   def testgetMetadataFromImage(self):
     """Test if metadata is extracted from image correctly"""
-    png_file = open("data/test.png").read()
+    with open("data/test.png", "rb") as f:
+      png_file = f.read()
     handler = Handler(self.tmp_url, png_file, "png", **self.kw)
     metadata = handler.getMetadata()
     self.assertEqual(metadata.get("Compression"), "Zip")

cloudooo/handler/imagemagick/tests/testImagemagickServer.py

@@ -44,14 +44,17 @@ class TestServer(TestCase):
     self.runConversionList(self.ConversionScenarioList())
 
   def FaultConversionScenarioList(self):
-    return [
-            # Test to verify if server fail when a empty string is sent
-            ('', '', ''),
-            # Try convert one video for a invalid format
-            (open(join('data', 'test.png')).read(), 'png', 'xyz'),
-            # Try convert one video to format not possible
-            (open(join('data', 'test.png')).read(), 'png', '8bim'),
+    scenario_list = [
+      # Test to verify if server fail when a empty file is sent
+      (b'', '', ''),
     ]
+    # Try convert one png for a invalid format
+    with open(join('data', 'test.png'), 'rb') as f:
+      scenario_list.append((f.read(), 'png', 'xyz'))
+    # Try convert one png to format not possible
+    with open(join('data', 'test.png'), 'rb') as f:
+      scenario_list.append((f.read(), 'png', '8bim'))
+    return scenario_list
 
   def testFaultConversion(self):
     """Test fail convertion of Invalid image files"""

cloudooo/handler/ooo/tests/testOooFileSystemDocument.py

@@ -58,7 +58,8 @@ class TestFileSystemDocument(unittest.TestCase):
     document_filename = "document"
     document_test_url = path.join(self.fsdocument.directory_name,
                                   document_filename)
-    open(document_test_url, 'wb').write(decodestring("Test Document"))
+    with open(document_test_url, 'wb') as f:
+      f.write(decodestring(b"Test Document"))
     self.fsdocument.reload(document_test_url)
     self.assertEqual(path.exists(old_document_url), False)
     self.assertNotEqual(self.fsdocument.original_data,
@@ -82,7 +83,8 @@ class TestFileSystemDocument(unittest.TestCase):
   def testLoadDocumentFile(self):
     """Test if the document is created correctly"""
     url = self.fsdocument.getUrl()
-    tmp_document = open(url, 'r').read()
+    with open(url, 'rb') as f:
+      tmp_document = f.read()
     self.assertEqual(self.data, tmp_document)
     self.fsdocument.trash()
     self.assertEqual(path.exists(url), False)
@@ -93,7 +95,8 @@ class TestFileSystemDocument(unittest.TestCase):
     document_filename = "document"
     document_test_url = path.join(self.fsdocument.directory_name,
                                                document_filename)
-    open(document_test_url, 'wb').write(self.data)
+    with open(document_test_url, 'wb') as f:
+      f.write(self.data)
     self.fsdocument.reload(document_test_url)
     url = self.fsdocument.getUrl()
     self.assertEqual(path.exists(old_document_url), False)
@@ -103,7 +106,8 @@ class TestFileSystemDocument(unittest.TestCase):
 
   def testZipDocumentList(self):
     """Tests if the zip file is returned correctly"""
-    open(path.join(self.fsdocument.directory_name, 'document2'), 'w').write('test')
+    with open(path.join(self.fsdocument.directory_name, 'document2'), 'w') as f:
+      f.write('test')
     zip_file = self.fsdocument.getContent(True)
     mime = magic.Magic(mime=True)
     mimetype = mime.from_buffer(zip_file)
@@ -118,8 +122,8 @@ class TestFileSystemDocument(unittest.TestCase):
 
   def testSendZipFile(self):
     """Tests if the htm is extrated from zipfile"""
-    zip_input_url = 'data/test.zip'
-    data = open(zip_input_url).read()
+    with open('./data/test.zip', 'rb') as f:
+      data = f.read()
     zipdocument = FileSystemDocument(self.tmp_url, data, 'zip')
     mime = magic.Magic(mime=True)
     mimetype = mime.from_buffer(zipdocument.getContent(True))

cloudooo/handler/ooo/tests/testOooGranulator.py

@@ -39,7 +39,8 @@ from cloudooo.handler.ooo.granulator import OOGranulator
 class TestOOGranulator(HandlerTestCase):
 
   def setUp(self):
-    data = open('./data/granulate_test.odt').read()
+    with open('./data/granulate_test.odt', 'rb') as f:
+      data = f.read()
     self.oogranulator = OOGranulator(data, 'odt')
 
   def testOdfWithoutContentXml(self):
@@ -56,7 +57,8 @@ class TestOOGranulator(HandlerTestCase):
 
   def testgetTableItemList(self):
     """Test if getTableItemList() returns the right tables list"""
-    data = open('./data/granulate_table_test.odt').read()
+    with open('./data/granulate_table_test.odt', 'rb') as f:
+      data = f.read()
     oogranulator = OOGranulator(data, 'odt')
     table_list = [('Developers', ''),
                   ('Prices', 'Table 1: Prices table from Mon Restaurant'),
@@ -65,7 +67,8 @@ class TestOOGranulator(HandlerTestCase):
 
   def testGetTable(self):
     """Test if getTable() returns on odf file with the right table"""
-    data = open('./data/granulate_table_test.odt').read()
+    with open('./data/granulate_table_test.odt', 'rb') as f:
+      data = f.read()
     oogranulator = OOGranulator(data, 'odt')
     table_data_doc = oogranulator.getTable('Developers')
     content_xml_str = ZipFile(StringIO(table_data_doc)).read('content.xml')
@@ -79,21 +82,24 @@ class TestOOGranulator(HandlerTestCase):
 
   def testGetTableItemWithoutSuccess(self):
     """Test if getTable() returns None for an non existent table name"""
-    data = open('./data/granulate_table_test.odt').read()
+    with open('./data/granulate_table_test.odt', 'rb') as f:
+      data = f.read()
     oogranulator = OOGranulator(data, 'odt')
     table_data = oogranulator.getTable('NonExistentTable')
     self.assertEqual(table_data, None)
 
   def testGetColumnItemList(self):
     """Test if getColumnItemList() returns the right table columns list"""
-    data = open('./data/granulate_table_test.odt').read()
+    with open('./data/granulate_table_test.odt', 'rb') as f:
+      data = f.read()
     oogranulator = OOGranulator(data, 'odt')
     self.assertEqual([[0, 'Name'], [1, 'Country']],
                       oogranulator.getColumnItemList('SoccerTeams'))
 
   def testGetLineItemList(self):
     """Test if getLineItemList() returns the right table lines list"""
-    data = open('./data/granulate_table_test.odt').read()
+    with open('./data/granulate_table_test.odt', 'rb') as f:
+      data = f.read()
     oogranulator = OOGranulator(data, 'odt')
     matrix = [['Name', 'Phone', 'Email'],
              ['Hugo', '+55 (22) 8888-8888', 'hugomaia@tiolive.com'],
@@ -121,7 +127,8 @@ class TestOOGranulator(HandlerTestCase):
 
   def testGetImageSuccessfully(self):
     """Test if getImage() returns the right image file successfully"""
-    data = open('./data/granulate_test.odt').read()
+    with open('./data/granulate_test.odt', 'rb') as f:
+      data = f.read()
     zip = ZipFile(StringIO(data))
     image_id = '10000000000000C80000009C38276C51.jpg'
     original_image = zip.read('Pictures/%s' % image_id)
@@ -131,13 +138,14 @@ class TestOOGranulator(HandlerTestCase):
   def testGetImageWithoutSuccess(self):
     """Test if getImage() returns an empty string for not existent id"""
     obtained_image = self.oogranulator.getImage('anything.png')
-    self.assertEqual('', obtained_image)
+    self.assertEqual(b'', obtained_image)
 
   def testGetParagraphItemList(self):
     """Test if getParagraphItemList() returns the right paragraphs list, with
     the ids always in the same order"""
-    for i in range(5):
-      data = open('./data/granulate_test.odt').read()
+    for _ in range(5):
+      with open('./data/granulate_test.odt', 'rb') as f:
+        data = f.read()
       oogranulator = OOGranulator(data, 'odt')
       paragraph_list = oogranulator.getParagraphItemList()
       self.assertEqual((0, 'P3'), paragraph_list[0])
@@ -162,7 +170,8 @@ class TestOOGranulator(HandlerTestCase):
 
   def testGetChapterItemList(self):
     """Test if getChapterItemList() returns the right chapters list"""
-    data = open('./data/granulate_chapters_test.odt').read()
+    with open('./data/granulate_chapters_test.odt', 'rb') as f:
+      data = f.read()
     oogranulator = OOGranulator(data, 'odt')
     self.assertEqual([(0, 'Title 0'), (1, 'Title 1'), (2, 'Title 2'),
                        (3, 'Title 3'), (4, 'Title 4'), (5, 'Title 5'),
@@ -172,7 +181,8 @@ class TestOOGranulator(HandlerTestCase):
 
   def testGetChapterItem(self):
     """Test if getChapterItem() returns the right chapter"""
-    data = open("./data/granulate_chapters_test.odt").read()
+    with open("./data/granulate_chapters_test.odt", 'rb') as f:
+      data = f.read()
     oogranulator = OOGranulator(data, 'odt')
     self.assertEqual(['Title 1', 1], oogranulator.getChapterItem(1))
 

cloudooo/handler/ooo/tests/testOooHandler.py

@@ -48,9 +48,8 @@ class TestHandler(HandlerTestCase):
 
   def _save_document(self, document_output_url, data):
     """Create document in file system"""
-    new_file = open(document_output_url, "w")
-    new_file.write(data)
-    new_file.close()
+    with open(document_output_url, "wb") as f:
+      f.write(data)
     self._file_path_list.append(document_output_url)
 
   def _assert_document_output(self, document, expected_mimetype):
@@ -70,18 +69,20 @@ class TestHandler(HandlerTestCase):
 
   def testConvertOdtToDoc(self):
     """Test convert ODT to DOC"""
-    data = encodestring(open("data/test.odt").read())
+    with open("data/test.odt", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'odt')
     doc_exported = handler.convert("doc")
     self._assert_document_output(doc_exported, "application/msword")
 
   def testConvertDocToOdt(self):
     """Test convert DOC to ODT"""
-    data = encodestring(open("data/test.doc").read())
+    with open("data/test.doc", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'doc')
     doc_exported = handler.convert("odt")
     self._assert_document_output(doc_exported,
@@ -89,9 +90,10 @@ class TestHandler(HandlerTestCase):
 
   def testGetMetadata(self):
     """Test getMetadata"""
-    data = encodestring(open("data/test.odt").read())
+    with open("data/test.odt", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'odt')
     metadata = handler.getMetadata()
     self.assertEqual(metadata.get('MIMEType'),
@@ -102,9 +104,10 @@ class TestHandler(HandlerTestCase):
 
   def testSetMetadata(self):
     """Test setMetadata"""
-    data = encodestring(open("data/test.odt").read())
+    with open("data/test.odt", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'odt')
     new_data = handler.setMetadata({"Title": "cloudooo Test -"})
     new_handler = Handler(self.tmp_url,
@@ -113,7 +116,7 @@ class TestHandler(HandlerTestCase):
     metadata = new_handler.getMetadata()
     self.assertEqual(metadata.get('Title'), "cloudooo Test -")
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'odt')
     new_data = handler.setMetadata({"Title": "Namie's working record"})
     new_handler = Handler(self.tmp_url,
@@ -125,9 +128,10 @@ class TestHandler(HandlerTestCase):
   def testConvertWithOpenOfficeStopped(self):
     """Test convert with openoffice stopped"""
     openoffice.stop()
-    data = encodestring(open("data/test.doc").read())
+    with open("data/test.doc", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'doc')
     doc_exported = handler.convert("odt")
     self._assert_document_output(doc_exported,
@@ -136,9 +140,10 @@ class TestHandler(HandlerTestCase):
   def testGetMetadataWithOpenOfficeStopped(self):
     """Test getMetadata with openoffice stopped"""
     openoffice.stop()
-    data = encodestring(open("data/test.odt").read())
+    with open("data/test.odt", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'odt')
     metadata = handler.getMetadata()
     self.assertEqual(metadata.get('Title'), 'title')
@@ -148,9 +153,10 @@ class TestHandler(HandlerTestCase):
   def testSetMetadataWithOpenOfficeStopped(self):
     """Test setMetadata with openoffice stopped"""
     openoffice.stop()
-    data = encodestring(open("data/test.doc").read())
+    with open("data/test.doc", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'doc')
     new_data = handler.setMetadata({"Title": "cloudooo Test -"})
     new_handler = Handler(self.tmp_url,
@@ -162,9 +168,10 @@ class TestHandler(HandlerTestCase):
   def testRefreshOdt(self):
     """Test refresh argument"""
     # Check when refreshing is disabled
-    data = encodestring(open("data/test_fields.odt").read())
+    with open("data/test_fields.odt", "rb") as f:
+      data = f.read()
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'odt',
                         refresh=False)
     doc_exported = handler.convert("odt")
@@ -176,9 +183,8 @@ class TestHandler(HandlerTestCase):
                                        namespaces=content_tree.nsmap))
 
     # Check when refreshing is enabled
-    data = encodestring(open("data/test_fields.odt").read())
     handler = Handler(self.tmp_url,
-                        decodestring(data),
+                        data,
                         'odt',
                         refresh=True)
     doc_exported = handler.convert("odt")

cloudooo/handler/ooo/tests/testOooOdfDocument.py

@@ -37,7 +37,8 @@ from cloudooo.handler.ooo.document import OdfDocument
 class TestOdfDocument(HandlerTestCase):
 
   def setUp(self):
-    data = open('./data/granulate_test.odt').read()
+    with open('./data/granulate_test.odt', 'rb') as f:
+      data = f.read()
     self.oodocument = OdfDocument(data, 'odt')
 
   def testReceivedGoodFile(self):
@@ -47,7 +48,7 @@ class TestOdfDocument(HandlerTestCase):
   def testGetContentXml(self):
     """Test if the getContentXml method returns the content.xml file"""
     content_xml = self.oodocument.getContentXml()
-    self.assertTrue('The content of this file is just' in content_xml)
+    self.assertIn(b'The content of this file is just', content_xml)
 
   def testGetExistentFile(self):
     """Test if the getFile method returns the requested file"""

cloudooo/handler/ooo/tests/testOooUnoConverter.py

@@ -50,7 +50,8 @@ class TestUnoConverter(HandlerTestCase):
     """ """
     openoffice.acquire()
     self.hostname, self.port = openoffice.getAddress()
-    data = open("data/test.odt", 'r').read()
+    with open("data/test.odt", "rb") as f:
+      data = f.read()
     self.document = FileSystemDocument(self.tmp_url, data, 'odt')
 
   def tearDown(self):

cloudooo/handler/pdf/tests/testPdfHandler.py

@@ -41,14 +41,16 @@ class TestHandler(HandlerTestCase):
 
   def testConvertPDFtoText(self):
     """Test conversion of pdf to txt"""
-    pdf_document = open("data/test.pdf").read()
+    with open("data/test.pdf", "rb") as f:
+      pdf_document = f.read()
     handler = Handler(self.tmp_url, pdf_document, "pdf", **self.kw)
     txt_document = handler.convert("txt")
-    self.assertTrue(txt_document.startswith("UNG Docs Architecture"))
+    self.assertTrue(txt_document.startswith(b"UNG Docs Architecture"))
 
   def testgetMetadata(self):
     """Test if the metadata are extracted correctly"""
-    pdf_document = open("data/test.pdf").read()
+    with open("data/test.pdf", "rb") as f:
+      pdf_document = f.read()
     handler = Handler(self.tmp_url, pdf_document, "pdf", **self.kw)
     metadata = handler.getMetadata()
     self.assertEqual(type(metadata), DictType)
@@ -57,7 +59,8 @@ class TestHandler(HandlerTestCase):
 
   def testsetMetadata(self):
     """Test if the metadata is inserted correctly"""
-    pdf_document = open("data/test.pdf").read()
+    with open("data/test.pdf", "rb") as f:
+      pdf_document = f.read()
     handler = Handler(self.tmp_url, pdf_document, "pdf", **self.kw)
     metadata_dict = {"title": "Set Metadata Test", "creator": "gabriel\'@"}
     new_document = handler.setMetadata(metadata_dict)

cloudooo/handler/pdf/tests/testPdfServer.py

@@ -45,14 +45,17 @@ class TestServer(TestCase):
     self.runConversionList(self.ConversionScenarioList())
 
   def FaultConversionScenarioList(self):
-    return [
-            # Test to verify if server fail when a empty string is sent
-            ('', '', ''),
+    scenario_list = [
+      # Test to verify if server fail when a empty file is sent
+      (b'', '', ''),
+    ]
     # Try convert one video for a invalid format
-            (open(join('data', 'test.pdf')).read(), 'pdf', 'xyz'),
+    with open(join('data', 'test.pdf'), 'rb') as f:
+      scenario_list.append((f.read(), 'pdf', 'xyz'))
     # Try convert one video to format not possible
-            (open(join('data', 'test.pdf')).read(), 'pdf', 'ogv'),
-            ]
+    with open(join('data', 'test.pdf'), 'rb') as f:
+      scenario_list.append((f.read(), 'pdf', 'ogv'))
+    return scenario_list
 
   def testFaultConversion(self):
     """Test fail convertion of Invalid pdf files"""

cloudooo/handler/wkhtmltopdf/tests/testWkhtmltopdfHandler.py

@@ -40,7 +40,8 @@ class TestHandler(HandlerTestCase):
     self.kw = dict(env=dict(PATH=self.env_path))
 
   def _testBase(self, html_path, **conversion_kw):
-    html_file = open(html_path).read()
+    with open(html_path, 'rb') as f:
+      html_file = f.read()
     handler = Handler(self.tmp_url, html_file, "html", **self.kw)
     pdf_file = handler.convert("pdf", **conversion_kw)
     mime = magic.Magic(mime=True)
@@ -67,16 +68,18 @@ class TestHandler(HandlerTestCase):
 
   def testConvertHtmlWithTableOfContent(self):
     """Test conversion of html with an additional table of content"""
+    with open("data/test_toc.xsl", 'rb') as f:
+      xsl_style_sheet_data = f.read()
     self._testBase(
       "data/test_with_toc.html",
       toc=True,
-      xsl_style_sheet_data=b64encode(open("data/test_toc.xsl").read()),
+      xsl_style_sheet_data=b64encode(xsl_style_sheet_data),
     )
     # XXX how to check for table of content presence ?
 
   def testsetMetadata(self):
     """ Test if metadata are inserted correclty """
-    handler = Handler(self.tmp_url, "", "png", **self.kw)
+    handler = Handler(self.tmp_url, b"", "png", **self.kw)
     self.assertRaises(NotImplementedError, handler.setMetadata)
 
   def testGetAllowedConversionFormatList(self):

cloudooo/handler/wkhtmltopdf/tests/testWkhtmltopdfServer.py

@@ -44,10 +44,11 @@ class TestServer(TestCase):
     self.runConversionList(self.ConversionScenarioList())
 
   def FaultConversionScenarioList(self):
-    return [
-      # Test to verify if server fail when a empty string is sent
-      ('', '', ''),
-      # Try convert one html for a invalid format
-      (open(join('data', 'test_with_png_dataurl.html')).read(), 'html', 'xyz'),
+    scenario_list = [
+      # Test to verify if server fail when a empty file is sent
+      (b'', '', ''),
     ]
-
+      # Try convert one html for a invalid format
+    with open(join('data', 'test_with_png_dataurl.html'), 'rb') as f:
+      scenario_list.append((f.read(), 'html', 'xyz'))
+    return scenario_list

cloudooo/handler/x2t/handler.py

@@ -321,7 +321,8 @@ class Handler(object):
               absolute_path = os.path.join(root, file_name)
               file_name = os.path.join(relative_root, file_name)
               zipfile.write(absolute_path, file_name)
-        output_data = open(output_file_name).read()
+        with open(output_file_name, 'rb') as f:
+          output_data = f.read()
       finally:
         os.unlink(output_file_name)
       return output_data

cloudooo/handler/x2t/tests/testX2tServer.py

@@ -52,12 +52,14 @@ class TestServer(TestCase):
     self.runConversionList(self.ConversionScenarioList())
 
   def FaultConversionScenarioList(self):
-    return [
-      # Test to verify if server fail when a empty string is sent
-      ('', '', ''),
-      # Try convert one xlsx for a invalid format
-      (open(join('data', 'test.xlsx')).read(), 'xlsx', 'xyz'),
+    scenario_list = [
+      # Test to verify if server fail when a empty file is sent
+      (b'', '', ''),
     ]
+    # Try convert one xlsx for a invalid format
+    with open(join('data', 'test.xlsx'), 'rb') as f:
+      scenario_list.append((f.read(), 'xlsx', 'xyz'))
+    return scenario_list
 
   def test_xlsx_to_xlsy(self):
     with open(join('data', 'test.xlsx')) as f: