Merge pull request #442 from jkloetzke/fix-pyyaml-compat

Fix pyyaml compatibility and other potential problems

Author: jkloetzke

.github/workflows/workflow.yaml

@@ -22,14 +22,14 @@ jobs:
             - name: Install dependencies
               run: |
                 python -m pip install --upgrade pip
-                pip install PyYAML codecov schema python-magic pyparsing sphinx wheel
+                pip install PyYAML coverage schema python-magic pyparsing sphinx wheel
                 sudo apt-get update
                 sudo apt-get install cvs
 
             - name: Run unit tests
               run: |
                 git config --global init.defaultBranch master # keep the old name
-                eatmydata ./test/run-tests.sh
+                eatmydata ./test/run-tests.sh -c xml
 
             - name: Build Python package
               run: |
@@ -38,7 +38,7 @@ jobs:
             - name: Upload coverage to Codecov
               # Coverage is not complete on Python <3.7 (see pym/bob/utils.py)
               if: matrix.python-version != '3.6'
-              uses: codecov/codecov-action@v1
+              uses: codecov/codecov-action@v2
 
             - name: Publish package
               if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')

pym/bob/archive.py

@@ -89,10 +89,11 @@ def canUploadJenkins(self):
     def canCache(self):
         return False
 
-    async def uploadPackage(self, step, buildId, audit, content):
+    async def uploadPackage(self, step, buildId, audit, content, executor=None):
         pass
 
-    async def downloadPackage(self, step, buildId, audit, content, caches=[]):
+    async def downloadPackage(self, step, buildId, audit, content, caches=[],
+                              executor=None):
         return False
 
     def upload(self, step, buildIdFile, tgzFile):
@@ -101,19 +102,19 @@ def upload(self, step, buildIdFile, tgzFile):
     def download(self, step, buildIdFile, tgzFile):
         return ""
 
-    async def uploadLocalLiveBuildId(self, step, liveBuildId, buildId):
+    async def uploadLocalLiveBuildId(self, step, liveBuildId, buildId, executor=None):
         pass
 
-    async def downloadLocalLiveBuildId(self, step, liveBuildId):
+    async def downloadLocalLiveBuildId(self, step, liveBuildId, executor=None):
         return None
 
     def uploadJenkinsLiveBuildId(self, step, liveBuildId, buildId, isWin):
         return ""
 
-    async def uploadLocalFingerprint(self, step, key, fingerprint):
+    async def uploadLocalFingerprint(self, step, key, fingerprint, executor=None):
         pass
 
-    async def downloadLocalFingerprint(self, step, key):
+    async def downloadLocalFingerprint(self, step, key, executor=None):
         return None
 
     def uploadJenkinsFingerprint(self, step, keyFile, fingerprintFile):
@@ -199,7 +200,8 @@ def __extractPackage(self, tar, audit, content):
     def _openDownloadFile(self, buildId, suffix):
         raise ArtifactNotFoundError()
 
-    async def downloadPackage(self, step, buildId, audit, content, caches=[]):
+    async def downloadPackage(self, step, buildId, audit, content, caches=[],
+                              executor=None):
         if not self.canDownloadLocal():
             return False
 
@@ -208,7 +210,7 @@ async def downloadPackage(self, step, buildId, audit, content, caches=[]):
         details = " from {}".format(self._remoteName(buildId, suffix))
         with stepAction(step, "DOWNLOAD", content, details=details) as a:
             try:
-                ret, msg, kind = await loop.run_in_executor(None, BaseArchive._downloadPackage,
+                ret, msg, kind = await loop.run_in_executor(executor, BaseArchive._downloadPackage,
                     self, buildId, suffix, audit, content, caches)
                 if not ret: a.fail(msg, kind)
                 return ret
@@ -255,14 +257,14 @@ def _downloadPackage(self, buildId, suffix, audit, content, caches):
             # to prevent ugly backtraces when user presses ctrl+c.
             signal.signal(signal.SIGINT, signal.SIG_DFL)
 
-    async def downloadLocalLiveBuildId(self, step, liveBuildId):
+    async def downloadLocalLiveBuildId(self, step, liveBuildId, executor=None):
         if not self.canDownloadLocal():
             return None
 
         loop = asyncio.get_event_loop()
         with stepAction(step, "MAP-SRC", self._remoteName(liveBuildId, BUILDID_SUFFIX), (INFO,TRACE)) as a:
             try:
-                ret, msg, kind = await loop.run_in_executor(None,
+                ret, msg, kind = await loop.run_in_executor(executor,
                     BaseArchive._downloadLocalFile, self, liveBuildId, BUILDID_SUFFIX)
                 if ret is None: a.fail(msg, kind)
                 return ret
@@ -294,7 +296,7 @@ def _downloadLocalFile(self, key, suffix):
     def _openUploadFile(self, buildId, suffix):
         raise ArtifactUploadError("not implemented")
 
-    async def uploadPackage(self, step, buildId, audit, content):
+    async def uploadPackage(self, step, buildId, audit, content, executor=None):
         if not self.canUploadLocal():
             return
         if not audit:
@@ -307,7 +309,7 @@ async def uploadPackage(self, step, buildId, audit, content):
         details = " to {}".format(self._remoteName(buildId, suffix))
         with stepAction(step, "UPLOAD", content, details=details) as a:
             try:
-                msg, kind = await loop.run_in_executor(None, BaseArchive._uploadPackage,
+                msg, kind = await loop.run_in_executor(executor, BaseArchive._uploadPackage,
                     self, buildId, suffix, audit, content)
                 a.setResult(msg, kind)
             except (concurrent.futures.CancelledError, concurrent.futures.process.BrokenProcessPool):
@@ -339,14 +341,14 @@ def _uploadPackage(self, buildId, suffix, audit, content):
             signal.signal(signal.SIGINT, signal.SIG_DFL)
         return ("ok", EXECUTED)
 
-    async def uploadLocalLiveBuildId(self, step, liveBuildId, buildId):
+    async def uploadLocalLiveBuildId(self, step, liveBuildId, buildId, executor=None):
         if not self.canUploadLocal():
             return
 
         loop = asyncio.get_event_loop()
         with stepAction(step, "CACHE-BID", self._remoteName(liveBuildId, BUILDID_SUFFIX), (INFO,TRACE)) as a:
             try:
-                msg, kind = await loop.run_in_executor(None, BaseArchive._uploadLocalFile, self, liveBuildId, BUILDID_SUFFIX, buildId)
+                msg, kind = await loop.run_in_executor(executor, BaseArchive._uploadLocalFile, self, liveBuildId, BUILDID_SUFFIX, buildId)
                 a.setResult(msg, kind)
             except (concurrent.futures.CancelledError, concurrent.futures.process.BrokenProcessPool):
                 raise BuildError("Upload of build-id interrupted.")
@@ -372,26 +374,26 @@ def _uploadLocalFile(self, key, suffix, content):
             signal.signal(signal.SIGINT, signal.SIG_DFL)
         return ("ok", EXECUTED)
 
-    async def uploadLocalFingerprint(self, step, key, fingerprint):
+    async def uploadLocalFingerprint(self, step, key, fingerprint, executor=None):
         if not self.canUploadLocal():
             return
 
         loop = asyncio.get_event_loop()
         with stepAction(step, "CACHE-FPR", self._remoteName(key, FINGERPRINT_SUFFIX)) as a:
             try:
-                msg, kind = await loop.run_in_executor(None, BaseArchive._uploadLocalFile, self, key, FINGERPRINT_SUFFIX, fingerprint)
+                msg, kind = await loop.run_in_executor(executor, BaseArchive._uploadLocalFile, self, key, FINGERPRINT_SUFFIX, fingerprint)
                 a.setResult(msg, kind)
             except (concurrent.futures.CancelledError, concurrent.futures.process.BrokenProcessPool):
                 raise BuildError("Upload of build-id interrupted.")
 
-    async def downloadLocalFingerprint(self, step, key):
+    async def downloadLocalFingerprint(self, step, key, executor=None):
         if not self.canDownloadLocal():
             return None
 
         loop = asyncio.get_event_loop()
         with stepAction(step, "MAP-FPRNT", self._remoteName(key, FINGERPRINT_SUFFIX)) as a:
             try:
-                ret, msg, kind = await loop.run_in_executor(None,
+                ret, msg, kind = await loop.run_in_executor(executor,
                     BaseArchive._downloadLocalFile, self, key, FINGERPRINT_SUFFIX)
                 if ret is None: a.fail(msg, kind)
                 return ret
@@ -1193,16 +1195,16 @@ def canDownloadJenkins(self):
     def canUploadJenkins(self):
         return any(i.canUploadJenkins() for i in self.__archives)
 
-    async def uploadPackage(self, step, buildId, audit, content):
+    async def uploadPackage(self, step, buildId, audit, content, executor=None):
         for i in self.__archives:
             if not i.canUploadLocal(): continue
-            await i.uploadPackage(step, buildId, audit, content)
+            await i.uploadPackage(step, buildId, audit, content, executor=executor)
 
-    async def downloadPackage(self, step, buildId, audit, content):
+    async def downloadPackage(self, step, buildId, audit, content, executor=None):
         for i in self.__archives:
             if not i.canDownloadLocal(): continue
             caches = [ a for a in self.__archives if (a is not i) and a.canCache() ]
-            if await i.downloadPackage(step, buildId, audit, content, caches):
+            if await i.downloadPackage(step, buildId, audit, content, caches, executor):
                 return True
         return False
 
@@ -1216,16 +1218,16 @@ def download(self, step, buildIdFile, tgzFile):
             i.download(step, buildIdFile, tgzFile) for i in self.__archives
             if i.canDownloadJenkins())
 
-    async def uploadLocalLiveBuildId(self, step, liveBuildId, buildId):
+    async def uploadLocalLiveBuildId(self, step, liveBuildId, buildId, executor=None):
         for i in self.__archives:
             if not i.canUploadLocal(): continue
-            await i.uploadLocalLiveBuildId(step, liveBuildId, buildId)
+            await i.uploadLocalLiveBuildId(step, liveBuildId, buildId, executor=executor)
 
-    async def downloadLocalLiveBuildId(self, step, liveBuildId):
+    async def downloadLocalLiveBuildId(self, step, liveBuildId, executor=None):
         ret = None
         for i in self.__archives:
             if not i.canDownloadLocal(): continue
-            ret = await i.downloadLocalLiveBuildId(step, liveBuildId)
+            ret = await i.downloadLocalLiveBuildId(step, liveBuildId, executor=executor)
             if ret is not None: break
         return ret
 
@@ -1234,16 +1236,16 @@ def uploadJenkinsLiveBuildId(self, step, liveBuildId, buildId, isWin):
             i.uploadJenkinsLiveBuildId(step, liveBuildId, buildId, isWin)
             for i in self.__archives if i.canUploadJenkins())
 
-    async def uploadLocalFingerprint(self, step, key, fingerprint):
+    async def uploadLocalFingerprint(self, step, key, fingerprint, executor=None):
         for i in self.__archives:
             if not i.canUploadLocal(): continue
-            await i.uploadLocalFingerprint(step, key, fingerprint)
+            await i.uploadLocalFingerprint(step, key, fingerprint, executor=executor)
 
-    async def downloadLocalFingerprint(self, step, key):
+    async def downloadLocalFingerprint(self, step, key, executor=None):
         ret = None
         for i in self.__archives:
             if not i.canDownloadLocal(): continue
-            ret = await i.downloadLocalFingerprint(step, key)
+            ret = await i.downloadLocalFingerprint(step, key, executor=executor)
             if ret is not None: break
         return ret
 

pym/bob/cmds/build/build.py

@@ -39,7 +39,7 @@ def _downloadArgument(arg):
             return arg
         raise argparse.ArgumentTypeError("{} invalid.".format(arg))
     def _downloadLayerArgument(arg):
-        if re.match('^(yes|no|forced)=\S+$', arg):
+        if re.match(r'^(yes|no|forced)=\S+$', arg):
             return arg
         raise argparse.ArgumentTypeError("{} invalid.".format(arg))
 
@@ -139,7 +139,7 @@ def _downloadLayerArgument(arg):
 
     startTime = time.time()
 
-    with EventLoopWrapper() as loop:
+    with EventLoopWrapper() as (loop, executor):
         recipes = RecipeSet()
         recipes.defineHook('releaseNameFormatter', LocalBuilder.releaseNameFormatter)
         recipes.defineHook('developNameFormatter', LocalBuilder.developNameFormatter)
@@ -223,6 +223,7 @@ def _downloadLayerArgument(arg):
                                args.preserve_env, envWhiteList, bobRoot, args.clean,
                                args.no_logfiles)
 
+        builder.setExecutor(executor)
         builder.setArchiveHandler(getArchiver(recipes))
         builder.setUploadMode(args.upload)
         builder.setDownloadMode(args.download)

pym/bob/cmds/build/builder.py

@@ -307,6 +307,10 @@ def __init__(self, verbose, force, skipDeps, buildOnly, preserveEnv,
         self.__share = NullShare()
         self.__useSharedPackages = False
         self.__installSharedPackages = False
+        self.__executor = None
+
+    def setExecutor(self, executor):
+        self.__executor = executor
 
     def setArchiveHandler(self, archive):
         self.__archive = archive
@@ -622,7 +626,8 @@ async def _runShell(self, step, scriptName, logger, cleanWorkspace=None):
 
         invoker = Invoker(spec, self.__preserveEnv, self.__noLogFile,
             self.__verbose >= INFO, self.__verbose >= NORMAL,
-            self.__verbose >= DEBUG, self.__bufferedStdIO)
+            self.__verbose >= DEBUG, self.__bufferedStdIO,
+            executor=self.__executor)
         if step.jobServer() and self.__jobServer:
             invoker.setMakeParameters(self.__jobServer.getMakeFd(), self.__jobs)
         ret = await invoker.executeStep(InvocationMode.CALL, cleanWorkspace)
@@ -643,7 +648,8 @@ async def _runLocalSCMs(self, step, logger):
         spec = StepSpec.fromStep(step, logFile=logFile)
         invoker = Invoker(spec, self.__preserveEnv, self.__noLogFile,
             self.__verbose >= INFO, self.__verbose >= NORMAL,
-            self.__verbose >= DEBUG, self.__bufferedStdIO)
+            self.__verbose >= DEBUG, self.__bufferedStdIO,
+            executor=self.__executor)
         ret = await invoker.executeLocalSCMs()
         if not self.__bufferedStdIO: ttyReinit() # work around MSYS2 messing up the console
         if ret == -int(signal.SIGINT):
@@ -1081,7 +1087,8 @@ async def _cookCheckoutStep(self, checkoutStep, depth):
         if created and self.__archive.canUploadLocal() and checkoutStep.hasLiveBuildId():
             liveBId = checkoutStep.calcLiveBuildId()
             if liveBId is not None:
-                await self.__archive.uploadLocalLiveBuildId(checkoutStep, liveBId, checkoutHash)
+                await self.__archive.uploadLocalLiveBuildId(checkoutStep, liveBId, checkoutHash,
+                    executor=self.__executor)
 
         # We're done. The sanity check below won't change the result but would
         # trigger this step again.
@@ -1333,7 +1340,7 @@ async def _downloadPackage(self, packageStep, depth, packageBuildId):
         if BobState().getResultHash(prettyPackagePath) is None:
             audit = os.path.join(prettyPackagePath, "..", "audit.json.gz")
             wasDownloaded = await self.__archive.downloadPackage(packageStep,
-                packageBuildId, audit, prettyPackagePath)
+                packageBuildId, audit, prettyPackagePath, executor=self.__executor)
             if wasDownloaded:
                 self.__statistic.packagesDownloaded += 1
                 BobState().setInputHashes(prettyPackagePath,
@@ -1400,7 +1407,7 @@ async def _cookPackageStep(self, packageStep, checkoutOnly, depth, mayUpOrDownlo
             audit = await self._generateAudit(packageStep, depth, packageHash)
             if mayUpOrDownload and self.__archive.canUploadLocal():
                 await self.__archive.uploadPackage(packageStep, packageBuildId,
-                    audit, prettyPackagePath)
+                    audit, prettyPackagePath, executor=self.__executor)
 
         # Rehash directory if content was changed
         if workspaceChanged:
@@ -1499,7 +1506,7 @@ async def __translateLiveBuildId(self, step, liveBId):
         if bid is not None:
             return bid
 
-        bid = await self.__archive.downloadLocalLiveBuildId(step, liveBId)
+        bid = await self.__archive.downloadLocalLiveBuildId(step, liveBId, executor=self.__executor)
         if bid is not None:
             BobState().setBuildId(key, bid)
 
@@ -1727,7 +1734,8 @@ async def __calcFingerprintTask(self, step, sandbox, key, depth):
         async with self.__runners:
             # If this is built in a sandbox then the artifact cache may help...
             if sandbox:
-                fingerprint = await self.__archive.downloadLocalFingerprint(sandbox, key)
+                fingerprint = await self.__archive.downloadLocalFingerprint(sandbox, key,
+                    executor=self.__executor)
             else:
                 fingerprint = None
 
@@ -1743,7 +1751,8 @@ async def __calcFingerprintTask(self, step, sandbox, key, depth):
                 # only be run once so we don't need to worry here about duplicate
                 # uploads.
                 if sandbox:
-                    await self.__archive.uploadLocalFingerprint(sandbox, key, fingerprint)
+                    await self.__archive.uploadLocalFingerprint(sandbox, key, fingerprint,
+                        executor=self.__executor)
 
             # Cache result so that we don't ever need to spawn a task
             self.__fingerprints[key] = fingerprint
@@ -1757,7 +1766,8 @@ async def __calcFingerprintTask(self, step, sandbox, key, depth):
 
     async def __runFingerprintScript(self, step, logger):
         spec = StepSpec.fromStep(step, None, self.__envWhiteList)
-        invoker = Invoker(spec, self.__preserveEnv, True, True, True, False, True)
+        invoker = Invoker(spec, self.__preserveEnv, True, True, True, False, True,
+                          executor=self.__executor)
         (ret, stdout, stderr) = await invoker.executeFingerprint()
 
         if ret == -int(signal.SIGINT):
@@ -1780,7 +1790,8 @@ async def __runScmSwitch(self, step, scmPath, scm, oldSpec):
         spec = StepSpec.fromStep(step, None, self.__envWhiteList, logFile)
         invoker = Invoker(spec, self.__preserveEnv, self.__noLogFile,
             self.__verbose >= INFO, self.__verbose >= NORMAL,
-            self.__verbose >= DEBUG, self.__bufferedStdIO)
+            self.__verbose >= DEBUG, self.__bufferedStdIO,
+            executor=self.__executor)
 
         class Abort(Exception):
             pass

pym/bob/cmds/invoke.py

@@ -45,18 +45,21 @@ def doInvoke(argv, bobRoot):
         raise BuildError("Error reading spec: " + str(e))
 
     # Let's do it...
-    with EventLoopWrapper() as loop:
+    with EventLoopWrapper() as (loop, executor):
         if args.mode == 'shell':
-            invoker = Invoker(spec, args.preserve_env, True, True, True, False, False)
+            invoker = Invoker(spec, args.preserve_env, True, True, True, False,
+                              False, executor=executor)
             ret = loop.run_until_complete(invoker.executeStep(InvocationMode.SHELL,
                 args.clean, args.keep_sandbox))
         elif args.mode == 'run':
             invoker = Invoker(spec, args.preserve_env, args.no_logfiles,
-                verbosity >= 2, verbosity >= 1, verbosity >= 3, False)
+                verbosity >= 2, verbosity >= 1, verbosity >= 3, False,
+                executor=executor)
             ret = loop.run_until_complete(invoker.executeStep(InvocationMode.CALL,
                 args.clean, args.keep_sandbox))
         elif args.mode == 'fingerprint':
-            invoker = Invoker(spec, args.preserve_env, True, True, True, verbosity >= 3, False)
+            invoker = Invoker(spec, args.preserve_env, True, True, True,
+                              verbosity >= 3, False, executor=executor)
             (ret, stdout, stderr) = loop.run_until_complete(invoker.executeFingerprint(args.keep_sandbox))
             if ret == 0:
                 sys.stdout.buffer.write(stdout)