Coverage for drivers/ipc.py : 17%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/python3
2#
3# Copyright (C) Citrix Systems Inc.
4#
5# This program is free software; you can redistribute it and/or modify
6# it under the terms of the GNU Lesser General Public License as published
7# by the Free Software Foundation; version 2.1 only.
8#
9# This program is distributed in the hope that it will be useful,
10# but WITHOUT ANY WARRANTY; without even the implied warranty of
11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12# GNU Lesser General Public License for more details.
13#
14# You should have received a copy of the GNU Lesser General Public License
15# along with this program; if not, write to the Free Software Foundation, Inc.,
16# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18"""Communication for processes"""
20import os
21import util
22import errno
25class IPCFlagException(util.SMException):
26 pass
29class IPCFlag:
30 """Flag-based communication for processes (set, test, clear).
31 Not thread-safe."""
33 BASE_DIR = "/var/run/sm/ipc"
35 def __init__(self, ns):
36 self.ns = ns
37 self.nsDir = os.path.join(self.BASE_DIR, self.ns)
38 if not util.pathexists(self.nsDir):
39 try:
40 os.makedirs(self.nsDir)
41 except OSError:
42 pass
43 if not util.pathexists(self.nsDir):
44 raise IPCFlagException("failed to create %s" % self.nsDir)
46 def set(self, name, soft=False):
47 """Set the flag
49 name: the file to set
50 soft: If set to False and the file gets created while this function is
51 trying to set create, the file MAY be overwritten.
53 returns: True if the file is written, False otherwise."""
54 if not soft and self.test(name): # XXX this is broken!
55 return
56 flagFile = os.path.join(self.nsDir, name)
57 try:
58 if soft:
59 f = util.open_atomic(flagFile, "w")
60 else:
61 f = open(flagFile, "w")
62 f.write("%s\n" % os.getpid())
63 f.close()
64 util.SMlog("IPCFlag: set %s:%s" % (self.ns, name))
65 return True
66 except OSError as e:
67 if e.errno == errno.EEXIST and soft:
68 return False
69 else:
70 raise IPCFlagException("failed to create %s: %s" \
71 % (flagFile, e))
72 except IOError as e:
73 raise IPCFlagException("failed to create %s: %s" % (flagFile, e))
75 def test(self, name):
76 """Test the flag"""
77 flagFile = os.path.join(self.nsDir, name)
78 return util.pathexists(flagFile)
80 def clear(self, name):
81 """Clear the flag"""
82 if self.test(name):
83 flagFile = os.path.join(self.nsDir, name)
84 try:
85 os.unlink(flagFile)
86 util.SMlog("IPCFlag: clear %s:%s" % (self.ns, name))
87 except OSError:
88 raise IPCFlagException("failed to remove %s" % flagFile)
90 def clearAll(self):
91 try:
92 for file in os.listdir(self.nsDir):
93 path = os.path.join(self.nsDir, file)
94 os.unlink(path)
95 except OSError:
96 raise IPCFlagException("failed to remove %s" % path)
99def _runTests():
100 flag = IPCFlag("A")
101 flag.set("X")
102 assert flag.test("X")
103 flag.clear("X")
104 assert not flag.test("X")
105 assert not flag.test("Y")
106 flag.set("X")
107 flag.set("Y")
108 flag.set("Z")
109 assert flag.test("X")
110 assert flag.test("Y")
111 assert flag.test("Z")
112 flag.clearAll()
113 assert not flag.test("X")
114 assert not flag.test("Y")
115 assert not flag.test("Z")
116 print("All tests passed")
118if __name__ == '__main__': 118 ↛ 119line 118 didn't jump to line 119, because the condition on line 118 was never true
119 _runTests()