archive: add retry to more functions

the archive cmd relies on those functions. they can also fail with an unreliable
server. applying the retry mechanism here might mitigate those issues

Author: MarcKe

pym/bob/archive.py

@@ -934,18 +934,18 @@ def _putUploadFile(self, path, tmp, overwrite):
         return self.__retry(lambda: self._webdav.upload(path, tmp, overwrite))
 
     def _listDir(self, path):
-        path_info = self._webdav.listdir(path)
+        path_info = self.__retry(lambda: self._webdav.listdir(path))
         entries = []
         for info in path_info:
             if info["path"]:
                 entries.append(removePrefix(info["path"], path + "/"))
         return entries
 
     def _delete(self, filename):
-        self._webdav.delete(filename)
+        self.__retry(lambda: self._webdav.delete(filename))
 
     def _stat(self, filename):
-        stats = self._webdav.stat(filename)
+        stats = self.__retry(lambda: self._webdav.stat(filename))
         if stats['etag'] is not None and not stats['etag'].startswith('W/'):
             return stats['etag']
         if not stats['mdate'] or not stats['len']:
@@ -954,14 +954,14 @@ def _stat(self, filename):
         return struct.pack('=dL', parsedate_to_datetime(stats['mdate']).timestamp(), stats['len'])
 
     def _getAudit(self, filename):
-        downloader = self._webdav.getPartialDownloader("/".join([self.__url.path, filename]))
+        downloader = self.__retry(lambda: self._webdav.getPartialDownloader("/".join([self.__url.path, filename])))
         while True:
             try:
                 file = io.BytesIO(downloader.get())
                 return self._extractAudit(fileobj=file)
             except (EOFError, tarfile.ReadError):
                 # partial downloader reached EOF or could not extract the audit from the tarfile, so we get more data
-                downloader.more()
+                self.__retry(lambda: downloader.more())
                 pass
 
     def _getArchiveUri(self):

test/unit/test_archive.py

@@ -9,7 +9,9 @@
 from unittest.mock import patch
 import asyncio
 import base64
+import gzip
 import http.server
+import posixpath
 import os, os.path
 import socketserver
 import stat
@@ -23,13 +25,16 @@
 from bob.archive import DummyArchive, HttpArchive, getArchiver
 from bob.errors import BuildError
 from bob.utils import runInEventLoop, getProcessPoolExecutor
+from bob.webdav import HTTPException
 
 DOWNLOAD_ARITFACT = b'\x00'*20
 NOT_EXISTS_ARTIFACT = b'\x01'*20
 WRONG_VERSION_ARTIFACT = b'\x02'*20
 ERROR_UPLOAD_ARTIFACT = b'\x03'*20
 ERROR_DOWNLOAD_ARTIFACT = b'\x04'*20
 BROKEN_ARTIFACT = b'\xba\xdc\x0f\xfe'*5
+VALID_ARTIFACT = b'\x05'*20
+EMPTY_AUDIT = '{"artifact":{"variant-id":"1","build-id":"1","artifact-id":"1","result-hash":"1","meta":"1","build":"1","dependencies":{}}, "references":[]}'
 
 UPLOAD1_ARTIFACT = b'\x10'*20
 UPLOAD2_ARTIFACT = b'\x11'*20
@@ -63,17 +68,23 @@ def run(coro):
 
 class Base:
 
-    def __createArtifact(self, bid, version="1"):
+    def _createArtifact(self, bid, version="1", valid_data=False):
         bid = hexlify(bid).decode("ascii")
         name = os.path.join(self.repo.name, bid[0:2], bid[2:4], bid[4:] + "-1.tgz")
         os.makedirs(os.path.dirname(name), exist_ok=True)
-        return self.__createArtifactByName(name, version)
+        return self.__createArtifactByName(name, version, valid_data)
 
-    def __createArtifactByName(self, name, version="1"):
+    def __createArtifactByName(self, name, version="1", valid_data=False):
         pax = { 'bob-archive-vsn' : version }
         with tarfile.open(name, "w|gz", format=tarfile.PAX_FORMAT, pax_headers=pax) as tar:
             with NamedTemporaryFile() as audit:
-                audit.write(b'AUDIT')
+                if valid_data:
+                    # add valid empty audit file
+                    gzf = gzip.GzipFile(mode='wb', fileobj=audit)
+                    gzf.write(EMPTY_AUDIT.encode('utf-8'))
+                    gzf.close()
+                else:
+                    audit.write(b'AUDIT')
                 audit.seek(0)
                 tar.addfile(tar.gettarinfo(arcname="meta/audit.json.gz", fileobj=audit), audit)
             with TemporaryDirectory() as content:
@@ -83,7 +94,7 @@ def __createArtifactByName(self, name, version="1"):
 
         return name
 
-    def __createBuildId(self, bid):
+    def _createBuildId(self, bid):
         bid = hexlify(bid).decode("ascii")
         name = os.path.join(self.repo.name, bid[0:2], bid[2:4], bid[4:] + "-1.buildid")
         os.makedirs(os.path.dirname(name), exist_ok=True)
@@ -94,12 +105,6 @@ def __createBuildId(self, bid):
     def setUp(self):
         # create repo
         self.repo = TemporaryDirectory()
-
-        # add artifacts
-        self.dummyFileName = self.__createArtifact(DOWNLOAD_ARITFACT)
-        self.__createArtifact(WRONG_VERSION_ARTIFACT, "0")
-        self.__createBuildId(DOWNLOAD_ARITFACT)
-
         self.executor = getProcessPoolExecutor()
 
     def tearDown(self):
@@ -172,6 +177,10 @@ def __getSingleArchiveInstance(self, spec):
     def setUp(self):
         super().setUp()
 
+        # add artifacts
+        self.dummyFileName = self._createArtifact(DOWNLOAD_ARITFACT)
+        self._createArtifact(WRONG_VERSION_ARTIFACT, "0")
+        self._createBuildId(DOWNLOAD_ARITFACT)
         # create ERROR_DOWNLOAD_ARTIFACT that is there but cannot be opened
         bid = hexlify(ERROR_DOWNLOAD_ARTIFACT).decode("ascii")
         os.makedirs(os.path.join(self.repo.name, bid[0:2], bid[2:4], bid[4:] + "-1.tgz"), exist_ok=True)
@@ -682,6 +691,7 @@ class TestHttpArchiveRetries(Base, TestCase):
 
     def setUp(self):
         super().setUp()
+        self.VALID_FILE = self._createArtifact(VALID_ARTIFACT, valid_data=True)
         self.spec = {'backend' : 'http', 'name' : 'http-archive', 'flags' : ['download', 'upload', 'managed']}
 
     def _getHttpArchiveInstance(self, port):
@@ -698,7 +708,7 @@ def _testRetries(self, r):
             with TemporaryDirectory() as tmp:
                 audit = os.path.join(tmp, "audit.json.gz")
                 content = os.path.join(tmp, "workspace")
-                self.assertTrue(run(archive.downloadPackage(DummyStep(), DOWNLOAD_ARITFACT, audit, content,
+                self.assertTrue(run(archive.downloadPackage(DummyStep(), VALID_ARTIFACT, audit, content,
                                                             executor=self.executor)))
         # server fails one more time than retries in archive sepc -> fail
         with HttpServerMock(repoPath=self.repo.name, retries=r+1) as srv:
@@ -707,7 +717,7 @@ def _testRetries(self, r):
             with TemporaryDirectory() as tmp:
                 audit = os.path.join(tmp, "audit.json.gz")
                 content = os.path.join(tmp, "workspace")
-                self.assertFalse(run(archive.downloadPackage(DummyStep(), DOWNLOAD_ARITFACT, audit, content,
+                self.assertFalse(run(archive.downloadPackage(DummyStep(), VALID_ARTIFACT, audit, content,
                                                              executor=self.executor)))
 
     def testRetriesWithNoRetries(self):
@@ -718,3 +728,53 @@ def testRetriesWithOneRetry(self):
 
     def testRetriesWithMultipleRetries(self):
         self._testRetries(5)
+
+    def testRetriesList(self):
+        self.spec['retries'] = 1
+        with HttpServerMock(repoPath=self.repo.name, retries=1) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            self.assertIsNotNone(archive._listDir('/'))
+        with HttpServerMock(repoPath=self.repo.name, retries=2) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            with self.assertRaises(HTTPException):
+                archive._listDir('/')
+
+    def testRetriesStat(self):
+        bid = hexlify(VALID_ARTIFACT).decode("ascii")
+        filepath = posixpath.join(bid[0:2], bid[2:4], bid[4:] + "-1.tgz")
+        self.spec['retries'] = 1
+        with HttpServerMock(repoPath=self.repo.name, retries=1) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            self.assertIsNotNone(archive._stat(filepath))
+        with HttpServerMock(repoPath=self.repo.name, retries=2) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            with self.assertRaises(HTTPException):
+                archive._stat(filepath)
+
+    def testRetriesDelete(self):
+        filename = 'test'
+        filepath = os.path.join(self.repo.name, filename)
+        with open(filepath, 'w') as f:
+            f.write('test')
+        self.spec['retries'] = 1
+        with HttpServerMock(repoPath=self.repo.name, retries=1) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            self.assertIsNone(archive._delete(filename))
+        with open(filepath, 'w') as f:
+            f.write('test')
+        with HttpServerMock(repoPath=self.repo.name, retries=2) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            with self.assertRaises(HTTPException):
+                archive._delete(filename)
+
+    def testRetriesAudit(self):
+        bid = hexlify(VALID_ARTIFACT).decode("ascii")
+        filepath = posixpath.join(bid[0:2], bid[2:4], bid[4:] + "-1.tgz")
+        self.spec['retries'] = 1
+        with HttpServerMock(repoPath=self.repo.name, retries=1) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            self.assertIsNotNone(archive._getAudit(filepath))
+        with HttpServerMock(repoPath=self.repo.name, retries=2) as srv:
+            archive = self._getHttpArchiveInstance(srv.port)
+            with self.assertRaises(HTTPException):
+                archive._getAudit(filepath)