Coverage for Applications / PyCharm.app / Contents / plugins / python-ce / helpers / pycharm / teamcity / output.py: 81%

1# coding=utf-8 

2import errno 

3import sys 

4from functools import wraps 

5import time 

6 

7# Capture some time functions to allow monkeypatching them in tests 

8_time = time.time 

9_stderr = sys.stderr 

10 

11 

12class TeamCityMessagesPrinter(object): 

13 def __init__(self, output=None, context_manager=None): 

14 if output is None: 

15 output = sys.stdout 

16 if sys.version_info < (3,) or not hasattr(output, 'buffer'): 

17 self.output = output 

18 else: 

19 self.output = output.buffer 

20 # context manager can be set for testing frameworks such as pytest, which may redirect 

21 # normal output, to temporarily disable redirection for Teamcity messages 

22 self.context_manager = context_manager 

23 

24 def send_message(self, message): 

25 if self.context_manager: 

26 with self.context_manager(): 

27 self._output(message) 

28 else: 

29 self._output(message) 

30 

31 def _output(self, message): 

32 # Python may buffer it for a long time, flushing helps to see real-time result 

33 retry_on_EAGAIN(self.output.write)(message) 

34 retry_on_EAGAIN(self.output.flush)() 

35 

36 

37def retry_on_EAGAIN(callable): 

38 # output seems to be non-blocking when running under teamcity. 

39 @wraps(callable) 

40 def wrapped(*args, **kwargs): 

41 start_time = _time() 

42 while True: 

43 try: 

44 return callable(*args, **kwargs) 

45 except IOError as e: 

46 if e.errno != errno.EAGAIN: 

47 raise 

48 # Give up after a minute. 

49 if _time() - start_time > 60: 

50 raise 

51 time.sleep(.1) 

52 

53 return wrapped