Utilities.TestModuleCommand Class Reference

Inheritance diagram for Utilities.TestModuleCommand:

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

Detailed Description

Definition at line 168 of file Utilities.py.

Member Function Documentation

def Utilities.TestModuleCommand.setup

(

self

)

Definition at line 169 of file Utilities.py.

        def setup(self):
            None

def Utilities.TestModuleCommand.testIgnoreFiltersOnPath

(

self

)

Definition at line 171 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 189 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 250 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])

    unittest.main()