Config.TestModuleCommand Class Reference

Inheritance diagram for Config.TestModuleCommand:

Public Member Functions
def	setUp (self)
def	testCloneSequence (self)
def	testContains (self)
def	testDelete (self)
def	testExamples (self)
def	testFreeze (self)
def	testGlobalReplace (self)
def	testImplicitSchedule (self)
def	testModifier (self)
def	testOverride (self)
def	testParameterizable (self)
def	testPath (self)
def	testPrefers (self)
def	testProcessDumpPython (self)
def	testProcessExtend (self)
def	testProcessInsertion (self)
def	testPrune (self)
def	testRefToPSet (self)
def	testSchedule (self)
def	testSecSource (self)
def	testSequence (self)
def	testSequence2 (self)
def	testServiceInProcess (self)
def	testSubProcess (self)
def	testTask (self)
def	testTaskPlaceholder (self)
def	testTypedParameterizable (self)
def	testUsing (self)

Public Attributes
	a
	proc_mod_

Detailed Description

Definition at line 1554 of file Config.py.

Member Function Documentation

def Config.TestModuleCommand.setUp

(

self

)

Nothing to do 

Definition at line 1555 of file Config.py.

        def setUp(self):
            """Nothing to do """
            None

def Config.TestModuleCommand.testCloneSequence

(

self

)

Definition at line 2275 of file Config.py.

References triggerObjects_cff.id, and str.

        def testCloneSequence(self):
            p = Process("test")
            a = EDAnalyzer("MyAnalyzer")
            p.a = a
            a.setLabel("a")
            b = EDAnalyzer("YOurAnalyzer")
            p.b = b
            b.setLabel("b")
            path = Path(a * b)
            p.path = Path(p.a*p.b)
            lookuptable = {id(a): p.a, id(b): p.b}
            #self.assertEqual(str(path),str(path._postProcessFixup(lookuptable)))
            #lookuptable = p._cloneToObjectDict
            #self.assertEqual(str(path),str(path._postProcessFixup(lookuptable)))
            self.assertEqual(str(path),str(p.path))

def Config.TestModuleCommand.testContains

(

self

)

Definition at line 2291 of file Config.py.

        def testContains(self):

            a = EDProducer("a")
            b = EDProducer("b")
            c = EDProducer("c")
            d = EDProducer("d")
            e = EDProducer("e")
            f = EDProducer("f")
            g = EDProducer("g")
            h = EDProducer("h")
            i = EDProducer("i")
            j = EDProducer("j")
            k = EDProducer("k")
            l = EDProducer("l")
            m = EDProducer("m")
            n = EDProducer("n")

            seq1 = Sequence(e)
            task1 = Task(g)
            path = Path(a * c * seq1, task1)

            self.assertTrue(path.contains(a))
            self.assertFalse(path.contains(b))
            self.assertTrue(path.contains(c))
            self.assertFalse(path.contains(d))
            self.assertTrue(path.contains(e))
            self.assertFalse(path.contains(f))
            self.assertTrue(path.contains(g))

            endpath = EndPath(h * i)
            self.assertFalse(endpath.contains(b))
            self.assertTrue(endpath.contains(i))

            seq = Sequence(a * c)
            self.assertFalse(seq.contains(b))
            self.assertTrue(seq.contains(c))

            task2 = Task(l)
            task = Task(j, k, task2)
            self.assertFalse(task.contains(b))
            self.assertTrue(task.contains(j))
            self.assertTrue(task.contains(k))
            self.assertTrue(task.contains(l))

            task3 = Task(m)
            path2 = Path(n)
            sch = Schedule(path, path2, tasks=[task,task3])
            self.assertFalse(sch.contains(b))
            self.assertTrue(sch.contains(a))
            self.assertTrue(sch.contains(c))
            self.assertTrue(sch.contains(e))
            self.assertTrue(sch.contains(g))
            self.assertTrue(sch.contains(n))
            self.assertTrue(sch.contains(j))
            self.assertTrue(sch.contains(k))
            self.assertTrue(sch.contains(l))
            self.assertTrue(sch.contains(m))

def Config.TestModuleCommand.testDelete

(

self

)

Definition at line 2899 of file Config.py.

References ConfigBuilder.dumpPython().

        def testDelete(self):
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.d = EDAnalyzer("OurAnalyzer")
            p.e = EDAnalyzer("OurAnalyzer")
            p.f = EDAnalyzer("OurAnalyzer")
            p.g = EDProducer("OurProducer")
            p.h = EDProducer("YourProducer")
            p.t1 = Task(p.g, p.h)
            t2 = Task(p.g, p.h)
            t3 = Task(p.g, p.h)
            p.s = Sequence(p.d+p.e)
            p.path1 = Path(p.a+p.f+p.s,t2)
            p.endpath1 = EndPath(p.b+p.f)
            p.schedule = Schedule(tasks=[t3])
            self.assertTrue(hasattr(p, 'f'))
            self.assertTrue(hasattr(p, 'g'))
            del p.e
            del p.f
            del p.g
            self.assertFalse(hasattr(p, 'f'))
            self.assertFalse(hasattr(p, 'g'))
            self.assertTrue(p.t1.dumpPython(None) == 'cms.Task(process.h)\n')
            self.assertTrue(p.s.dumpPython(None) == 'cms.Sequence(process.d)\n')
            self.assertTrue(p.path1.dumpPython(None) == 'cms.Path(process.a+process.s, cms.Task(process.h))\n')
            self.assertTrue(p.endpath1.dumpPython(None) == 'cms.EndPath(process.b)\n')
            del p.s
            self.assertTrue(p.path1.dumpPython(None) == 'cms.Path(process.a+(process.d), cms.Task(process.h))\n')
            self.assertTrue(p.schedule_().dumpPython(None) == 'cms.Schedule(tasks=[cms.Task(process.h)])\n')

def Config.TestModuleCommand.testExamples

(

self

)

Definition at line 2517 of file Config.py.

        def testExamples(self):
            p = Process("Test")
            p.source = Source("PoolSource",fileNames = untracked(string("file:reco.root")))
            p.foos = EDProducer("FooProducer")
            p.bars = EDProducer("BarProducer", foos=InputTag("foos"))
            p.out = OutputModule("PoolOutputModule",fileName=untracked(string("file:foos.root")))
            p.bars.foos = 'Foosball'
            self.assertEqual(p.bars.foos, InputTag('Foosball'))
            p.p = Path(p.foos*p.bars)
            p.e = EndPath(p.out)
            p.add_(Service("MessageLogger"))

def Config.TestModuleCommand.testFreeze

(

self

)

Definition at line 2583 of file Config.py.

        def testFreeze(self):
            process = Process("Freeze")
            m = EDProducer("M", p=PSet(i = int32(1)))
            m.p.i = 2
            process.m = m
            # should be frozen
            #self.assertRaises(ValueError, setattr, m.p, 'i', 3)
            #self.assertRaises(ValueError, setattr, m, 'p', PSet(i=int32(1)))
            #self.assertRaises(ValueError, setattr, m.p, 'j', 1)
            #self.assertRaises(ValueError, setattr, m, 'j', 1)
            # But OK to change through the process
            process.m.p.i = 4
            self.assertEqual(process.m.p.i.value(), 4)
            process.m.p = PSet(j=int32(1))
            # should work to clone it, though
            m2 = m.clone(p = PSet(i = int32(5)), j = int32(8))
            m2.p.i = 6
            m2.j = 8

def Config.TestModuleCommand.testGlobalReplace

(

self

)

Definition at line 1953 of file Config.py.

References list(), and cmsPerfHarvest.visit().

        def testGlobalReplace(self):
            p = Process('test')
            p.a = EDAnalyzer("MyAnalyzer")
            old = p.a
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.d = EDProducer("MyProducer")
            old2 = p.d
            p.t1 = Task(p.d)
            t2 = Task(p.d)
            t3 = Task(p.d)
            t4 = Task(p.d)
            t5 = Task(p.d)
            t6 = Task(p.d)
            s = Sequence(p.a*p.b)
            p.s4 = Sequence(p.a*p.b)
            s.associate(t2)
            p.s4.associate(t2)
            p.p = Path(p.c+s+p.a)
            p.e3 = EndPath(p.c+s+p.a)
            new = EDAnalyzer("NewAnalyzer")
            new2 = EDProducer("NewProducer")
            visitor1 = NodeVisitor()
            p.p.visit(visitor1)
            self.assertTrue(visitor1.modules == set([old,old2,p.b,p.c]))
            p.schedule = Schedule(tasks=[t6])
            p.globalReplace("a",new)
            p.globalReplace("d",new2)
            visitor2 = NodeVisitor()
            p.p.visit(visitor2)
            self.assertTrue(visitor2.modules == set([new,new2,p.b,p.c]))
            visitor3 = NodeVisitor()
            p.e3.visit(visitor3)
            self.assertTrue(visitor3.modules == set([new,new2,p.b,p.c]))
            visitor4 = NodeVisitor()
            p.s4.visit(visitor4)
            self.assertTrue(visitor4.modules == set([new,new2,p.b]))
            visitor5 = NodeVisitor()
            p.t1.visit(visitor5)
            self.assertTrue(visitor5.modules == set([new2]))
            visitor6 = NodeVisitor()
            listOfTasks = list(p.schedule._tasks)
            listOfTasks[0].visit(visitor6)
            self.assertTrue(visitor6.modules == set([new2]))

def Config.TestModuleCommand.testImplicitSchedule

(

self

)

Definition at line 2454 of file Config.py.

        def testImplicitSchedule(self):
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.path1 = Path(p.a)
            p.path2 = Path(p.b)
            self.assert_(p.schedule is None)
            pths = p.paths
            keys = pths.keys()
            self.assertEqual(pths[keys[0]],p.path1)
            self.assertEqual(pths[keys[1]],p.path2)
            p.prune()
            self.assert_(hasattr(p, 'a'))
            self.assert_(hasattr(p, 'b'))
            self.assert_(not hasattr(p, 'c'))
            self.assert_(hasattr(p, 'path1'))
            self.assert_(hasattr(p, 'path2'))


            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.path2 = Path(p.b)
            p.path1 = Path(p.a)
            self.assert_(p.schedule is None)
            pths = p.paths
            keys = pths.keys()
            self.assertEqual(pths[keys[1]],p.path1)
            self.assertEqual(pths[keys[0]],p.path2)

def Config.TestModuleCommand.testModifier

(

self

)

Definition at line 2930 of file Config.py.

References cmsPerfStripChart.dict, and resolutioncreator_cfi.object.

        def testModifier(self):
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1))
            def _mod_fred(obj):
                obj.fred = 2
            m1.toModify(p.a,_mod_fred)
            self.assertEqual(p.a.fred.value(),2)
            p.b = EDAnalyzer("YourAnalyzer", wilma = int32(1))
            m1.toModify(p.b, wilma = 2)
            self.assertEqual(p.b.wilma.value(),2)
            self.assert_(p.isUsingModifier(m1))
            #check that Modifier not attached to a process doesn't run
            m1 = Modifier()
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1))
            m1.toModify(p.a,_mod_fred)
            p.b = EDAnalyzer("YourAnalyzer", wilma = int32(1))
            m1.toModify(p.b, wilma = 2)
            self.assertEqual(p.a.fred.value(),1)
            self.assertEqual(p.b.wilma.value(),1)
            self.assertEqual(p.isUsingModifier(m1),False)
            #make sure clones get the changes
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            m1.toModify(p.a, fred = int32(2))
            p.b = p.a.clone(wilma = int32(3))
            self.assertEqual(p.a.fred.value(),2)
            self.assertEqual(p.a.wilma.value(),1)
            self.assertEqual(p.b.fred.value(),2)
            self.assertEqual(p.b.wilma.value(),3)
            #test removal of parameter
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1), fintstones = PSet(fred = int32(1)))
            m1.toModify(p.a, fred = None, fintstones = dict(fred = None))
            self.assertEqual(hasattr(p.a, "fred"), False)
            self.assertEqual(hasattr(p.a.fintstones, "fred"), False)
            self.assertEqual(p.a.wilma.value(),1)
            #test adding a parameter
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1))
            m1.toModify(p.a, wilma = int32(2))
            self.assertEqual(p.a.fred.value(), 1)
            self.assertEqual(p.a.wilma.value(),2)
            #test setting of value in PSet
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", flintstones = PSet(fred = int32(1), wilma = int32(1)))
            m1.toModify(p.a, flintstones = dict(fred = int32(2)))
            self.assertEqual(p.a.flintstones.fred.value(),2)
            self.assertEqual(p.a.flintstones.wilma.value(),1)
            #test proper exception from nonexisting parameter name
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", flintstones = PSet(fred = PSet(wilma = int32(1))))
            self.assertRaises(KeyError, lambda: m1.toModify(p.a, flintstones = dict(imnothere = dict(wilma=2))))
            self.assertRaises(KeyError, lambda: m1.toModify(p.a, foo = 1))
            #test setting a value in a VPSet
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", flintstones = VPSet(PSet(fred = int32(1)), PSet(wilma = int32(1))))
            m1.toModify(p.a, flintstones = {1:dict(wilma = int32(2))})
            self.assertEqual(p.a.flintstones[0].fred.value(),1)
            self.assertEqual(p.a.flintstones[1].wilma.value(),2)
            #test setting a value in a list of values
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = vuint32(1,2,3))
            m1.toModify(p.a, fred = {1:7})
            self.assertEqual(p.a.fred[0],1)
            self.assertEqual(p.a.fred[1],7)
            self.assertEqual(p.a.fred[2],3)
            #test IndexError setting a value in a list to an item key not in the list
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = vuint32(1,2,3))
            raised = False
            try: m1.toModify(p.a, fred = {5:7})
            except IndexError as e: raised = True
            self.assertEqual(raised, True)
            #test TypeError setting a value in a list using a key that is not an int
            m1 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", flintstones = VPSet(PSet(fred = int32(1)), PSet(wilma = int32(1))))
            raised = False
            try: m1.toModify(p.a, flintstones = dict(bogus = int32(37)))
            except TypeError as e: raised = True
            self.assertEqual(raised, True)
            #test that load causes process wide methods to run
            def _rem_a(proc):
                del proc.a
            class ProcModifierMod(object):
                def __init__(self,modifier,func):
                    self.proc_mod_ = modifier.makeProcessModifier(func)
            class DummyMod(object):
                def __init__(self):
                    self.a = EDAnalyzer("Dummy")
            testMod = DummyMod()
            p.extend(testMod)
            self.assert_(hasattr(p,"a"))
            m1 = Modifier()
            p = Process("test",m1)
            testProcMod = ProcModifierMod(m1,_rem_a)
            p.extend(testMod)
            p.extend(testProcMod)
            self.assert_(not hasattr(p,"a"))
            #test ModifierChain
            m1 = Modifier()
            mc = ModifierChain(m1)
            p = Process("test",mc)
            self.assert_(p.isUsingModifier(m1))
            self.assert_(p.isUsingModifier(mc))
            testMod = DummyMod()
            p.b = EDAnalyzer("Dummy2", fred = int32(1))
            m1.toModify(p.b, fred = int32(3))
            p.extend(testMod)
            testProcMod = ProcModifierMod(m1,_rem_a)
            p.extend(testProcMod)
            self.assert_(not hasattr(p,"a"))
            self.assertEqual(p.b.fred.value(),3)
            #check cloneAndExclude
            m1 = Modifier()
            m2 = Modifier()
            mc = ModifierChain(m1,m2)
            mclone = mc.copyAndExclude([m2])
            self.assert_(not mclone._isOrContains(m2))
            self.assert_(mclone._isOrContains(m1))
            m3 = Modifier()
            mc2 = ModifierChain(mc,m3)
            mclone = mc2.copyAndExclude([m2])
            self.assert_(not mclone._isOrContains(m2))
            self.assert_(mclone._isOrContains(m1))
            self.assert_(mclone._isOrContains(m3))
            #check combining
            m1 = Modifier()
            m2 = Modifier()
            p = Process("test",m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            (m1 & m2).toModify(p.a, fred = int32(2))
            self.assertRaises(TypeError, lambda: (m1 & m2).toModify(p.a, 1, wilma=2))
            self.assertEqual(p.a.fred, 1)
            m1 = Modifier()
            m2 = Modifier()
            p = Process("test",m1,m2)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            (m1 & m2).toModify(p.a, fred = int32(2))
            self.assertEqual(p.a.fred, 2)
            m1 = Modifier()
            m2 = Modifier()
            m3 = Modifier()
            p = Process("test",m1,m2,m3)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            (m1 & m2 & m3).toModify(p.a, fred = int32(2))
            self.assertEqual(p.a.fred, 2)
            (m1 & (m2 & m3)).toModify(p.a, fred = int32(3))
            self.assertEqual(p.a.fred, 3)
            ((m1 & m2) & m3).toModify(p.a, fred = int32(4))
            self.assertEqual(p.a.fred, 4)
            #check inverse
            m1 = Modifier()
            m2 = Modifier()
            p = Process("test", m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            (~m1).toModify(p.a, fred=2)
            self.assertEqual(p.a.fred, 1)
            (~m2).toModify(p.a, wilma=2)
            self.assertEqual(p.a.wilma, 2)
            self.assertRaises(TypeError, lambda: (~m1).toModify(p.a, 1, wilma=2))
            self.assertRaises(TypeError, lambda: (~m2).toModify(p.a, 1, wilma=2))
            # check or
            m1 = Modifier()
            m2 = Modifier()
            m3 = Modifier()
            p = Process("test", m1)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            (m1 | m2).toModify(p.a, fred=2)
            self.assertEqual(p.a.fred, 2)
            (m1 | m2 | m3).toModify(p.a, fred=3)
            self.assertEqual(p.a.fred, 3)
            (m3 | m2 | m1).toModify(p.a, fred=4)
            self.assertEqual(p.a.fred, 4)
            ((m1 | m2) | m3).toModify(p.a, fred=5)
            self.assertEqual(p.a.fred, 5)
            (m1 | (m2 | m3)).toModify(p.a, fred=6)
            self.assertEqual(p.a.fred, 6)
            (m2 | m3).toModify(p.a, fred=7)
            self.assertEqual(p.a.fred, 6)
            self.assertRaises(TypeError, lambda: (m1 | m2).toModify(p.a, 1, wilma=2))
            self.assertRaises(TypeError, lambda: (m2 | m3).toModify(p.a, 1, wilma=2))
            # check combinations
            m1 = Modifier()
            m2 = Modifier()
            m3 = Modifier()
            m4 = Modifier()
            p = Process("test", m1, m2)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            (m1 & ~m2).toModify(p.a, fred=2)
            self.assertEqual(p.a.fred, 1)
            (m1 & ~m3).toModify(p.a, fred=2)
            self.assertEqual(p.a.fred, 2)
            (m1 | ~m2).toModify(p.a, fred=3)
            self.assertEqual(p.a.fred, 3)
            (~m1 | ~m2).toModify(p.a, fred=4)
            self.assertEqual(p.a.fred, 3)
            (~m3 & ~m4).toModify(p.a, fred=4)
            self.assertEqual(p.a.fred, 4)
            ((m1 & m3) | ~m4).toModify(p.a, fred=5)
            self.assertEqual(p.a.fred, 5)
            #check toReplaceWith
            m1 = Modifier()
            p = Process("test",m1)
            p.a =EDAnalyzer("MyAnalyzer", fred = int32(1))
            m1.toReplaceWith(p.a, EDAnalyzer("YourAnalyzer", wilma = int32(3)))
            self.assertRaises(TypeError, lambda: m1.toReplaceWith(p.a, EDProducer("YourProducer")))
            p.b =EDAnalyzer("BAn")
            p.c =EDProducer("c")
            p.d =EDProducer("d")
            p.tc = Task(p.c)
            p.td = Task(p.d)
            p.s = Sequence(p.a, p.tc)
            m1.toReplaceWith(p.s, Sequence(p.a+p.b, p.td))
            self.assertEqual(p.a.wilma.value(),3)
            self.assertEqual(p.a.type_(),"YourAnalyzer")
            self.assertEqual(hasattr(p,"fred"),False)
            self.assertTrue(p.s.dumpPython("") == "cms.Sequence(process.a+process.b, process.td)\n")
            p.e =EDProducer("e")
            m1.toReplaceWith(p.td, Task(p.e))
            self.assertTrue(p.td._collection == OrderedSet([p.e]))
            #check toReplaceWith doesn't activate not chosen
            m1 = Modifier()
            p = Process("test")
            p.a =EDAnalyzer("MyAnalyzer", fred = int32(1))
            m1.toReplaceWith(p.a, EDAnalyzer("YourAnalyzer", wilma = int32(3)))
            self.assertEqual(p.a.type_(),"MyAnalyzer")
            #check toReplaceWith and and/not/or combinations
            m1 = Modifier()
            m2 = Modifier()
            m3 = Modifier()
            m4 = Modifier()
            p = Process("test", m1, m2)
            p.a = EDAnalyzer("MyAnalyzer", fred = int32(1), wilma = int32(1))
            self.assertRaises(TypeError, lambda: (m1 & m2).toReplaceWith(p.a, EDProducer("YourProducer")))
            self.assertRaises(TypeError, lambda: (m3 & m4).toReplaceWith(p.a, EDProducer("YourProducer")))
            self.assertRaises(TypeError, lambda: (~m3).toReplaceWith(p.a, EDProducer("YourProducer")))
            self.assertRaises(TypeError, lambda: (~m1).toReplaceWith(p.a, EDProducer("YourProducer")))
            self.assertRaises(TypeError, lambda: (m1 | m3).toReplaceWith(p.a, EDProducer("YourProducer")))
            self.assertRaises(TypeError, lambda: (m3 | m4).toReplaceWith(p.a, EDProducer("YourProducer")))
            (m1 & m2).toReplaceWith(p.a, EDAnalyzer("YourAnalyzer1"))
            self.assertEqual(p.a.type_(), "YourAnalyzer1")
            (m1 & m3).toReplaceWith(p.a, EDAnalyzer("YourAnalyzer2"))
            self.assertEqual(p.a.type_(), "YourAnalyzer1")
            (~m1).toReplaceWith(p.a, EDAnalyzer("YourAnalyzer2"))
            self.assertEqual(p.a.type_(), "YourAnalyzer1")
            (~m3).toReplaceWith(p.a, EDAnalyzer("YourAnalyzer2"))
            self.assertEqual(p.a.type_(), "YourAnalyzer2")
            (m1 | m3).toReplaceWith(p.a, EDAnalyzer("YourAnalyzer3"))
            self.assertEqual(p.a.type_(), "YourAnalyzer3")
            (m3 | m4).toReplaceWith(p.a, EDAnalyzer("YourAnalyzer4"))
            self.assertEqual(p.a.type_(), "YourAnalyzer3")
    unittest.main()

def Config.TestModuleCommand.testOverride

(

self

)

Definition at line 2494 of file Config.py.

        def testOverride(self):
            p = Process('test')
            a = EDProducer("A", a1=int32(0))
            self.assert_(not a.isModified())
            a.a1 = 1
            self.assert_(a.isModified())
            p.a = a
            self.assertEqual(p.a.a1.value(), 1)
            # try adding an unmodified module.
            # should accept it
            p.a = EDProducer("A", a1=int32(2))
            self.assertEqual(p.a.a1.value(), 2)
            # try adding a modified module.  Should throw
            # no longer, since the same (modified) say, geometry
            # could come from more than one cff
            b = EDProducer("A", a1=int32(3))
            b.a1 = 4
            #self.assertRaises(RuntimeError, setattr, *(p,'a',b))
            ps1 = PSet(a = int32(1))
            ps2 = PSet(a = int32(2))
            self.assertRaises(ValueError, EDProducer, 'C', ps1, ps2)
            self.assertRaises(ValueError, EDProducer, 'C', ps1, a=int32(3))

def Config.TestModuleCommand.testParameterizable

(

self

)

Definition at line 1558 of file Config.py.

        def testParameterizable(self):
            p = _Parameterizable()
            self.assertEqual(len(p.parameterNames_()),0)
            p.a = int32(1)
            self.assert_('a' in p.parameterNames_())
            self.assertEqual(p.a.value(), 1)
            p.a = 10
            self.assertEqual(p.a.value(), 10)
            p.a = untracked(int32(1))
            self.assertEqual(p.a.value(), 1)
            self.failIf(p.a.isTracked())
            p.a = untracked.int32(1)
            self.assertEqual(p.a.value(), 1)
            self.failIf(p.a.isTracked())
            p = _Parameterizable(foo=int32(10), bar = untracked(double(1.0)))
            self.assertEqual(p.foo.value(), 10)
            self.assertEqual(p.bar.value(),1.0)
            self.failIf(p.bar.isTracked())
            self.assertRaises(TypeError,setattr,(p,'c',1))
            p = _Parameterizable(a=PSet(foo=int32(10), bar = untracked(double(1.0))))
            self.assertEqual(p.a.foo.value(),10)
            self.assertEqual(p.a.bar.value(),1.0)
            p.b = untracked(PSet(fii = int32(1)))
            self.assertEqual(p.b.fii.value(),1)
            self.failIf(p.b.isTracked())
            #test the fact that values can be shared
            v = int32(10)
            p=_Parameterizable(a=v)
            v.setValue(11)
            self.assertEqual(p.a.value(),11)
            p.a = 12
            self.assertEqual(p.a.value(),12)
            self.assertEqual(v.value(),12)

def Config.TestModuleCommand.testPath

(

self

)

Definition at line 2235 of file Config.py.

References SequenceTypes.ignore(), and str.

        def testPath(self):
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            path = Path(p.a)
            path *= p.b
            path += p.c
            self.assertEqual(str(path),'a+b+c')
            path = Path(p.a*p.b+p.c)
            self.assertEqual(str(path),'a+b+c')
#            path = Path(p.a)*p.b+p.c #This leads to problems with sequences
#            self.assertEqual(str(path),'((a*b)+c)')
            path = Path(p.a+ p.b*p.c)
            self.assertEqual(str(path),'a+b+c')
            path = Path(p.a*(p.b+p.c))
            self.assertEqual(str(path),'a+b+c')
            path = Path(p.a*(p.b+~p.c))
            pathx = Path(p.a*(p.b+ignore(p.c)))
            self.assertEqual(str(path),'a+b+~c')
            p.es = ESProducer("AnESProducer")
            self.assertRaises(TypeError,Path,p.es)

            t = Path()
            self.assertTrue(t.dumpPython(PrintOptions()) == 'cms.Path()\n')

            t = Path(p.a)
            self.assertTrue(t.dumpPython(PrintOptions()) == 'cms.Path(process.a)\n')

            t = Path(Task())
            self.assertTrue(t.dumpPython(PrintOptions()) == 'cms.Path(cms.Task())\n')

            t = Path(p.a, Task())
            self.assertTrue(t.dumpPython(PrintOptions()) == 'cms.Path(process.a, cms.Task())\n')

            p.prod = EDProducer("prodName")
            p.t1 = Task(p.prod)
            t = Path(p.a, p.t1, Task(), p.t1)
            self.assertTrue(t.dumpPython(PrintOptions()) == 'cms.Path(process.a, cms.Task(), process.t1)\n')

def Config.TestModuleCommand.testPrefers

(

self

)

Definition at line 2529 of file Config.py.

        def testPrefers(self):
            p = Process("Test")
            p.add_(ESSource("ForceSource"))
            p.juicer = ESProducer("JuicerProducer")
            p.prefer("ForceSource")
            p.prefer("juicer")
            self.assertEqual(p.dumpConfig(),
"""process Test = {
    es_module juicer = JuicerProducer { 
    }
    es_source  = ForceSource { 
    }
    es_prefer  = ForceSource { 
    }
    es_prefer juicer = JuicerProducer { 
    }
}
""")
            p.prefer("juicer",fooRcd=vstring("Foo"))
            self.assertEqual(p.dumpConfig(),
"""process Test = {
    es_module juicer = JuicerProducer { 
    }
    es_source  = ForceSource { 
    }
    es_prefer  = ForceSource { 
    }
    es_prefer juicer = JuicerProducer { 
        vstring fooRcd = {
            'Foo'
        }

    }
}
""")
            self.assertEqual(p.dumpPython(),
"""import FWCore.ParameterSet.Config as cms

process = cms.Process("Test")

process.juicer = cms.ESProducer("JuicerProducer")


process.ForceSource = cms.ESSource("ForceSource")


process.prefer("ForceSource")

process.prefer("juicer",
    fooRcd = cms.vstring('Foo')
)

""")

def Config.TestModuleCommand.testProcessDumpPython

(

self

)

Definition at line 1693 of file Config.py.

        def testProcessDumpPython(self):
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.p = Path(p.a)
            p.s = Sequence(p.a)
            p.r = Sequence(p.s)
            p.p2 = Path(p.s)
            p.schedule = Schedule(p.p2,p.p)
            d=p.dumpPython()
            self.assertEqual(d,
"""import FWCore.ParameterSet.Config as cms

process = cms.Process("test")

process.a = cms.EDAnalyzer("MyAnalyzer")


process.s = cms.Sequence(process.a)


process.r = cms.Sequence(process.s)


process.p = cms.Path(process.a)


process.p2 = cms.Path(process.s)


process.schedule = cms.Schedule(*[ process.p2, process.p ])
""")
            #Reverse order of 'r' and 's'
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.p = Path(p.a)
            p.r = Sequence(p.a)
            p.s = Sequence(p.r)
            p.p2 = Path(p.r)
            p.schedule = Schedule(p.p2,p.p)
            p.b = EDAnalyzer("YourAnalyzer")
            d=p.dumpPython()
            self.assertEqual(d,
"""import FWCore.ParameterSet.Config as cms

process = cms.Process("test")

process.a = cms.EDAnalyzer("MyAnalyzer")


process.b = cms.EDAnalyzer("YourAnalyzer")


process.r = cms.Sequence(process.a)


process.s = cms.Sequence(process.r)


process.p = cms.Path(process.a)


process.p2 = cms.Path(process.r)


process.schedule = cms.Schedule(*[ process.p2, process.p ])
""")

def Config.TestModuleCommand.testProcessExtend

(

self

)

Definition at line 1623 of file Config.py.

References resolutioncreator_cfi.object, and str.

        def testProcessExtend(self):
            class FromArg(object):
                def __init__(self,*arg,**args):
                    for name in args.iterkeys():
                        self.__dict__[name]=args[name]

            a=EDAnalyzer("MyAnalyzer")
            t=EDAnalyzer("MyAnalyzer")
            t.setLabel("foo")
            s1 = Sequence(a)
            s2 = Sequence(s1)
            s3 = Sequence(s2)
            d = FromArg(
                    a=a,
                    b=Service("Full"),
                    c=Path(a),
                    d=s2,
                    e=s1,
                    f=s3,
                    g=Sequence(s1+s2+s3)
                )
            p = Process("Test")
            p.extend(d)
            self.assertEqual(p.a.type_(),"MyAnalyzer")
            self.assertEqual(p.a.label_(),"a")
            self.assertRaises(AttributeError,getattr,p,'b')
            self.assertEqual(p.Full.type_(),"Full")
            self.assertEqual(str(p.c),'a')
            self.assertEqual(str(p.d),'a')

            z1 = FromArg(
                    a=a,
                    b=Service("Full"),
                    c=Path(a),
                    d=s2,
                    e=s1,
                    f=s3,
                    s4=s3,
                    g=Sequence(s1+s2+s3)
                 )

            p1 = Process("Test")
            #p1.extend(z1)
            self.assertRaises(ValueError, p1.extend, z1)

            z2 = FromArg(
                    a=a,
                    b=Service("Full"),
                    c=Path(a),
                    d=s2,
                    e=s1,
                    f=s3,
                    aaa=copy.deepcopy(a),
                    s4=copy.deepcopy(s3),
                    g=Sequence(s1+s2+s3),
                    t=t
                )
            p2 = Process("Test")
            p2.extend(z2)
            #self.assertRaises(ValueError, p2.extend, z2)
            self.assertEqual(p2.s4.label_(),"s4")
            #p2.s4.setLabel("foo")
            self.assertRaises(ValueError, p2.s4.setLabel, "foo")
            p2.s4.setLabel("s4")
            p2.s4.setLabel(None)
            p2.s4.setLabel("foo")
            p2._Process__setObjectLabel(p2.s4, "foo")
            p2._Process__setObjectLabel(p2.s4, None)
            p2._Process__setObjectLabel(p2.s4, "bar")

def Config.TestModuleCommand.testProcessInsertion

(

self

)

Definition at line 1598 of file Config.py.

        def testProcessInsertion(self):
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            self.assert_( 'a' in p.analyzers_() )
            self.assert_( 'a' in p.analyzers)
            p.add_(Service("MessageLogger"))
            self.assert_('MessageLogger' in p.services_())
            self.assertEqual(p.MessageLogger.type_(), "MessageLogger")
            p.Tracer = Service("Tracer")
            self.assert_('Tracer' in p.services_())
            self.assertRaises(TypeError, setattr, *(p,'b',"this should fail"))
            self.assertRaises(TypeError, setattr, *(p,'bad',Service("MessageLogger")))
            self.assertRaises(ValueError, setattr, *(p,'bad',Source("PoolSource")))
            p.out = OutputModule("Outer")
            self.assertEqual(p.out.type_(), 'Outer')
            self.assert_( 'out' in p.outputModules_() )

            p.geom = ESSource("GeomProd")
            self.assert_('geom' in p.es_sources_())
            p.add_(ESSource("ConfigDB"))
            self.assert_('ConfigDB' in p.es_sources_())

            p.aliasfoo1 = EDAlias(foo1 = VPSet(PSet(type = string("Foo1"))))
            self.assert_('aliasfoo1' in p.aliases_())

def Config.TestModuleCommand.testPrune

(

self

)

Definition at line 2657 of file Config.py.

        def testPrune(self):
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.d = EDAnalyzer("OurAnalyzer")
            p.s = Sequence(p.d)
            p.path1 = Path(p.a)
            p.path2 = Path(p.b)
            self.assert_(p.schedule is None)
            pths = p.paths
            keys = pths.keys()
            self.assertEqual(pths[keys[0]],p.path1)
            self.assertEqual(pths[keys[1]],p.path2)
            p.pset1 = PSet(parA = string("pset1"))
            p.pset2 = untracked.PSet(parA = string("pset2"))
            p.vpset1 = VPSet()
            p.vpset2 = untracked.VPSet()
            p.prune()
            self.assert_(hasattr(p, 'a'))
            self.assert_(hasattr(p, 'b'))
            self.assert_(not hasattr(p, 'c'))
            self.assert_(not hasattr(p, 'd'))
            self.assert_(not hasattr(p, 's'))
            self.assert_(hasattr(p, 'path1'))
            self.assert_(hasattr(p, 'path2'))
#            self.assert_(not hasattr(p, 'pset1'))
#            self.assert_(hasattr(p, 'pset2'))
#            self.assert_(not hasattr(p, 'vpset1'))
#            self.assert_(not hasattr(p, 'vpset2'))

            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.d = EDAnalyzer("OurAnalyzer")
            p.e = EDAnalyzer("OurAnalyzer")
            p.s = Sequence(p.d)
            p.s2 = Sequence(p.b)
            p.s3 = Sequence(p.e)
            p.path1 = Path(p.a)
            p.path2 = Path(p.b)
            p.path3 = Path(p.b+p.s2)
            p.path4 = Path(p.b+p.s3)
            p.schedule = Schedule(p.path1,p.path2,p.path3)
            pths = p.paths
            keys = pths.keys()
            self.assertEqual(pths[keys[0]],p.path1)
            self.assertEqual(pths[keys[1]],p.path2)
            p.prune()
            self.assert_(hasattr(p, 'a'))
            self.assert_(hasattr(p, 'b'))
            self.assert_(not hasattr(p, 'c'))
            self.assert_(not hasattr(p, 'd'))
            self.assert_(not hasattr(p, 'e'))
            self.assert_(not hasattr(p, 's'))
            self.assert_(hasattr(p, 's2'))
            self.assert_(not hasattr(p, 's3'))
            self.assert_(hasattr(p, 'path1'))
            self.assert_(hasattr(p, 'path2'))
            self.assert_(hasattr(p, 'path3'))
            self.assert_(not hasattr(p, 'path4'))
            #test SequencePlaceholder
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.s = Sequence(SequencePlaceholder("a")+p.b)
            p.pth = Path(p.s)
            p.prune()
            self.assert_(hasattr(p, 'a'))
            self.assert_(hasattr(p, 'b'))
            self.assert_(hasattr(p, 's'))
            self.assert_(hasattr(p, 'pth'))
            #test unresolved SequencePlaceholder
            p = Process("test")
            p.b = EDAnalyzer("YourAnalyzer")
            p.s = Sequence(SequencePlaceholder("a")+p.b)
            p.pth = Path(p.s)
            p.prune(keepUnresolvedSequencePlaceholders=True)
            self.assert_(hasattr(p, 'b'))
            self.assert_(hasattr(p, 's'))
            self.assert_(hasattr(p, 'pth'))
            self.assertEqual(p.s.dumpPython(''),'cms.Sequence(cms.SequencePlaceholder("a")+process.b)\n')

def Config.TestModuleCommand.testRefToPSet

(

self

)

Definition at line 2641 of file Config.py.

        def testRefToPSet(self):
            proc = Process("test")
            proc.top = PSet(a = int32(1))
            proc.ref = PSet(refToPSet_ = string("top"))
            proc.ref2 = PSet( a = int32(1), b = PSet( refToPSet_ = string("top")))
            proc.ref3 = PSet(refToPSet_ = string("ref"))
            proc.ref4 = VPSet(PSet(refToPSet_ = string("top")),
                              PSet(refToPSet_ = string("ref2")))
            p = TestMakePSet()
            proc.fillProcessDesc(p)
            self.assertEqual((True,1),p.values["ref"][1].values["a"])
            self.assertEqual((True,1),p.values["ref3"][1].values["a"])
            self.assertEqual((True,1),p.values["ref2"][1].values["a"])
            self.assertEqual((True,1),p.values["ref2"][1].values["b"][1].values["a"])
            self.assertEqual((True,1),p.values["ref4"][1][0].values["a"])
            self.assertEqual((True,1),p.values["ref4"][1][1].values["a"])

def Config.TestModuleCommand.testSchedule

(

self

)

Definition at line 2349 of file Config.py.

References list().

        def testSchedule(self):
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.d = EDAnalyzer("OurAnalyzer")
            p.path1 = Path(p.a)
            p.path2 = Path(p.b)
            p.path3 = Path(p.d)

            s = Schedule(p.path1,p.path2)
            self.assertEqual(s[0],p.path1)
            self.assertEqual(s[1],p.path2)
            p.schedule = s
            self.assert_('b' in p.schedule.moduleNames())
            self.assert_(hasattr(p, 'b'))
            self.assert_(hasattr(p, 'c'))
            self.assert_(hasattr(p, 'd'))
            self.assert_(hasattr(p, 'path1'))
            self.assert_(hasattr(p, 'path2'))
            self.assert_(hasattr(p, 'path3'))
            p.prune()
            self.assert_('b' in p.schedule.moduleNames())
            self.assert_(hasattr(p, 'b'))
            self.assert_(not hasattr(p, 'c'))
            self.assert_(not hasattr(p, 'd'))
            self.assert_(hasattr(p, 'path1'))
            self.assert_(hasattr(p, 'path2'))
            self.assert_(not hasattr(p, 'path3'))

            self.assertTrue(len(p.schedule._tasks) == 0)

            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.d = EDAnalyzer("dAnalyzer")
            p.e = EDProducer("eProducer")
            p.f = EDProducer("fProducer")
            p.Tracer = Service("Tracer")
            p.path1 = Path(p.a)
            p.path2 = Path(p.b)
            p.path3 = Path(p.d)
            p.task1 = Task(p.e)
            p.task2 = Task(p.f, p.Tracer)
            s = Schedule(p.path1,p.path2,tasks=[p.task1,p.task2,p.task1])
            self.assertEqual(s[0],p.path1)
            self.assertEqual(s[1],p.path2)
            self.assertTrue(len(s._tasks) == 2)
            self.assertTrue(p.task1 in s._tasks)
            self.assertTrue(p.task2 in s._tasks)
            listOfTasks = list(s._tasks)
            self.assertTrue(len(listOfTasks) == 2)
            self.assertTrue(p.task1 == listOfTasks[0])
            self.assertTrue(p.task2 == listOfTasks[1])
            p.schedule = s
            self.assert_('b' in p.schedule.moduleNames())

            process2 = Process("test")
            process2.a = EDAnalyzer("MyAnalyzer")
            process2.e = EDProducer("eProducer")
            process2.path1 = Path(process2.a)
            process2.task1 = Task(process2.e)
            process2.schedule = Schedule(process2.path1,tasks=process2.task1)
            listOfTasks = list(process2.schedule._tasks)
            self.assertTrue(listOfTasks[0] == process2.task1)

            # test Schedule copy
            s2 = s.copy()
            self.assertEqual(s2[0],p.path1)
            self.assertEqual(s2[1],p.path2)
            self.assertTrue(len(s2._tasks) == 2)
            self.assertTrue(p.task1 in s2._tasks)
            self.assertTrue(p.task2 in s2._tasks)
            listOfTasks = list(s2._tasks)
            self.assertTrue(len(listOfTasks) == 2)
            self.assertTrue(p.task1 == listOfTasks[0])
            self.assertTrue(p.task2 == listOfTasks[1])

            names = s.moduleNames()
            self.assertTrue(names == set(['a', 'b', 'e', 'Tracer', 'f']))
            #adding a path not attached to the Process should cause an exception
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            path1 = Path(p.a)
            s = Schedule(path1)
            self.assertRaises(RuntimeError, lambda : p.setSchedule_(s) )

            #make sure anonymous sequences work
            p = Process("test")
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("MyOtherAnalyzer")
            p.c = EDProducer("MyProd")
            path1 = Path(p.c*Sequence(p.a+p.b))
            s = Schedule(path1)
            self.assert_('a' in s.moduleNames())
            self.assert_('b' in s.moduleNames())
            self.assert_('c' in s.moduleNames())
            p.path1 = path1
            p.schedule = s
            p.prune()
            self.assert_('a' in s.moduleNames())
            self.assert_('b' in s.moduleNames())
            self.assert_('c' in s.moduleNames())

def Config.TestModuleCommand.testSecSource

(

self

)

Definition at line 1948 of file Config.py.

References python.rootplot.root2matplotlib.replace().

        def testSecSource(self):
            p = Process('test')
            p.a = SecSource("MySecSource")
            self.assertEqual(p.dumpPython().replace('\n',''),'import FWCore.ParameterSet.Config as cmsprocess = cms.Process("test")process.a = cms.SecSource("MySecSource")')

def Config.TestModuleCommand.testSequence

(

self

)

Definition at line 1998 of file Config.py.

References str.

        def testSequence(self):
            p = Process('test')
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            p.s = Sequence(p.a*p.b)
            self.assertEqual(str(p.s),'a+b')
            self.assertEqual(p.s.label_(),'s')
            path = Path(p.c+p.s)
            self.assertEqual(str(path),'c+a+b')
            p._validateSequence(path, 'p1')
            notInProcess = EDAnalyzer('NotInProcess')
            p2 = Path(p.c+p.s*notInProcess)
            self.assertRaises(RuntimeError, p._validateSequence, p2, 'p2')

def Config.TestModuleCommand.testSequence2

(

self

)

Definition at line 2013 of file Config.py.

        def testSequence2(self):
            p = Process('test')
            p.a = EDAnalyzer("MyAnalyzer")
            p.b = EDAnalyzer("YourAnalyzer")
            p.c = EDAnalyzer("OurAnalyzer")
            testseq = Sequence(p.a*p.b)
            p.s = testseq
            #p.y = testseq
            self.assertRaises(ValueError, p.__setattr__, "y", testseq) 

def Config.TestModuleCommand.testServiceInProcess

(

self

)

Definition at line 2023 of file Config.py.

        def testServiceInProcess(self):
            service = Service("d")
            self.assertFalse(service._inProcess)
            process = Process("test")
            process.d = service
            self.assertTrue(service._inProcess)
            service2 = Service("d")
            process.d = service2
            self.assertFalse(service._inProcess)
            self.assertTrue(service2._inProcess)
            del process.d
            self.assertFalse(service2._inProcess)

def Config.TestModuleCommand.testSubProcess

(

self

)

Definition at line 2601 of file Config.py.

References cond.hash, and str.

        def testSubProcess(self):
            process = Process("Parent")
            subProcess = Process("Child")
            subProcess.a = EDProducer("A")
            subProcess.p = Path(subProcess.a)
            subProcess.add_(Service("Foo"))
            process.addSubProcess(SubProcess(subProcess))
            d = process.dumpPython()
            equalD ="""import FWCore.ParameterSet.Config as cms

process = cms.Process("Parent")

parentProcess = process
import FWCore.ParameterSet.Config as cms

process = cms.Process("Child")

process.a = cms.EDProducer("A")


process.Foo = cms.Service("Foo")


process.p = cms.Path(process.a)


childProcess = process
process = parentProcess
process.addSubProcess(cms.SubProcess(process = childProcess, SelectEvents = cms.untracked.PSet(

), outputCommands = cms.untracked.vstring()))

"""
            equalD = equalD.replace("parentProcess","parentProcess"+str(hash(process.subProcesses_()[0])))
            self.assertEqual(d,equalD)
            p = TestMakePSet()
            process.fillProcessDesc(p)
            self.assertEqual((True,['a']),p.values["subProcesses"][1][0].values["process"][1].values['@all_modules'])
            self.assertEqual((True,['p']),p.values["subProcesses"][1][0].values["process"][1].values['@paths'])
            self.assertEqual({'@service_type':(True,'Foo')}, p.values["subProcesses"][1][0].values["process"][1].values["services"][1][0].values)

def Config.TestModuleCommand.testTask

(

self

)

Definition at line 2036 of file Config.py.

References list().

        def testTask(self):

            # create some objects to use in tests
            edanalyzer = EDAnalyzer("a")
            edproducer = EDProducer("b")
            edproducer2 = EDProducer("b2")
            edproducer3 = EDProducer("b3")
            edproducer4 = EDProducer("b4")
            edproducer8 = EDProducer("b8")
            edproducer9 = EDProducer("b9")
            edfilter = EDFilter("c")
            service = Service("d")
            service3 = Service("d")
            essource = ESSource("e")
            esproducer = ESProducer("f")
            testTask2 = Task()

            # test adding things to Tasks
            testTask1 = Task(edproducer, edfilter)
            self.assertRaises(RuntimeError, testTask1.add, edanalyzer)
            testTask1.add(essource, service)
            testTask1.add(essource, esproducer)
            testTask1.add(testTask2)
            coll = testTask1._collection
            self.assertTrue(edproducer in coll)
            self.assertTrue(edfilter in coll)
            self.assertTrue(service in coll)
            self.assertTrue(essource in coll)
            self.assertTrue(esproducer in coll)
            self.assertTrue(testTask2 in coll)
            self.assertTrue(len(coll) == 6)
            self.assertTrue(len(testTask2._collection) == 0)

            taskContents = []
            for i in testTask1:
                taskContents.append(i)
            self.assertTrue(taskContents == [edproducer, edfilter, essource, service, esproducer, testTask2])

            # test attaching Task to Process
            process = Process("test")

            process.mproducer = edproducer
            process.mproducer2 = edproducer2
            process.mfilter = edfilter
            process.messource = essource
            process.mesproducer = esproducer
            process.d = service

            testTask3 = Task(edproducer, edproducer2)
            testTask1.add(testTask3)
            process.myTask1 = testTask1

            # test the validation that occurs when attaching a Task to a Process
            # first a case that passes, then one the fails on an EDProducer
            # then one that fails on a service
            l = set()
            visitor = NodeNameVisitor(l)
            testTask1.visit(visitor)
            self.assertTrue(l == set(['mesproducer', 'mproducer', 'mproducer2', 'mfilter', 'd', 'messource']))
            l2 = testTask1.moduleNames
            self.assertTrue(l == set(['mesproducer', 'mproducer', 'mproducer2', 'mfilter', 'd', 'messource']))

            testTask4 = Task(edproducer3)
            l.clear()
            self.assertRaises(RuntimeError, testTask4.visit, visitor)
            try:
                process.myTask4 = testTask4
                self.assertTrue(False)
            except RuntimeError:
                pass

            testTask5 = Task(service3)
            l.clear()
            self.assertRaises(RuntimeError, testTask5.visit, visitor)
            try:
                process.myTask5 = testTask5
                self.assertTrue(False)
            except RuntimeError:
                pass

            process.d = service3
            process.myTask5 = testTask5

            # test placement into the Process and the tasks property
            expectedDict = { 'myTask1' : testTask1, 'myTask5' : testTask5 }
            expectedFixedDict = DictTypes.FixedKeysDict(expectedDict);
            self.assertTrue(process.tasks == expectedFixedDict)
            self.assertTrue(process.tasks['myTask1'] == testTask1)
            self.assertTrue(process.myTask1 == testTask1)

            # test replacing an EDProducer in a Task when calling __settattr__
            # for the EDProducer on the Process.
            process.mproducer2 = edproducer4
            process.d = service
            l = list()
            visitor1 = ModuleNodeVisitor(l)
            testTask1.visit(visitor1)
            l.sort()
            expectedList = sorted([edproducer,essource,esproducer,service,edfilter,edproducer,edproducer4])
            self.assertTrue(expectedList == l)

            process.myTask6 = Task()
            process.myTask7 = Task()
            process.mproducer8 = edproducer8
            process.myTask8 = Task(process.mproducer8)
            process.myTask6.add(process.myTask7)
            process.myTask7.add(process.myTask8)
            process.myTask1.add(process.myTask6)
            process.myTask8.add(process.myTask5)

            testDict = process._itemsInDependencyOrder(process.tasks)
            expectedLabels = ["myTask5", "myTask8", "myTask7", "myTask6", "myTask1"]
            expectedTasks = [process.myTask5, process.myTask8, process.myTask7, process.myTask6, process.myTask1]
            index = 0
            for testLabel, testTask in testDict.items():
                self.assertTrue(testLabel == expectedLabels[index])
                self.assertTrue(testTask == expectedTasks[index])
                index += 1

            pythonDump = testTask1.dumpPython(PrintOptions())


            expectedPythonDump = 'cms.Task(process.d, process.mesproducer, process.messource, process.mfilter, process.mproducer, process.mproducer2, process.myTask6)\n'
            self.assertTrue(pythonDump == expectedPythonDump)

            process.myTask5 = Task()
            process.myTask100 = Task()
            process.mproducer9 = edproducer9
            sequence1 = Sequence(process.mproducer8, process.myTask1, process.myTask5, testTask2, testTask3)
            sequence2 = Sequence(process.mproducer8 + process.mproducer9)
            process.sequence3 = Sequence((process.mproducer8 + process.mfilter))
            sequence4 = Sequence()
            process.path1 = Path(process.mproducer+process.mproducer8+sequence1+sequence2+process.sequence3+sequence4)
            process.path1.associate(process.myTask1, process.myTask5, testTask2, testTask3)
            process.path11 = Path(process.mproducer+process.mproducer8+sequence1+sequence2+process.sequence3+ sequence4,process.myTask1, process.myTask5, testTask2, testTask3, process.myTask100)
            process.path2 = Path(process.mproducer)
            process.path3 = Path(process.mproducer9+process.mproducer8,testTask2)

            self.assertTrue(process.path1.dumpPython(PrintOptions()) == 'cms.Path(process.mproducer+process.mproducer8+cms.Sequence(process.mproducer8, cms.Task(), cms.Task(process.None, process.mproducer), process.myTask1, process.myTask5)+(process.mproducer8+process.mproducer9)+process.sequence3, cms.Task(), cms.Task(process.None, process.mproducer), process.myTask1, process.myTask5)\n')

            self.assertTrue(process.path11.dumpPython(PrintOptions()) == 'cms.Path(process.mproducer+process.mproducer8+cms.Sequence(process.mproducer8, cms.Task(), cms.Task(process.None, process.mproducer), process.myTask1, process.myTask5)+(process.mproducer8+process.mproducer9)+process.sequence3, cms.Task(), cms.Task(process.None, process.mproducer), process.myTask1, process.myTask100, process.myTask5)\n')

            # test NodeNameVisitor and moduleNames
            l = set()
            nameVisitor = NodeNameVisitor(l)
            process.path1.visit(nameVisitor)
            self.assertTrue(l == set(['mproducer', 'd', 'mesproducer', None, 'mproducer9', 'mproducer8', 'messource', 'mproducer2', 'mfilter']))
            self.assertTrue(process.path1.moduleNames() == set(['mproducer', 'd', 'mesproducer', None, 'mproducer9', 'mproducer8', 'messource', 'mproducer2', 'mfilter']))

            # test copy
            process.mproducer10 = EDProducer("b10")
            process.path21 = process.path11.copy()
            process.path21.replace(process.mproducer, process.mproducer10)

            self.assertTrue(process.path11.dumpPython(PrintOptions()) == 'cms.Path(process.mproducer+process.mproducer8+cms.Sequence(process.mproducer8, cms.Task(), cms.Task(process.None, process.mproducer), process.myTask1, process.myTask5)+(process.mproducer8+process.mproducer9)+process.sequence3, cms.Task(), cms.Task(process.None, process.mproducer), process.myTask1, process.myTask100, process.myTask5)\n')

            # Some peculiarities of the way things work show up here. dumpPython sorts tasks and
            # removes duplication at the level of strings. The Task and Sequence objects themselves
            # remove duplicate tasks in their contents if the instances are the same (exact same python
            # object id which is not the same as the string representation being the same).
            # Also note that the mutating visitor replaces sequences and tasks that have
            # modified contents with their modified contents, it does not modify the sequence
            # or task itself.
            self.assertTrue(process.path21.dumpPython(PrintOptions()) == 'cms.Path(process.mproducer10+process.mproducer8+process.mproducer8+(process.mproducer8+process.mproducer9)+process.sequence3, cms.Task(), cms.Task(process.None, process.mproducer10), cms.Task(process.d, process.mesproducer, process.messource, process.mfilter, process.mproducer10, process.mproducer2, process.myTask6), process.myTask100, process.myTask5)\n')

            process.path22 = process.path21.copyAndExclude([process.d, process.mesproducer, process.mfilter])
            self.assertTrue(process.path22.dumpPython(PrintOptions()) == 'cms.Path(process.mproducer10+process.mproducer8+process.mproducer8+(process.mproducer8+process.mproducer9)+process.mproducer8, cms.Task(), cms.Task(process.None, process.mproducer10), cms.Task(process.messource, process.mproducer10, process.mproducer2, process.myTask6), process.myTask100, process.myTask5)\n')

            process.path23 = process.path22.copyAndExclude([process.messource, process.mproducer10])
            self.assertTrue(process.path23.dumpPython(PrintOptions()) == 'cms.Path(process.mproducer8+process.mproducer8+(process.mproducer8+process.mproducer9)+process.mproducer8, cms.Task(), cms.Task(process.None), cms.Task(process.mproducer2, process.myTask6), process.myTask100, process.myTask5)\n')

            process.a = EDAnalyzer("MyAnalyzer")
            process.b = OutputModule("MyOutputModule")
            process.c = EDFilter("MyFilter")
            process.d = EDProducer("MyProducer")
            process.e = ESProducer("MyESProducer")
            process.f = ESSource("MyESSource")
            process.g = ESProducer("g")
            process.path24 = Path(process.a+process.b+process.c+process.d)
            process.path25 = process.path24.copyAndExclude([process.a,process.b,process.c])
            self.assertTrue(process.path25.dumpPython(None) == 'cms.Path(process.d)\n')
            #print process.path3
            #print process.dumpPython()

            process.path200 = EndPath(Sequence(process.c,Task(process.e)))
            process.path200.replace(process.c,process.b)
            process.path200.replace(process.e,process.f)
            self.assertEqual(process.path200.dumpPython(None), "cms.EndPath(process.b, cms.Task(process.f))\n")
            process.path200.replace(process.b,process.c)
            process.path200.replace(process.f,process.e)
            self.assertEqual(process.path200.dumpPython(None), "cms.EndPath(process.c, cms.Task(process.e))\n")
            process.path200.replace(process.c,process.a)
            process.path200.replace(process.e,process.g)
            self.assertEqual(process.path200.dumpPython(None), "cms.EndPath(process.a, cms.Task(process.g))\n")
            process.path200.replace(process.a,process.c)
            process.path200.replace(process.g,process.e)
            self.assertEqual(process.path200.dumpPython(None), "cms.EndPath(process.c, cms.Task(process.e))\n")

def Config.TestModuleCommand.testTaskPlaceholder

(

self

)

Definition at line 2740 of file Config.py.

        def testTaskPlaceholder(self):
            p = Process("test")
            p.a = EDProducer("ma")
            p.b = EDAnalyzer("mb")
            p.t1 = Task(TaskPlaceholder("c"))
            p.t2 = Task(p.a, TaskPlaceholder("d"), p.t1)
            p.t3 = Task(TaskPlaceholder("e"))
            p.path1 = Path(p.b, p.t2, p.t3)
            p.t5 = Task(p.a, TaskPlaceholder("g"), TaskPlaceholder("t4"))
            p.t4 = Task(TaskPlaceholder("f"))
            p.endpath1 = EndPath(p.b, p.t5)
            p.t6 = Task(TaskPlaceholder("h"))
            p.t7 = Task(p.a, TaskPlaceholder("i"), p.t6)
            p.t8 = Task(TaskPlaceholder("j"))
            p.schedule = Schedule(p.path1, p.endpath1,tasks=[p.t7,p.t8])
            p.c = EDProducer("mc")
            p.d = EDProducer("md")
            p.e = EDProducer("me")
            p.f = EDProducer("mf")
            p.g = EDProducer("mg")
            p.h = EDProducer("mh")
            p.i = EDProducer("mi")
            p.j = EDProducer("mj")
            self.assertEqual(p.dumpPython(),
"""import FWCore.ParameterSet.Config as cms

process = cms.Process("test")

process.a = cms.EDProducer("ma")


process.c = cms.EDProducer("mc")


process.d = cms.EDProducer("md")


process.e = cms.EDProducer("me")


process.f = cms.EDProducer("mf")


process.g = cms.EDProducer("mg")


process.h = cms.EDProducer("mh")


process.i = cms.EDProducer("mi")


process.j = cms.EDProducer("mj")


process.b = cms.EDAnalyzer("mb")


process.t8 = cms.Task(cms.TaskPlaceholder("j"))


process.t6 = cms.Task(cms.TaskPlaceholder("h"))


process.t7 = cms.Task(cms.TaskPlaceholder("i"), process.a, process.t6)


process.t4 = cms.Task(cms.TaskPlaceholder("f"))


process.t5 = cms.Task(cms.TaskPlaceholder("g"), cms.TaskPlaceholder("t4"), process.a)


process.t3 = cms.Task(cms.TaskPlaceholder("e"))


process.t1 = cms.Task(cms.TaskPlaceholder("c"))


process.t2 = cms.Task(cms.TaskPlaceholder("d"), process.a, process.t1)


process.path1 = cms.Path(process.b, process.t2, process.t3)


process.endpath1 = cms.EndPath(process.b, process.t5)


process.schedule = cms.Schedule(*[ process.path1, process.endpath1 ], tasks=[process.t7, process.t8])
""")
            p.resolve()
            self.assertEqual(p.dumpPython(),
"""import FWCore.ParameterSet.Config as cms

process = cms.Process("test")

process.a = cms.EDProducer("ma")


process.c = cms.EDProducer("mc")


process.d = cms.EDProducer("md")


process.e = cms.EDProducer("me")


process.f = cms.EDProducer("mf")


process.g = cms.EDProducer("mg")


process.h = cms.EDProducer("mh")


process.i = cms.EDProducer("mi")


process.j = cms.EDProducer("mj")


process.b = cms.EDAnalyzer("mb")


process.t8 = cms.Task(process.j)


process.t6 = cms.Task(process.h)


process.t7 = cms.Task(process.a, process.i, process.t6)


process.t4 = cms.Task(process.f)


process.t5 = cms.Task(process.a, process.g, process.t4)


process.t3 = cms.Task(process.e)


process.t1 = cms.Task(process.c)


process.t2 = cms.Task(process.a, process.d, process.t1)


process.path1 = cms.Path(process.b, process.t2, process.t3)


process.endpath1 = cms.EndPath(process.b, process.t5)


process.schedule = cms.Schedule(*[ process.path1, process.endpath1 ], tasks=[process.t7, process.t8])
""")

def Config.TestModuleCommand.testTypedParameterizable

(

self

)

Definition at line 1591 of file Config.py.

        def testTypedParameterizable(self):
            p = _TypedParameterizable("blah", b=int32(1))
            #see if copy works deeply
            other = p.copy()
            other.b = 2
            self.assertNotEqual(p.b,other.b)

def Config.TestModuleCommand.testUsing

(

self

)

Definition at line 2487 of file Config.py.

        def testUsing(self):
            p = Process('test')
            p.block = PSet(a = int32(1))
            p.modu = EDAnalyzer('Analyzer', p.block, b = int32(2))
            self.assertEqual(p.modu.a.value(),1)
            self.assertEqual(p.modu.b.value(),2)

Member Data Documentation

Config.TestModuleCommand.a

Definition at line 3029 of file Config.py.

Config.TestModuleCommand.proc_mod_

Definition at line 3026 of file Config.py.