Utilities.TestModuleCommand Class Reference

Inheritance diagram for Utilities.TestModuleCommand:

Public Member Functions
def	setup (self)
def	testConvertToSingleModuleEndPaths (self)
def	testCreateTaskWithAllProducersAndFilters (self)
def	testIgnoreFiltersOnPath (self)
def	testNoSchedule (self)
def	testWithSchedule (self)

Detailed Description

Definition at line 214 of file Utilities.py.

Member Function Documentation

setup()

def Utilities.TestModuleCommand.setup

(

self

)

Definition at line 215 of file Utilities.py.

        def setup(self):
            None

testConvertToSingleModuleEndPaths()

def Utilities.TestModuleCommand.testConvertToSingleModuleEndPaths

(

self

)

Definition at line 379 of file Utilities.py.

        def testConvertToSingleModuleEndPaths(self):
            import FWCore.ParameterSet.Config as cms
            process = cms.Process("TEST")
            process.a = cms.EDAnalyzer("A")
            process.b = cms.EDAnalyzer("B")
            process.c = cms.EDProducer("C")
            process.ep = cms.EndPath(process.a+process.b,cms.Task(process.c))
            self.assertEqual(process.ep.dumpPython(),'cms.EndPath(process.a+process.b, cms.Task(process.c))\n')
            convertToSingleModuleEndPaths(process)
            self.assertEqual(False,hasattr(process,"ep"))
            self.assertEqual(process.a_endpath.dumpPython(),'cms.EndPath(process.a, cms.Task(process.c))\n')
            self.assertEqual(process.b_endpath.dumpPython(),'cms.EndPath(process.b, cms.Task(process.c))\n')
 
    unittest.main()

References Utilities.convertToSingleModuleEndPaths().

testCreateTaskWithAllProducersAndFilters()

def Utilities.TestModuleCommand.testCreateTaskWithAllProducersAndFilters

(

self

)

Definition at line 354 of file Utilities.py.

        def testCreateTaskWithAllProducersAndFilters(self):
 
            import FWCore.ParameterSet.Config as cms
            process = cms.Process("TEST")
 
            process.a = cms.EDProducer("AProd")
            process.b = cms.EDProducer("BProd")
            process.c = cms.EDProducer("CProd")
 
            process.f1 = cms.EDFilter("Filter")
            process.f2 = cms.EDFilter("Filter2")
            process.f3 = cms.EDFilter("Filter3")
 
            process.out1 = cms.OutputModule("Output1")
            process.out2 = cms.OutputModule("Output2")
 
            process.analyzer1 = cms.EDAnalyzer("analyzerType1")
            process.analyzer2 = cms.EDAnalyzer("analyzerType2")
 
            process.task = createTaskWithAllProducersAndFilters(process)
            process.path = cms.Path(process.a, process.task)
 
            self.assertEqual(process.task.dumpPython(),'cms.Task(process.a, process.b, process.c, process.f1, process.f2, process.f3)\n')
            self.assertEqual(process.path.dumpPython(),'cms.Path(process.a, process.task)\n')
 

References Utilities.createTaskWithAllProducersAndFilters().

testIgnoreFiltersOnPath()

def Utilities.TestModuleCommand.testIgnoreFiltersOnPath

(

self

)

Definition at line 217 of file Utilities.py.

        def testIgnoreFiltersOnPath(self):
            import FWCore.ParameterSet.Config as cms
            process = cms.Process("Test")
            
            process.f1 = cms.EDFilter("F1")
            process.f2 = cms.EDFilter("F2")
            process.f3 = cms.EDFilter("F3")
            process.f4 = cms.EDFilter("F4")
            process.f5 = cms.EDFilter("F5")
            process.f6 = cms.EDFilter("F6")
            process.t1 = cms.Task(process.f5)
            process.t2 = cms.Task(process.f6)
            process.s = cms.Sequence(process.f4, process.t1)
            
            process.p =  cms.Path(process.f1+cms.ignore(process.f2)+process.f3+process.s, process.t2)
            ignoreAllFiltersOnPath(process.p)
            self.assertEqual(process.p.dumpPython(),'cms.Path(cms.ignore(process.f1)+cms.ignore(process.f2)+cms.ignore(process.f3)+cms.ignore(process.f4), process.t1, process.t2)\n')
 

testNoSchedule()

def Utilities.TestModuleCommand.testNoSchedule

(

self

)

Definition at line 235 of file Utilities.py.

        def testNoSchedule(self):
            import FWCore.ParameterSet.Config as cms
            process = cms.Process("TEST")
 
            process.a = cms.EDProducer("AProd")
            process.b = cms.EDProducer("BProd")
            process.c = cms.EDProducer("CProd")
            process.d = cms.EDProducer("DProd")
            process.m = cms.EDProducer("MProd")
            process.n = cms.EDProducer("NProd")
            process.r = cms.EDProducer("RProd")
            process.s = cms.EDProducer("SProd")
 
            process.t1 = cms.Task(process.m)
            t2 = cms.Task(process.n)
 
            process.f1 = cms.EDFilter("Filter")
            process.f2 = cms.EDFilter("Filter2")
            process.f3 = cms.EDFilter("Filter3")
            process.f4 = cms.EDFilter("FIlter4")
 
            process.out1 = cms.OutputModule("Output1")
            process.out2 = cms.OutputModule("Output2")
 
            process.analyzer1 = cms.EDAnalyzer("analyzerType1")
            process.analyzer2 = cms.EDAnalyzer("analyzerType2")
 
            process.p1 = cms.Path(process.a+process.b+process.f1+process.analyzer1+cms.ignore(process.d)+cms.ignore(process.f2))
            process.p4 = cms.Path(process.a+process.f2+process.b+~process.f1+cms.ignore(process.f4))
            process.p2 = cms.Path(process.a+process.b)
            process.p3 = cms.Path(process.f1, process.t1, t2)
 
            process.t3 = cms.Task(process.r)
            process.t4 = cms.Task(process.s)
            process.s1 = cms.Sequence(~process.a, process.t3)
            process.p5 = cms.Path(process.b + process.s1, process.t4)
            process.end1 = cms.EndPath(process.out1+process.out2+process.analyzer1+process.analyzer2+process.a+process.b+cms.ignore(process.f1))
            process.end2 = cms.EndPath()
            convertToUnscheduled(process)
            self.assert_(hasattr(process,'p2'))
            self.assert_(hasattr(process,'a'))
            self.assert_(hasattr(process,'b'))
            self.assert_(hasattr(process,'c'))
            self.assert_(hasattr(process,'d'))
            self.assert_(hasattr(process,'f1'))
            self.assert_(hasattr(process,'f2'))
            self.assert_(hasattr(process,'f3'))
            self.assert_(hasattr(process,'f4'))
            self.assert_(hasattr(process,'out1'))
            self.assert_(hasattr(process,'out2'))
            self.assert_(hasattr(process,'analyzer1'))
            self.assert_(hasattr(process,'analyzer2'))
 
            self.assertEqual(process.p1.dumpPython(),'cms.Path(process.f1+process.analyzer1, cms.Task(process.a, process.b, process.d, process.f2))\n')
            self.assertEqual(process.p2.dumpPython(),'cms.Path(cms.Task(process.a, process.b))\n')
            self.assertEqual(process.p3.dumpPython(),'cms.Path(process.f1, cms.Task(process.m, process.n))\n')
            self.assertEqual(process.p4.dumpPython(),'cms.Path(process.f2+~process.f1, cms.Task(process.a, process.b, process.f4))\n')
            self.assertEqual(process.p5.dumpPython(),'cms.Path(cms.Task(process.a, process.b), cms.Task(process.r, process.s))\n')
            self.assertEqual(process.end1.dumpPython(),'cms.EndPath(process.out1+process.out2+process.analyzer1+process.analyzer2, cms.Task(process.a, process.b, process.f1))\n')
            self.assertEqual(process.end2.dumpPython(),'cms.EndPath()\n')
 

References Utilities.convertToUnscheduled().

testWithSchedule()

def Utilities.TestModuleCommand.testWithSchedule

(

self

)

Definition at line 296 of file Utilities.py.

        def testWithSchedule(self):
            import FWCore.ParameterSet.Config as cms
            process = cms.Process("TEST")
 
            process.a = cms.EDProducer("AProd")
            process.b = cms.EDProducer("BProd")
            process.c = cms.EDProducer("CProd")
            process.d = cms.EDProducer("DProd")
            process.m = cms.EDProducer("MProd")
            process.n = cms.EDProducer("NProd")
 
            process.t1 = cms.Task(process.m)
            t2 = cms.Task(process.n)
 
            process.f1 = cms.EDFilter("Filter")
            process.f2 = cms.EDFilter("Filter2")
            process.f3 = cms.EDFilter("Filter3")
            process.f4 = cms.EDFilter("Filter4")
 
            process.out1 = cms.OutputModule("Output1")
            process.out2 = cms.OutputModule("Output2")
 
            process.analyzer1 = cms.EDAnalyzer("analyzerType1")
            process.analyzer2 = cms.EDAnalyzer("analyzerType2")
 
            process.p1 = cms.Path(process.a+process.b+cms.ignore(process.f1)+process.d+process.f2)
            process.p4 = cms.Path(process.a+process.f2+process.b+~process.f1)
            process.p2 = cms.Path(process.a+process.b)
            process.p3 = cms.Path(process.f1)
            process.p5 = cms.Path(process.a+process.f4) #not used on schedule
            process.end1 = cms.EndPath(process.out1+process.out2+process.analyzer1+process.analyzer2+process.a+process.b+cms.ignore(process.f1))
            process.end2 = cms.EndPath()
 
            process.schedule = cms.Schedule(process.p1,process.p4,process.p2,process.p3,process.end1,process.end2,tasks=[process.t1,t2])
            convertToUnscheduled(process)
            self.assert_(hasattr(process,'p2'))
            self.assert_(hasattr(process,'a'))
            self.assert_(hasattr(process,'b'))
            self.assert_(hasattr(process,'c'))
            self.assert_(hasattr(process,'d'))
            self.assert_(hasattr(process,'f1'))
            self.assert_(hasattr(process,'f2'))
            self.assert_(hasattr(process,'f3'))
            self.assert_(hasattr(process,"f4"))
            self.assert_(hasattr(process,"p5"))
 
            self.assertEqual(process.p1.dumpPython(),'cms.Path(process.f2, cms.Task(process.a, process.b, process.d, process.f1))\n')
            self.assertEqual(process.p2.dumpPython(),'cms.Path(cms.Task(process.a, process.b))\n')
            self.assertEqual(process.p3.dumpPython(),'cms.Path(process.f1)\n')
            self.assertEqual(process.p4.dumpPython(),'cms.Path(process.f2+~process.f1, cms.Task(process.a, process.b))\n')
            self.assertEqual(process.p5.dumpPython(),'cms.Path(process.f4, cms.Task(process.a))\n')
            self.assertEqual(process.end1.dumpPython(),'cms.EndPath(process.out1+process.out2+process.analyzer1+process.analyzer2, cms.Task(process.a, process.b, process.f1))\n')
            self.assertEqual(process.end2.dumpPython(),'cms.EndPath()\n')
 
            self.assertEqual([p for p in process.schedule],[process.p1,process.p4,process.p2,process.p3,process.end1,process.end2])
            listOfTasks = list(process.schedule._tasks)
            self.assertEqual(listOfTasks, [process.t1,t2])
 

References Utilities.convertToUnscheduled(), and list().