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

def Utilities.TestModuleCommand.setup

(

self

)

Definition at line 215 of file Utilities.py.

        def setup(self):
            None

def Utilities.TestModuleCommand.testConvertToSingleModuleEndPaths

(

self

)

Definition at line 379 of file Utilities.py.

References Utilities.convertToSingleModuleEndPaths().

        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(None),'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(None),'cms.EndPath(process.a, cms.Task(process.c))\n')
            self.assertEqual(process.b_endpath.dumpPython(None),'cms.EndPath(process.b, cms.Task(process.c))\n')

    unittest.main()

def Utilities.TestModuleCommand.testCreateTaskWithAllProducersAndFilters

(

self

)

Definition at line 354 of file Utilities.py.

References Utilities.createTaskWithAllProducersAndFilters().

        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(None),'cms.Task(process.a, process.b, process.c, process.f1, process.f2, process.f3)\n')
            self.assertEqual(process.path.dumpPython(None),'cms.Path(process.a, process.task)\n')

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(None),'cms.Path(cms.ignore(process.f1)+cms.ignore(process.f2)+cms.ignore(process.f3)+cms.ignore(process.f4), process.t1, process.t2)\n')

def Utilities.TestModuleCommand.testNoSchedule

(

self

)

Definition at line 235 of file Utilities.py.

References Utilities.convertToUnscheduled().

        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(None),'cms.Path(process.f1+process.analyzer1, cms.Task(process.a, process.b, process.d, process.f2))\n')
            self.assertEqual(process.p2.dumpPython(None),'cms.Path(cms.Task(process.a, process.b))\n')
            self.assertEqual(process.p3.dumpPython(None),'cms.Path(process.f1, cms.Task(process.m, process.n))\n')
            self.assertEqual(process.p4.dumpPython(None),'cms.Path(process.f2+~process.f1, cms.Task(process.a, process.b, process.f4))\n')
            self.assertEqual(process.p5.dumpPython(None),'cms.Path(cms.Task(process.a, process.b), cms.Task(process.r, process.s))\n')
            self.assertEqual(process.end1.dumpPython(None),'cms.EndPath(process.out1+process.out2+process.analyzer1+process.analyzer2, cms.Task(process.a, process.b, process.f1))\n')
            self.assertEqual(process.end2.dumpPython(None),'cms.EndPath()\n')

def Utilities.TestModuleCommand.testWithSchedule

(

self

)

Definition at line 296 of file Utilities.py.

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

        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(None),'cms.Path(process.f2, cms.Task(process.a, process.b, process.d, process.f1))\n')
            self.assertEqual(process.p2.dumpPython(None),'cms.Path(cms.Task(process.a, process.b))\n')
            self.assertEqual(process.p3.dumpPython(None),'cms.Path(process.f1)\n')
            self.assertEqual(process.p4.dumpPython(None),'cms.Path(process.f2+~process.f1, cms.Task(process.a, process.b))\n')
            self.assertEqual(process.p5.dumpPython(None),'cms.Path(process.f4, cms.Task(process.a))\n')
            self.assertEqual(process.end1.dumpPython(None),'cms.EndPath(process.out1+process.out2+process.analyzer1+process.analyzer2, cms.Task(process.a, process.b, process.f1))\n')
            self.assertEqual(process.end2.dumpPython(None),'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])