TestWriteFileMetaDataHITS Namespace Reference

Functions
def	writeFileMetaData (flags)
def	testMetaData (file_name)
def	main ()

Function Documentation

main()

def TestWriteFileMetaDataHITS.main

(

)

Run a job writing a file with FileMetaData

Definition at line 45 of file TestWriteFileMetaDataHITS.py.

def main():
    """Run a job writing a file with FileMetaData"""
    msg = Logging.logging.getLogger("TestFileMetaData")
 
    flags = AllConfigFlags.initConfigFlags()
    flags.Exec.OutputLevel = Constants.DEBUG
    flags.Input.Files = TestDefaults.defaultTestFiles.HITS_RUN2
    flags.Output.HITSFileName = "test.pool.root"
    flags.Concurrency.NumThreads = 4
    flags.Concurrency.NumConcurrentEvents = 4
    flags.lock()
 
    write = MainServicesConfig.MainServicesCfg(flags)
    write.merge(PoolReadCfg(flags))
    write.merge(writeFileMetaData(flags))
    write.run(100)
 
    try:
        if testMetaData(flags.Output.HITSFileName):
            msg.info("File contains xAOD::FileMetaData")
            return 0
        msg.error("File does not contain xAOD::FileMetaData")
    except ReferenceError:
        msg.error("Failed to produce output file")
    except KeyError:
        msg.error("Failed to get metadata item list from file")
    return 1
 
 

testMetaData()

def TestWriteFileMetaDataHITS.testMetaData

(

file_name

)

Check that file metadata is in output

Definition at line 35 of file TestWriteFileMetaDataHITS.py.

def testMetaData(file_name):
    """Check that file metadata is in output"""
    meta = MetaReader.read_metadata([file_name])[file_name]["metadata_items"]
    file_info_items = [
        v for _, v in meta.items() if "FileMetaData" in v
    ]
    print("found file metadata objects:", file_info_items)
    return bool(file_info_items)
 
 

writeFileMetaData()

def TestWriteFileMetaDataHITS.writeFileMetaData

(

flags

)

set up an output stream and xAOD::EventInfo

Definition at line 21 of file TestWriteFileMetaDataHITS.py.

def writeFileMetaData(flags):
    """set up an output stream and xAOD::EventInfo"""
    accumulator = OutputStreamConfig.OutputStreamCfg(flags, streamName="HITS")
    accumulator.merge(SetupMetaDataForStreamCfg(flags, streamName="HITS"))
 
    accumulator.merge(
        xAODEventInfoCnvConfig.EventInfoCnvAlgCfg(
            flags=flags, inputKey="", disableBeamSpot=flags.Common.Project is not Enums.Project.Athena
        )
    )
 
    return accumulator