Coverage for libs/sm/ipc.py: 17%

1# 

2# Copyright (C) Citrix Systems Inc. 

3# 

4# This program is free software; you can redistribute it and/or modify 

5# it under the terms of the GNU Lesser General Public License as published 

6# by the Free Software Foundation; version 2.1 only. 

7# 

8# This program is distributed in the hope that it will be useful, 

9# but WITHOUT ANY WARRANTY; without even the implied warranty of 

10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

11# GNU Lesser General Public License for more details. 

12# 

13# You should have received a copy of the GNU Lesser General Public License 

14# along with this program; if not, write to the Free Software Foundation, Inc., 

15# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA 

16 

17"""Communication for processes""" 

18 

19import os 

20from sm.core import util 

21import errno 

22 

23 

24class IPCFlagException(util.SMException): 

25 pass 

26 

27 

28class IPCFlag: 

29 """Flag-based communication for processes (set, test, clear). 

30 Not thread-safe.""" 

31 

32 BASE_DIR = "/run/sm/ipc" 

33 

34 def __init__(self, ns): 

35 self.ns = ns 

36 self.nsDir = os.path.join(self.BASE_DIR, self.ns) 

37 if not util.pathexists(self.nsDir): 

38 try: 

39 os.makedirs(self.nsDir) 

40 except OSError: 

41 pass 

42 if not util.pathexists(self.nsDir): 

43 raise IPCFlagException("failed to create %s" % self.nsDir) 

44 

45 def set(self, name, soft=False): 

46 """Set the flag 

47 

48 name: the file to set 

49 soft: If set to False and the file gets created while this function is 

50 trying to set create, the file MAY be overwritten. 

51 

52 returns: True if the file is written, False otherwise.""" 

53 if not soft and self.test(name): # XXX this is broken! 

54 return 

55 flagFile = os.path.join(self.nsDir, name) 

56 try: 

57 if soft: 

58 f = util.open_atomic(flagFile, "w") 

59 else: 

60 f = open(flagFile, "w") 

61 f.write("%s\n" % os.getpid()) 

62 f.close() 

63 util.SMlog("IPCFlag: set %s:%s" % (self.ns, name)) 

64 return True 

65 except OSError as e: 

66 if e.errno == errno.EEXIST and soft: 

67 return False 

68 else: 

69 raise IPCFlagException("failed to create %s: %s" \ 

70 % (flagFile, e)) 

71 except IOError as e: 

72 raise IPCFlagException("failed to create %s: %s" % (flagFile, e)) 

73 

74 def test(self, name): 

75 """Test the flag""" 

76 flagFile = os.path.join(self.nsDir, name) 

77 return util.pathexists(flagFile) 

78 

79 def clear(self, name): 

80 """Clear the flag""" 

81 if self.test(name): 

82 flagFile = os.path.join(self.nsDir, name) 

83 try: 

84 os.unlink(flagFile) 

85 util.SMlog("IPCFlag: clear %s:%s" % (self.ns, name)) 

86 except OSError: 

87 raise IPCFlagException("failed to remove %s" % flagFile) 

88 

89 def clearAll(self): 

90 try: 

91 for file in os.listdir(self.nsDir): 

92 path = os.path.join(self.nsDir, file) 

93 os.unlink(path) 

94 except OSError: 

95 raise IPCFlagException("failed to remove %s" % path) 

96 

97 

98def _runTests(): 

99 flag = IPCFlag("A") 

100 flag.set("X") 

101 assert flag.test("X") 

102 flag.clear("X") 

103 assert not flag.test("X") 

104 assert not flag.test("Y") 

105 flag.set("X") 

106 flag.set("Y") 

107 flag.set("Z") 

108 assert flag.test("X") 

109 assert flag.test("Y") 

110 assert flag.test("Z") 

111 flag.clearAll() 

112 assert not flag.test("X") 

113 assert not flag.test("Y") 

114 assert not flag.test("Z") 

115 print("All tests passed") 

116 

117if __name__ == '__main__': 117 ↛ 118line 117 didn't jump to line 118, because the condition on line 117 was never true

118 _runTests()