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 209 of file Utilities.py.

Member Function Documentation

def Utilities.TestModuleCommand.setup

(

self

)

Definition at line 210 of file Utilities.py.

        def setup(self):
            None

def Utilities.TestModuleCommand.testConvertToSingleModuleEndPaths

(

self

)

Definition at line 374 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 349 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 212 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 230 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 291 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])