Coverage for gwcelery/sentry/integrations/condor.py: 40%

1import os 

2 

3from sentry_sdk.api import configure_scope 

4from sentry_sdk.integrations import Integration 

5 

6 

7def _read_classad(filename): 

8 with open(filename) as f: 

9 for line in f: 

10 key, _, value = line.partition('=') 

11 key = key.strip() 

12 value = value.strip().strip('"') 

13 yield key, value 

14 

15 

16class CondorIntegration(Integration): 

17 """Custom Sentry integration to report the HTCondor job ID.""" 

18 

19 identifier = 'condor' 

20 

21 @staticmethod 

22 def setup_once(): 

23 try: 

24 data = dict(_read_classad(os.environ['_CONDOR_JOB_AD'])) 

25 except (KeyError, IOError): 

26 pass 

27 else: 

28 with configure_scope() as scope: 

29 scope.set_tag('htcondor.cluster_id', '{}.{}'.format( 

30 data['ClusterId'], data['ProcId']))