ipsec: T8975: add unit tests for vti_updown_db lock wiring

The module previously had no tests. Cover the DB logic (add/remove,
multiple connections per interface, removeAllOtherInterfaces,
setPersistentInterfaces, seed-from-file) and assert each open_* helper
acquires the serialising lock, exercised with vyos.utils.locking.Lock
patched out.

1 files changed, 249 insertions, 0 deletions

diff --git a/src/tests/test_vti_updown_db.py b/src/tests/test_vti_updown_db.py
new file mode 100644
index 000000000..b246f89eb
--- /dev/null
+++ b/src/tests/test_vti_updown_db.py
@@ -0,0 +1,249 @@
+# Copyright VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# 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/>.
+
+import os
+import tempfile
+import unittest
+
+from unittest.mock import MagicMock, patch
+
+import vyos.utils.vti_updown_db as _vti_mod
+from vyos.utils.vti_updown_db import VTIUpDownDB
+
+
+class TestVTIUpDownDBLogic(unittest.TestCase):
+    """Exercise the in-memory DB API without calling commit()."""
+
+    def setUp(self):
+        self.tmpfile = tempfile.NamedTemporaryFile(delete=False)
+        self.tmpfile.close()
+        self._path_patcher = patch.object(
+            _vti_mod, 'VTI_WANT_UP_IFLIST', self.tmpfile.name
+        )
+        self._path_patcher.start()
+
+    def tearDown(self):
+        self._path_patcher.stop()
+        if os.path.exists(self.tmpfile.name):
+            os.unlink(self.tmpfile.name)
+
+    def test_add_makes_interface_wanted_up(self):
+        """add() marks the interface as wanted-up."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_for_create_or_update
+
+        with patch('vyos.utils.vti_updown_db.Lock', MagicMock()):
+            with open_vti_updown_db_for_create_or_update() as db:
+                db.add('vti0', 'conn-a', 'IKEv2')
+                self.assertTrue(db.wantsInterfaceUp('vti0'))
+
+    def test_remove_clears_interface(self):
+        """remove() clears the entry; wantsInterfaceUp returns False when no entries remain."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_for_create_or_update
+
+        with patch('vyos.utils.vti_updown_db.Lock', MagicMock()):
+            with open_vti_updown_db_for_create_or_update() as db:
+                db.add('vti0', 'conn-a', 'IKEv2')
+                db.remove('vti0', 'conn-a', 'IKEv2')
+                self.assertFalse(db.wantsInterfaceUp('vti0'))
+
+    def test_multiple_connections_same_interface(self):
+        """Interface stays wanted-up until all its connection entries are removed."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_for_create_or_update
+
+        with patch('vyos.utils.vti_updown_db.Lock', MagicMock()):
+            with open_vti_updown_db_for_create_or_update() as db:
+                db.add('vti0', 'conn-a', 'IKEv2')
+                db.add('vti0', 'conn-b', 'IKEv2')
+                db.remove('vti0', 'conn-a', 'IKEv2')
+                # conn-b still present
+                self.assertTrue(db.wantsInterfaceUp('vti0'))
+                db.remove('vti0', 'conn-b', 'IKEv2')
+                self.assertFalse(db.wantsInterfaceUp('vti0'))
+
+    def test_remove_all_other_interfaces(self):
+        """removeAllOtherInterfaces() drops entries for interfaces not in the keep list."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_for_create_or_update
+
+        with patch('vyos.utils.vti_updown_db.Lock', MagicMock()):
+            with open_vti_updown_db_for_create_or_update() as db:
+                db.add('vti0', 'conn-a', 'IKEv2')
+                db.add('vti1', 'conn-b', 'IKEv2')
+                db.removeAllOtherInterfaces(['vti0'])
+                self.assertTrue(db.wantsInterfaceUp('vti0'))
+                self.assertFalse(db.wantsInterfaceUp('vti1'))
+
+    def test_set_persistent_interfaces(self):
+        """setPersistentInterfaces() adds bare-name (persistent) entries."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_for_create_or_update
+
+        with patch('vyos.utils.vti_updown_db.Lock', MagicMock()):
+            with open_vti_updown_db_for_create_or_update() as db:
+                db.setPersistentInterfaces(['vti0', 'vti1'])
+                # Persistent entries have no colon in the ifspec.
+                persistent = {s for s in db._ifspecs if ':' not in s}
+                self.assertEqual(persistent, {'vti0', 'vti1'})
+                self.assertTrue(db.wantsInterfaceUp('vti0'))
+                self.assertTrue(db.wantsInterfaceUp('vti1'))
+
+    def test_seed_from_file(self):
+        """VTIUpDownDB reads pre-existing entries from the file on open."""
+        with open(self.tmpfile.name, 'w') as f:
+            f.write('vti0:conn-a:IKEv2 vti1:conn-b:IKEv2')
+
+        from vyos.utils.vti_updown_db import open_vti_updown_db_readonly
+
+        with patch('vyos.utils.vti_updown_db.Lock', MagicMock()):
+            with open_vti_updown_db_readonly() as db:
+                self.assertTrue(db.wantsInterfaceUp('vti0'))
+                self.assertTrue(db.wantsInterfaceUp('vti1'))
+
+    def test_readonly_yields_none_when_db_absent(self):
+        """open_vti_updown_db_readonly() yields None (no raise) when the DB file is absent."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_readonly
+
+        # Point at a path that does not exist.
+        missing = os.path.join(tempfile.gettempdir(), 'nonexistent_vti_updown_db_test')
+        if os.path.exists(missing):
+            os.unlink(missing)
+        with patch.object(_vti_mod, 'VTI_WANT_UP_IFLIST', missing):
+            with patch('vyos.utils.vti_updown_db.Lock', MagicMock()):
+                with open_vti_updown_db_readonly() as db:
+                    self.assertIsNone(db)
+
+
+class TestVTIUpDownDBLockWiring(unittest.TestCase):
+    """Verify Lock is acquired before yielding and released on context exit."""
+
+    def setUp(self):
+        self.tmpfile = tempfile.NamedTemporaryFile(delete=False)
+        self.tmpfile.close()
+        self._path_patcher = patch.object(
+            _vti_mod, 'VTI_WANT_UP_IFLIST', self.tmpfile.name
+        )
+        self._path_patcher.start()
+
+    def tearDown(self):
+        self._path_patcher.stop()
+        if os.path.exists(self.tmpfile.name):
+            os.unlink(self.tmpfile.name)
+
+    def _make_lock_mock(self):
+        lock_instance = MagicMock()
+        lock_cls = MagicMock(return_value=lock_instance)
+        return lock_cls, lock_instance
+
+    def test_create_or_update_acquires_and_releases(self):
+        """open_vti_updown_db_for_create_or_update() acquires the lock before yielding and releases it on exit."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_for_create_or_update
+
+        lock_cls, lock_inst = self._make_lock_mock()
+        with patch('vyos.utils.vti_updown_db.Lock', lock_cls):
+            with open_vti_updown_db_for_create_or_update() as _db:
+                lock_inst.acquire.assert_called_once()
+                lock_inst.release.assert_not_called()
+            lock_inst.release.assert_called_once()
+
+    def test_update_acquires_and_releases(self):
+        """open_vti_updown_db_for_update() acquires the lock before yielding and releases it on exit."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_for_update
+
+        lock_cls, lock_inst = self._make_lock_mock()
+        with patch('vyos.utils.vti_updown_db.Lock', lock_cls):
+            with open_vti_updown_db_for_update() as _db:
+                lock_inst.acquire.assert_called_once()
+                lock_inst.release.assert_not_called()
+            lock_inst.release.assert_called_once()
+
+    def test_readonly_acquires_and_releases(self):
+        """open_vti_updown_db_readonly() acquires the lock before yielding and releases it on exit."""
+        from vyos.utils.vti_updown_db import open_vti_updown_db_readonly
+
+        lock_cls, lock_inst = self._make_lock_mock()
+        with patch('vyos.utils.vti_updown_db.Lock', lock_cls):
+            with open_vti_updown_db_readonly() as _db:
+                lock_inst.acquire.assert_called_once()
+                lock_inst.release.assert_not_called()
+            lock_inst.release.assert_called_once()
+
+
+class TestRemoveVTIUpDownDBLockOrdering(unittest.TestCase):
+    """remove_vti_updown_db() must hold the lock across os.unlink."""
+
+    def setUp(self):
+        self.tmpfile = tempfile.NamedTemporaryFile(delete=False)
+        self.tmpfile.close()
+        self._path_patcher = patch.object(
+            _vti_mod, 'VTI_WANT_UP_IFLIST', self.tmpfile.name
+        )
+        self._path_patcher.start()
+
+    def tearDown(self):
+        self._path_patcher.stop()
+        # File may have been unlinked by the function under test.
+        if os.path.exists(self.tmpfile.name):
+            os.unlink(self.tmpfile.name)
+
+    def test_lock_held_across_unlink(self):
+        """os.unlink fires before lock.release(), proving the lock covers the unlink."""
+        from vyos.utils.vti_updown_db import remove_vti_updown_db
+
+        lock_inst = MagicMock()
+        lock_cls = MagicMock(return_value=lock_inst)
+
+        # Record lock.release call_count at the moment os.unlink fires.
+        release_count_at_unlink = []
+
+        def _observe_unlink(path):
+            release_count_at_unlink.append(lock_inst.release.call_count)
+
+        with patch('vyos.utils.vti_updown_db.Lock', lock_cls):
+            with patch.object(VTIUpDownDB, 'commit'):
+                with patch.object(_vti_mod.os, 'unlink', side_effect=_observe_unlink):
+                    remove_vti_updown_db()
+
+        lock_inst.acquire.assert_called_once()
+        lock_inst.release.assert_called_once()
+        self.assertTrue(release_count_at_unlink, 'os.unlink was never called')
+        self.assertEqual(
+            release_count_at_unlink[0],
+            0,
+            'lock.release() was called before os.unlink -- unlink is not protected by the lock',
+        )
+
+    def test_remove_noop_when_db_absent(self):
+        """remove_vti_updown_db() returns without raising and does not unlink when the DB is absent."""
+        from vyos.utils.vti_updown_db import remove_vti_updown_db
+
+        # No DB file present.
+        os.unlink(self.tmpfile.name)
+        self.assertFalse(os.path.exists(self.tmpfile.name))
+
+        lock_inst = MagicMock()
+        lock_cls = MagicMock(return_value=lock_inst)
+
+        with patch('vyos.utils.vti_updown_db.Lock', lock_cls):
+            with patch.object(
+                _vti_mod.os,
+                'unlink',
+                side_effect=AssertionError('os.unlink must not be called'),
+            ) as unlink_mock:
+                remove_vti_updown_db()
+                unlink_mock.assert_not_called()
+
+        lock_inst.acquire.assert_called_once()
+        lock_inst.release.assert_called_once()
+
+
+if __name__ == '__main__':
+    unittest.main()