Check for existing symlink while force creating symlink

[sshedi]

If a dead symlink by the same name is present, os.path.exists returns
false.

Signed-off-by: Shreenidhi Shedi sshedi@vmware.com
Signed-off-by: Shreenidhi Shedi yesshedi@vmware.com

Proposed Commit Message

summary: Check for existing symlink while force creating symlink

If a dead symlink by the same name is present, os.path.exists returns
false.

Additional Context

Test Steps

Checklist:

• My code follows the process laid out in the documentation
• I have updated or added any unit tests accordingly
• I have updated or added any documentation accordingly

[sshedi]

https://github.com/canonical/cloud-init/commit/217ef6ba6c52788f4363b998b6da08863fea5cd9

This tries to remove existing symlink of cloud-id and create a new one
but it fails with

Traceback (most recent call last):
File "/usr/lib/python3.10/site-packages/cloudinit/sources/init.py", line 912, in find_source
if s.update_metadata_if_supported(
File "/usr/lib/python3.10/site-packages/cloudinit/sources/init.py", line 796, in update_metadata_if_supported
result = self.get_data()
File "/usr/lib/python3.10/site-packages/cloudinit/sources/init.py", line 358, in get_data
self.persist_instance_data()
File "/usr/lib/python3.10/site-packages/cloudinit/sources/init.py", line 421, in persist_instance_data
util.sym_link(f"{cloud_id_file}-{cloud_id}", cloud_id_file, force=True)
File "/usr/lib/python3.10/site-packages/cloudinit/util.py", line 1891, in sym_link
os.symlink(source, link)
FileExistsError: [Errno 17] File exists: '/run/cloud-init/cloud-id-none' -> '/run/cloud-init/cloud-id'
2022-02-20 05:57:12,829 - util.py[WARNING]: No instance datasource found! Likely bad things to come!

Steps to reproduce:
- Trigger cloud-init services with an existing DS on the first run
- On the second run, there should not be any DS, this should trigger above error (because cloud-id symlink is already present but dead and os.path.exists("/run/cloud-init/cloud-id") return False)

[sshedi]

Minimal program to reproduce the issue:

#!/usr/bin/python3

import os

from cloudinit import util

src = "/tmp/test-file"
link = "/tmp/test_link"

os.system("touch " + src)
os.system("ln -sfv " + src +  " " + link)
os.system("rm -f " + src)

try:
    util.sym_link(src, link, force=True)
finally:
    os.system("rm -f " + src + " " + link)

[sshedi]

@blackboxsw / @TheRealFalcon - Can you please take a look at this PR?

[smoser]

I have a couple thoughts here.

• you might just be fixing a symptom of a different root cause issue
• It seems that create_symlink with force should do what you’re wanting it to.
• this may be related to some of the bugs referenced from https://bugs.launchpad.net/cloud-init/+bug/1883903
• your unit test is really testing 3 different behaviors all in one. We should split that test up into multiple tests with specific names. (source_exists, target_exists, source_exists_dangling …)

[smoser]

Possibly better test cases, or at least 4 of them vs 2:

diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 53f322ed6..23439ccac 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -360,51 +360,47 @@ class TestUtil(CiTestCase):
         is_rw = util.mount_is_read_write("/")
         self.assertEqual(is_rw, False)
 
-    def sym_link_test_helper(self):
-        link = tempfile.mktemp()
-        _, src = tempfile.mkstemp()
 
-        util.del_file(link)
-        util.sym_link(src, link)
-        return link, src
+class TestSymlink(CiTestCase):
+    def test_sym_link_simple(self):
+        tmpd = self.tmp_dir()
+        link = self.tmp_path("link", tmpd)
+        target = self.tmp_path("target", tmpd)
+        util.write_file(target, "hello")
 
-    def test_sym_link_source_exists(self):
-        exception = False
-        link, src = self.sym_link_test_helper()
-
-        # recreate symlink
-        try:
-            util.sym_link(src, link)
-        except FileExistsError:
-            exception = True
+        util.sym_link(target, link)
+        self.assertTrue(os.path.exists(link))
+        self.assertTrue(os.path.islink(link))
 
-        assert exception
+    def test_sym_link_source_exists(self):
+        tmpd = self.tmp_dir()
+        oldlink = self.tmp_path("link", tmpd)
+        newlink = self.tmp_path("newlink", tmpd)
+        target = self.tmp_path("target", tmpd)
+        util.write_file(target, "hello")
+        os.symlink(target, oldlink)
 
-        # force create
-        try:
-            util.sym_link(src, link, force=True)
-        finally:
-            util.del_file(link)
-            util.del_file(src)
+        util.sym_link(target, newlink, force=True)
+        self.assertTrue(os.path.exists(newlink))
 
     def test_sym_link_dangling_link(self):
-        exception = False
-        link, src = self.sym_link_test_helper()
-
-        # make symlink dangling & recreate
-        util.del_file(src)
-        try:
-            util.sym_link(src, link)
-        except FileExistsError:
-            exception = True
-
-        assert exception
-
-        # force recreate
-        try:
-            util.sym_link(src, link, force=True)
-        finally:
-            util.del_file(link)
+        tmpd = self.tmp_dir()
+        oldlink = self.tmp_path("link", tmpd)
+        newlink = self.tmp_path("newlink", tmpd)
+        target = self.tmp_path("target", tmpd)
+        os.symlink(target, oldlink)
+
+        util.sym_link(target, newlink, force=True)
+        self.assertTrue(os.path.islink(newlink))
+
+    def test_sym_link_create_dangling(self):
+        tmpd = self.tmp_dir()
+        link = self.tmp_path("link", tmpd)
+        target = self.tmp_path("target", tmpd)
+
+        util.sym_link(target, link)
+        self.assertTrue(os.path.islink(link))
+        self.assertFalse(os.path.exists(link))
 
 
 class TestUptime(CiTestCase):

[sshedi]

@smoser thanks for the assistance. I have updated the tests like you suggested.
I tried to keep the tests minimal earlier.

[smoser]

These changes seem sane to me.

[holmanb]

Why not use os.path.lexists(), since it seems that already has the preferred behavior?

[holmanb]

Could you please add a test that demonstrates how former behavior was flawed? I'd like to see a test that fails with the current (flawed) upstream behavior but succeeds with your "fix". If I revert your change in cloudinit/util.py but leave your new tests, none of the tests fail.

[sshedi]

Could you please add a test that demonstrates how former behavior was flawed? I'd like to see a test that fails with the current (flawed) upstream behavior but succeeds with your "fix". If I revert your change in cloudinit/util.py but leave your new tests, none of the tests fail.

Fixed the tests. Please check now.

[holmanb]

Thanks!

[blackboxsw]

Agreed, this traceback would affect any system which detects a new or different datasource as the sym_link(force=True) would have failed with this traceback. Not an ideal failure path. and I agree with the fix and test coverage here.

[TheRealFalcon]

Closed and re-opened to retrigger CI