[node-patches] Change in ovirt-node[master]: fs: Introduce File and FakeFs
fabiand at fedoraproject.org
fabiand at fedoraproject.org
Mon Jun 10 12:29:13 UTC 2013
Fabian Deutsch has uploaded a new change for review.
Change subject: fs: Introduce File and FakeFs
......................................................................
fs: Introduce File and FakeFs
Two classes are added to allow unittesting without access to the
filesystem.
Previously unit tests were nearly impossible because of the many
informations which are retrieved from the filesystem (at runtime).
by wrapping the file i/o in a class a mocking framework (e.g.
python-mock) can be used to patch this file i/o class during unit tests
and redirect file i/o requests to a fake filesystem (e.g. an in-memory
fs).
FakeFs is a simple class implementing such a fake in-memory filesystem
which can be used during unit tests.
Change-Id: I7889240a956b359f2375955766f3cafcfbeaf0ee
Signed-off-by: Fabian Deutsch <fabiand at fedoraproject.org>
---
M src/ovirt/node/utils/fs.py
1 file changed, 125 insertions(+), 49 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/ovirt-node refs/changes/09/15509/1
diff --git a/src/ovirt/node/utils/fs.py b/src/ovirt/node/utils/fs.py
index e597712..34d4779 100644
--- a/src/ovirt/node/utils/fs.py
+++ b/src/ovirt/node/utils/fs.py
@@ -19,7 +19,7 @@
# MA 02110-1301, USA. A copy of the GNU General Public License is
# also available at http://www.gnu.org/copyleft/gpl.html.
from ovirt.node import base
-from ovirt.node.utils import is_fileobj
+import StringIO
import logging
import os
import shutil
@@ -76,6 +76,117 @@
"""
with open(filename, "wb"):
pass
+
+
+class File(base.Base):
+ """Convenience API to access files.
+ Used to make code testable
+ """
+ filename = None
+
+ def __init__(self, filename):
+ self.filename = filename
+
+ def read(self):
+ """Read the contents of a file
+ """
+ return get_contents(self.filename)
+
+ def write(self, contents, mode="wb"):
+ """Write the contents of a file
+ """
+ try:
+ atomic_write(self.filename, contents)
+ except:
+ with open(self.filename, mode) as dst:
+ dst.write(contents)
+
+ def touch(self):
+ """Touch a file
+ """
+ return truncate(self.filename)
+
+ def exists(self):
+ """Determin if a file exists
+ """
+ return os.path.isfile(self.filename)
+
+ def delete(self):
+ """Delete a file
+ """
+ return os.unlink(self.filename)
+
+ def __iter__(self):
+ with open(self.filename, "r") as src:
+ for line in src:
+ yield line
+
+
+class FakeFs(base.Base):
+ filemap = {}
+
+ @staticmethod
+ def erase():
+ """Erase all files
+ """
+ FakeFs.filemap = {}
+
+ class File(File):
+ """A fake file - residing in a dictiniory for testing
+
+ >>> FakeFs.filemap
+ {}
+ >>> f = FakeFs.File("/etc/foo")
+ >>> f.write("Hello World!")
+ >>> f.read()
+ 'Hello World!'
+ >>> FakeFs.filemap.keys()
+ ['/etc/foo']
+ >>> f.write("Hey Mars!\\nWhat's up?")
+ >>> f.read()
+ "Hey Mars!\\nWhat's up?"
+ >>> for line in f:
+ ... print("line: %s" % line)
+ line: Hey Mars!
+ <BLANKLINE>
+ line: What's up?
+ >>> f.delete()
+ >>> FakeFs.filemap
+ {}
+
+ >>> FakeFs.File("foo").write("bar")
+ >>> FakeFs.filemap
+ {'foo': 'bar'}
+ >>> FakeFs.erase()
+ >>> FakeFs.filemap
+ {}
+ """
+
+ def _cond_create(self):
+ if not self.exists():
+ FakeFs.filemap[self.filename] = ""
+
+ def read(self):
+ self._cond_create()
+ return FakeFs.filemap[self.filename]
+
+ def write(self, contents, mode=None):
+ self._cond_create()
+ FakeFs.filemap[self.filename] = contents
+
+ def touch(self):
+ self._cond_create()
+
+ def exists(self):
+ return self.filename in FakeFs.filemap
+
+ def delete(self):
+ if self.exists():
+ del FakeFs.filemap[self.filename]
+
+ def __iter__(self):
+ for line in StringIO.StringIO(self.read()):
+ yield line
class BackupedFiles(base.Base):
@@ -175,11 +286,11 @@
True if the file is a bind mount target
"""
bind_mount_found = False
- with open("/proc/mounts") as mounts:
- pattern = "%s %s" % (filename, fsprefix)
- for mount in mounts:
- if pattern in mount:
- bind_mount_found = True
+ mounts = File("/proc/mounts")
+ pattern = "%s %s" % (filename, fsprefix)
+ for mount in mounts:
+ if pattern in mount:
+ bind_mount_found = True
return bind_mount_found
@@ -227,13 +338,11 @@
class ShellVarFile(base.Base):
"""ShellVarFile writes simple KEY=VALUE (shell-like) configuration file
- >>> import StringIO
- >>> fileobj = StringIO.StringIO()
>>> cfg = {
... "IP_ADDR": "127.0.0.1",
... "NETMASK": "255.255.255.0",
... }
- >>> p = ShellVarFile(fileobj)
+ >>> p = ShellVarFile(FakeFs.File("dst-file"))
>>> p.get_dict()
{}
>>> p.update(cfg, True)
@@ -248,42 +357,27 @@
super(ShellVarFile, self).__init__()
self.filename = filename
self.create = create
- if is_fileobj(filename) or self._fileobj:
+ if File in type(filename).mro():
self._fileobj = filename
else:
- if not create and not os.path.exists(self.filename):
+ self._fileobj = File(self.filename)
+ if not create and not self._fileobj.exists():
raise RuntimeError("File does not exist: %s" %
self.filename)
- def _is_fileobj(self):
- return is_fileobj(self.filename)
-
def _read_contents(self):
- data = None
- if self._fileobj:
- self._fileobj.seek(0)
- data = self._fileobj.read()
- else:
- with open(self.filename) as src:
- data = src.read()
- return data
+ return self._fileobj.read()
def raw_read(self):
return self._read_contents()
def _write_contents(self, data):
- if self._fileobj:
- self._fileobj.seek(0)
- self._fileobj.write(data)
- self._fileobj.flush()
- else:
- with open(self.filename, "w") as dst:
- dst.write(data)
+ self._fileobj.write(data)
def exists(self):
"""Return true if this file exists
"""
- return is_fileobj(self.filename) or os.path.isfile(self.filename)
+ return self._fileobj.exists()
def get_dict(self):
"""Returns a dict of (key, value) pairs
@@ -305,7 +399,7 @@
for key in sorted(cfg.iterkeys()):
lines.append("%s=\"%s\"" % (key, cfg[key]))
contents = "\n".join(lines) + "\n"
- self._write(contents)
+ self._write_contents(contents)
def update(self, new_dict, remove_empty):
"""Update the file using new_dict
@@ -349,21 +443,3 @@
except:
self.logger.info("Failed to parse line: '%s'" % line)
return cfg
-
- def _write(self, contents):
- if self._is_fileobj():
- self._write_contents(contents)
- else:
- # The following logic is mainly needed to allow an
- # "offline" testing
- config_fs = Config()
- if config_fs.is_enabled():
- with config_fs.open_file(self.filename, "w") as dst:
- dst.write(contents)
- else:
- try:
- atomic_write(self.filename, contents)
- except Exception as e:
- self.logger.warning("Atomic write failed: %s" % e)
- with open(self.filename, "w") as dst:
- dst.write(contents)
--
To view, visit http://gerrit.ovirt.org/15509
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I7889240a956b359f2375955766f3cafcfbeaf0ee
Gerrit-PatchSet: 1
Gerrit-Project: ovirt-node
Gerrit-Branch: master
Gerrit-Owner: Fabian Deutsch <fabiand at fedoraproject.org>
More information about the node-patches
mailing list