'force' option for 'file' module.

library/file

@@ -224,6 +224,7 @@ def main():
             state = dict(choices=['file','directory','link','absent'], default='file'),
             path  = dict(aliases=['dest', 'name'], required=True),
             src   = dict(),
+            force = dict(default='no', choices=['yes', 'no']),
             mode  = dict(),
             owner = dict(),
             group = dict(),
@@ -240,6 +241,7 @@ def main():
     src    = params.get('src', None)
     if src:
         src = os.path.expanduser(src)
+    force = module.boolean(params['force'])
 
     mode   = params.get('mode', None)
     owner  = params.get('owner', None)
@@ -288,7 +290,7 @@ def main():
             module_fail_json(path=path, msg=str(e))
         module_exit_json(path=path, changed=True)
 
-    if prev_state != 'absent' and prev_state != state:
+    if prev_state != 'absent' and prev_state != state and not force:
         module_fail_json(path=path, msg='refusing to convert between %s and %s' % (prev_state, state))
 
     if prev_state == 'absent' and state == 'absent':
@@ -296,7 +298,7 @@ def main():
 
     if state == 'file':
 
-        if prev_state == 'absent':
+        if prev_state != 'file':
             module_fail_json(path=path, msg='file does not exist, use copy or template module to create')
 
         # set modes owners and context as needed
@@ -312,6 +314,13 @@ def main():
         if prev_state == 'absent':
             os.makedirs(path)
             changed = True
+        elif prev_state != 'directory' and force:
+            try:
+                os.unlink(path)
+            except Exception, e:
+                module_fail_json(path=path, msg=str(e))
+            os.makedirs(path)
+            changed = True
 
         # set modes owners and context as needed
         changed = set_context_if_different(path, secontext, changed)
@@ -333,6 +342,16 @@ def main():
         if prev_state == 'absent':
             os.symlink(src, path)
             changed = True
+        elif prev_state != 'link' and force:
+            try:
+                if os.path.isfile(path):
+                    os.unlink(path)
+                else:
+                    shutil.rmtree(path, ignore_errors=False, onerror=rmtree_error)
+            except Exception, e:
+                module_fail_json(path=path, msg=str(e))
+            os.symlink(src, path)
+            changed = True
         elif prev_state == 'link':
             old_src = os.readlink(path)
             if not os.path.isabs(old_src):

test/TestRunner.py

@@ -157,6 +157,111 @@ class TestRunner(unittest.TestCase):
         # test the force option when set
         result = self._run('git', [ "repo=\"file:///tmp/gitdemo\"", "dest=/tmp/gd", "force=yes" ])
         assert result['changed']
+    
+    def test_file(self):
+        filedemo = tempfile.mkstemp()[1]
+        assert self._run('file', ['dest=' + filedemo, 'state=directory'])['failed']
+        assert os.path.isfile(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link'])['failed']
+        assert os.path.isfile(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'mode=604', 'state=file'])['changed']
+        assert os.path.isfile(filedemo) and os.stat(filedemo).st_mode == 0100604
+
+        assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
+        assert not os.path.exists(filedemo)
+        assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
+
+
+        filedemo = tempfile.mkdtemp()
+        assert self._run('file', ['dest=' + filedemo, 'state=file'])['failed']
+        assert os.path.isdir(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link'])['failed']
+        assert os.path.isdir(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'mode=701', 'state=directory'])['changed']
+        assert os.path.isdir(filedemo) and os.stat(filedemo).st_mode == 040701
+
+        assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
+        assert not os.path.exists(filedemo)
+        assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
+
+
+        tmp_dir = tempfile.mkdtemp()
+        filedemo = os.path.join(tmp_dir, 'link')
+        os.symlink('/dev/zero', filedemo)
+        assert self._run('file', ['dest=' + filedemo, 'state=file'])['failed']
+        assert os.path.islink(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'state=directory'])['failed']
+        assert os.path.islink(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link'])['changed']
+        assert os.path.islink(filedemo) and os.path.realpath(filedemo) == '/dev/null'
+
+        assert self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
+        assert not os.path.exists(filedemo)
+        assert not self._run('file', ['dest=' + filedemo, 'state=absent'])['changed']
+        os.rmdir(tmp_dir)
+
+
+        filedemo = tempfile.mkstemp()[1]
+        assert self._run('file', ['dest=' + filedemo, 'state=directory', 'force=yes'])['changed']
+        assert os.path.isdir(filedemo)
+        os.rmdir(filedemo)
+
+        filedemo = tempfile.mkstemp()[1]
+        assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link', 'force=yes'])['changed']
+        assert os.path.islink(filedemo)
+        os.unlink(filedemo)
+
+        filedemo = tempfile.mkstemp()[1]
+        assert self._run('file', ['dest=' + filedemo, 'mode=604', 'state=file', 'force=yes'])['changed']
+        assert os.path.isfile(filedemo) and os.stat(filedemo).st_mode == 0100604
+
+        assert self._run('file', ['dest=' + filedemo, 'state=absent', 'force=yes'])['changed']
+        assert not os.path.exists(filedemo)
+        assert not self._run('file', ['dest=' + filedemo, 'state=absent', 'force=yes'])['changed']
+
+
+        filedemo = tempfile.mkdtemp()
+        assert self._run('file', ['dest=' + filedemo, 'state=file', 'force=yes'])['failed']
+        assert os.path.isdir(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link', 'force=yes'])['changed']
+        assert os.path.islink(filedemo)
+        os.unlink(filedemo)
+
+        filedemo = tempfile.mkdtemp()
+        assert self._run('file', ['dest=' + filedemo, 'mode=701', 'state=directory', 'force=yes'])['changed']
+        assert os.path.isdir(filedemo) and os.stat(filedemo).st_mode == 040701
+        os.path.isdir(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'state=absent', 'force=yes'])['changed']
+        assert not os.path.exists(filedemo)
+        assert not self._run('file', ['dest=' + filedemo, 'state=absent', 'force=yes'])['changed']
+
+
+        tmp_dir = tempfile.mkdtemp()
+        filedemo = os.path.join(tmp_dir, 'link')
+        os.symlink('/dev/zero', filedemo)
+        assert self._run('file', ['dest=' + filedemo, 'state=file', 'force=yes'])['failed']
+        assert os.path.islink(filedemo)
+
+        assert self._run('file', ['dest=' + filedemo, 'state=directory', 'force=yes'])['changed']
+        assert os.path.isdir(filedemo)
+        os.rmdir(filedemo)
+
+        os.symlink('/dev/zero', filedemo)
+        assert self._run('file', ['dest=' + filedemo, 'src=/dev/null', 'state=link', 'force=yes'])['changed']
+        assert os.path.islink(filedemo) and os.path.realpath(filedemo) == '/dev/null'
+
+        assert self._run('file', ['dest=' + filedemo, 'state=absent', 'force=yes'])['changed']
+        assert not os.path.exists(filedemo)
+        assert not self._run('file', ['dest=' + filedemo, 'state=absent', 'force=yes'])['changed']
+        os.rmdir(tmp_dir)
 
     def test_large_output(self):
         large_path = "/usr/share/dict/words"