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
127
128
|
# Copyright (C) 2010
# Red Hat, Inc. All rights reserved.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# Author(s): Martin Sivak <msivak@redhat.com>
from StringIO import StringIO
import fnmatch
import os
class DiskIO(object):
"""Simple object to simplify mocking of file operations in Mock
based testing"""
class TestFile(StringIO):
def __init__(self, store, path, content = ""):
StringIO.__init__(self, content)
self._store = store
self._path = path
self._ro = False
def flush(self):
self._store[self._path] = self.getvalue()
def close(self):
self.flush()
return StringIO.close(self)
def __del__(self):
try:
self.close()
except (AttributeError, ValueError):
pass
def __enter__(self):
return self
def __exit__(self, *_):
self.close()
class Dir(object):
pass
class Link(object):
pass
def __init__(self):
self.reset()
def __getitem__(self, key):
return self.fs[key]
def __setitem__(self, key, value):
self.fs[key] = value
def reset(self):
self.fs = {
"/proc": self.Dir,
"/proc/cmdline": "linux",
}
self._pwd = "/"
#Emulate file objects
def open(self, filename, mode = "r"):
path = os.path.join(self._pwd, filename)
content = self.fs.get(path, None)
if content == self.Dir:
raise IOError("[Errno 21] Is a directory: '%s'" % (path))
elif mode.startswith("w"):
self.fs[path] = ""
f = self.TestFile(self.fs, path, self.fs[path])
elif mode.endswith("a"):
if not path in self.fs:
self.fs[path] = ""
f = self.TestFile(self.fs, path, self.fs[path])
f.seek(0, os.SEEK_END)
elif content == None:
raise IOError("[Errno 2] No such file or directory: '%s'" % (path,))
elif mode.endswith("+"):
f = self.TestFile(self.fs, path, content)
if mode.startswith('r'):
f.seek(0, os.SEEK_SET)
else:
f.seek(0, os.SEEK_END)
else:
f = self.TestFile(self.fs, path, content)
return f
#Emulate os calls
def glob_glob(self, pattern):
return fnmatch.filter(self.fs.keys(), pattern)
def os_listdir(self, path):
return [entry[len(path):].lstrip('/') for entry in self.fs.keys()\
if entry.startswith(path) and entry != path]
def os_path_exists(self, path):
path = os.path.join(self._pwd, path)
return self.fs.has_key(path)
def os_path_isdir(self, path):
if not path.endswith("/"):
path += "/"
path += "*"
return len(fnmatch.filter(self.fs.keys(), path)) > 0
def os_remove(self, path):
path = os.path.join(self._pwd, path)
try:
del self.fs[path]
except KeyError:
raise OSError("[Errno 2] No such file or directory: '%s'" % (path,))
def os_access(self, path, mode):
return self.os_path_exists(path)
|
