-
Notifications
You must be signed in to change notification settings - Fork 106
/
EmulatedUnitTestCase.py
126 lines (110 loc) · 5.98 KB
/
EmulatedUnitTestCase.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
#! /usr/bin/env python
"""
Unit testing base class that turns on emulators
"""
from __future__ import (division, print_function)
import unittest
import mock
from WMQuality.Emulators.CRICClient.MockCRICApi import MockCRICApi
from WMQuality.Emulators.Cache.MockMemoryCacheStruct import MockMemoryCacheStruct
from WMQuality.Emulators.DBSClient.MockDbsApi import MockDbsApi
from WMQuality.Emulators.LogDB.MockLogDB import MockLogDB
from WMQuality.Emulators.PyCondorAPI.MockPyCondorAPI import MockPyCondorAPI
from WMQuality.Emulators.ReqMgrAux.MockReqMgrAux import MockReqMgrAux
from WMQuality.Emulators.RucioClient.MockRucioApi import MockRucioApi
from WMQuality.Emulators.MSPileup.MockMSPileupAPI import getPileupDocs as mockPileupDocs
class EmulatedUnitTestCase(unittest.TestCase):
"""
Class that can be imported to switch to 'mock'ed versions of
services.
"""
def __init__(self, methodName='runTest', mockDBS=True,
mockReqMgrAux=True, mockLogDB=True,
mockMemoryCache=True, mockPyCondor=True,
mockCRIC=True, mockRucio=True, mockMSPileup=True):
self.mockDBS = mockDBS
self.mockReqMgrAux = mockReqMgrAux
self.mockLogDB = mockLogDB
self.mockMemoryCache = mockMemoryCache
self.mockPyCondor = mockPyCondor
self.mockCRIC = mockCRIC
self.mockRucio = mockRucio
self.mockMSPileup = mockMSPileup
super().__init__(methodName)
def setUp(self):
"""
Start the various mocked versions and add cleanups in case of exceptions
Note: patch has to be applied to where the object is loaded/imported, not
to its origin location.
TODO: parameters to turn off emulators individually
"""
print("Running setUp with emulated services.")
if self.mockDBS:
self.dbsPatchers = []
patchDBSAt = ["dbs.apis.dbsClient.DbsApi",
"WMCore.Services.DBS.DBS3Reader.DbsApi"]
for module in patchDBSAt:
self.dbsPatchers.append(mock.patch(module, new=MockDbsApi))
self.dbsPatchers[-1].start()
self.addCleanup(self.dbsPatchers[-1].stop)
if self.mockRucio:
self.rucioPatchers = []
patchRucioAt = ['WMCore.WorkQueue.WorkQueue.Rucio',
'WMCore.WorkQueue.WorkQueueReqMgrInterface.Rucio',
'WMCore.WorkQueue.Policy.Start.StartPolicyInterface.Rucio',
'WMCore.MicroService.MSCore.MSCore.Rucio',
'WMComponent.RucioInjector.RucioInjectorPoller.Rucio',
'WMComponent.WorkflowUpdater.WorkflowUpdaterPoller.Rucio',
'WMCore.WMSpec.Steps.Fetchers.PileupFetcher.Rucio',
'WMCore_t.WMSpec_t.Steps_t.Fetchers_t.PileupFetcher_t.Rucio',
'WMCore_t.WorkQueue_t.WMBSHelper_t.Rucio']
for module in patchRucioAt:
self.rucioPatchers.append(mock.patch(module, new=MockRucioApi))
self.rucioPatchers[-1].start()
self.addCleanup(self.rucioPatchers[-1].stop)
if self.mockReqMgrAux:
self.reqMgrAuxPatchers = []
patchReqMgrAuxAt = ['WMCore.Services.ReqMgrAux.ReqMgrAux.ReqMgrAux',
'WMComponent.JobSubmitter.JobSubmitterPoller.ReqMgrAux',
'WMComponent.ErrorHandler.ErrorHandlerPoller.ReqMgrAux']
for module in patchReqMgrAuxAt:
self.reqMgrAuxPatchers.append(mock.patch(module, new=MockReqMgrAux))
self.reqMgrAuxPatchers[-1].start()
self.addCleanup(self.reqMgrAuxPatchers[-1].stop)
if self.mockLogDB:
self.logDBPatcher = mock.patch('WMCore.Services.LogDB.LogDB.LogDB',
new=MockLogDB)
self.inUseLogDB = self.logDBPatcher.start()
self.addCleanup(self.logDBPatcher.stop)
if self.mockMemoryCache:
self.memoryCachePatcher = mock.patch('WMCore.Cache.GenericDataCache.MemoryCacheStruct',
new=MockMemoryCacheStruct)
self.inUseMemoryCache = self.memoryCachePatcher.start()
self.addCleanup(self.memoryCachePatcher.stop)
if self.mockPyCondor:
self.condorPatcher = mock.patch('WMCore.Services.PyCondor.PyCondorAPI.PyCondorAPI',
new=MockPyCondorAPI)
self.condorPatcher.start()
self.addCleanup(self.condorPatcher.stop)
if self.mockCRIC:
self.cricPatchers = []
patchCRICAt = ['WMCore.ReqMgr.Tools.cms.CRIC',
'WMCore.WorkQueue.WorkQueue.CRIC',
'WMCore.WorkQueue.WorkQueueUtils.CRIC',
'WMCore.WorkQueue.Policy.Start.StartPolicyInterface.CRIC',
'WMComponent.AgentStatusWatcher.ResourceControlUpdater.CRIC']
for module in patchCRICAt:
self.cricPatchers.append(mock.patch(module, new=MockCRICApi))
self.cricPatchers[-1].start()
self.addCleanup(self.cricPatchers[-1].stop)
if self.mockMSPileup:
self.msPileupPatchers = []
patchMSPileupAt = ['WMCore.MicroService.MSTransferor.MSTransferor.getPileupDocs',
'WMCore.WorkQueue.Policy.Start.StartPolicyInterface.getPileupDocs',
'WMCore.WorkQueue.DataLocationMapper.getPileupDocs',
'WMComponent.WorkflowUpdater.WorkflowUpdaterPoller.getPileupDocs',
'WMCore_t.Services_t.MSPileup_t.MSPileupUtils_t.getPileupDocs']
for module in patchMSPileupAt:
self.msPileupPatchers.append(mock.patch(module, new=mockPileupDocs))
self.msPileupPatchers[-1].start()
self.addCleanup(self.msPileupPatchers[-1].stop)