def _queryRPMTags(rpmfile, taglist):
 queryFormat = RECORD_SEPARATOR.join([('[%%{%s}%s]' % (tag, UNIT_SEPARATOR)) for tag in taglist])
 def parseTagValues(output):
  res = {}
  for (tag, values) in zip(taglist, output.split(RECORD_SEPARATOR)):
   values = values.strip(UNIT_SEPARATOR).split(UNIT_SEPARATOR)
   res[tag] = values
  return res
 def checkErrorResult(failure):
  failure.trap(IOError)
  if str(failure.value).startswith("got stderr: 'error: "):
   newFailure = Failure(SkipTest("rpm is missing its package database. Run 'sudo rpm -qa > /dev/null' to create one."))
  else:
   newFailure = failure
  d = failure.value.processEnded
  d.addBoth((lambda _: newFailure))
  return d
 d = utils.getProcessOutput('rpm', ('-q', '--queryformat', queryFormat, '-p', rpmfile))
 d.addCallbacks(parseTagValues, checkErrorResult)
 return d

def _queryRPMTags ( rpmfile , taglist ) : queryFormat = RECORD_SEPARATOR . join ( [ ( '[%%{%s}%s]' % ( tag , UNIT_SEPARATOR ) ) for tag in taglist ] ) def parseTagValues ( output ) : res = { } for ( tag , values ) in zip ( taglist , output . split ( RECORD_SEPARATOR ) ) : values = values . strip ( UNIT_SEPARATOR ) . split ( UNIT_SEPARATOR ) res [ tag ] = values return res def checkErrorResult ( failure ) : failure . trap ( IOError ) if str ( failure . value ) . startswith ( "got stderr: 'error: " ) : newFailure = Failure ( SkipTest ( "rpm is missing its package database. Run 'sudo rpm -qa > /dev/null' to create one." ) ) else : newFailure = failure d = failure . value . processEnded d . addBoth ( ( lambda _ : newFailure ) ) return d d = utils . getProcessOutput ( 'rpm' , ( '-q' , '--queryformat' , queryFormat , '-p' , rpmfile ) ) d . addCallbacks ( parseTagValues , checkErrorResult ) return d
helper function to read the given header tags from the given rpm file .