Merge ~przemeklal/charm-local-users:add-tests into charm-local-users:main

[Przemyslaw Lal]

Add tests and fix minor issues detected by tests

Signed-off-by: Przemysław Lal <email address hidden>

[Przemyslaw Lal (przemeklal)]

Tests are passing, total coverage is 74%:

Name Stmts Miss Branch BrPart Cover Missing
--------------------------------------------------------------------
src/__init__.py 0 0 0 0 100%
src/charm.py 77 21 34 7 69% 55, 68->72, 77->90, 79-84, 111, 113-120, 127-129, 139-145, 149
src/lib/__init__.py 0 0 0 0 100%
src/lib/local_users.py 113 22 32 2 78% 35-43, 52-61, 66-67, 82-84, 101-102, 141->exit
--------------------------------------------------------------------
TOTAL 190 43 66 9 74%

[🤖 Canonical IS Merge Bot (canonical-is-mergebot)]

This merge proposal is being monitored by mergebot. Change the status to Approved to merge.

[🤖 Canonical IS Merge Bot (canonical-is-mergebot)]

Change successfully merged at revision 3fcfb4b15c5a94f0eb7dd29c899ae5e2dfb6a57f

diff --git a/.gitignore b/.gitignore
index 2c3f0e5..f55930e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,5 +3,6 @@ build/
 *.charm
 
 .coverage
+coverage.xml
 __pycache__/
 *.py[cod]
diff --git a/run_tests b/run_tests
index 78c5a62..9f708b5 100755
--- a/run_tests
+++ b/run_tests
@@ -25,4 +25,5 @@ fi
 
 flake8
 coverage run --branch --source=src -m unittest -v "$@"
+coverage xml -o coverage.xml
 coverage report -m
diff --git a/src/lib/local_users.py b/src/lib/local_users.py
index 7a32124..7267967 100755
--- a/src/lib/local_users.py
+++ b/src/lib/local_users.py
@@ -27,6 +27,8 @@ log = logging.getLogger(__name__)
 
 User = namedtuple("User", ["name", "gecos", "ssh_key"])
 
+HOME_DIR_PATH = "/home"
+
 
 def add_user(username, shell="/bin/bash", home_dir=None, gecos=None):
     """Add a user to the system."""
@@ -102,15 +104,16 @@ def remove_group(group_name):
 
 def rename_group(old_name, new_name):
     """Rename the `old_name` group to `new_name` using `groupmod` command."""
-    cmd = ["groupmod", "-n", new_name, old_name]
-    subprocess.check_call(cmd)
+    if old_name != new_name:
+        cmd = ["groupmod", "-n", new_name, old_name]
+        subprocess.check_call(cmd)
 
 
 def set_ssh_authorized_key(user):
     """Idempotently set up the SSH public key in `authorized_keys`."""
     comment = "# charm-local-users"
     authorized_key = " ".join([user.ssh_key, comment])
-    ssh_path = os.path.join("/home", user.name, ".ssh")
+    ssh_path = os.path.join(HOME_DIR_PATH, user.name, ".ssh")
     authorized_keys_path = os.path.join(ssh_path, "authorized_keys")
     if not os.path.exists(ssh_path):
         os.makedirs(ssh_path, mode=0o700)
diff --git a/tests/__init__.py b/tests/__init__.py
index e69de29..bb15096 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -0,0 +1,7 @@
+import sys
+from unittest import mock
+
+# mock shutil for all tests
+shutil = mock.MagicMock()
+shutil.chown = mock.MagicMock()
+sys.modules['shutil'] = shutil
diff --git a/tests/test_charm.py b/tests/test_charm.py
index f0bbaa2..ce70df8 100644
--- a/tests/test_charm.py
+++ b/tests/test_charm.py
@@ -13,8 +13,10 @@
 # limitations under the License.
 
 import unittest
+from unittest.mock import patch
 
 from charm import CharmLocalUsersCharm
+from ops.model import ActiveStatus, BlockedStatus
 from ops.testing import Harness
 
 
@@ -23,3 +25,74 @@ class TestCharm(unittest.TestCase):
         self.harness = Harness(CharmLocalUsersCharm)
         self.addCleanup(self.harness.cleanup)
         self.harness.begin()
+
+    @patch("os.makedirs")
+    @patch("charmhelpers.core.host.group_exists")
+    @patch("charm.rename_group")
+    @patch("charm.get_group_users")
+    @patch("charmhelpers.core.host.add_group")
+    @patch("charm.configure_user")
+    def test_config_changed(
+        self,
+        mock_conf_user,
+        mock_add_group,
+        mock_get_gr_users,
+        mock_rename,
+        mock_exists,
+        _,
+    ):
+        # group doesn't exist yet
+        mock_exists.return_value = False
+
+        # correct configuration
+        self.harness.update_config(
+            {"group": "testgroup", "users": "test1;Test 1;key1\ntest2;Test 2;key2"}
+        )
+        # a new group must be created
+        mock_add_group.assert_called_once_with("testgroup")
+        # first execution, no rename expected
+        mock_rename.assert_not_called()
+        mock_get_gr_users.assert_called_once()
+        # 2 users to configure
+        self.assertEqual(mock_conf_user.call_count, 2)
+        # everything went well
+        self.assertIsInstance(self.harness.model.unit.status, ActiveStatus)
+
+    @patch("os.makedirs")
+    @patch("charmhelpers.core.host.group_exists")
+    @patch("charm.rename_group")
+    @patch("charm.get_group_users")
+    @patch("charmhelpers.core.host.add_group")
+    @patch("charm.configure_user")
+    def test_config_changed_invalid_userlist(
+        self,
+        mock_conf_user,
+        mock_add_group,
+        mock_get_gr_users,
+        mock_rename,
+        mock_exists,
+        _,
+    ):
+        # group doesn't exist yet
+        mock_exists.return_value = False
+
+        # empty users list
+        self.harness.update_config(
+            {"group": "testgroup", "users": ";User with no name;\n"}
+        )
+
+        # a new group must be created
+        mock_add_group.assert_called_once_with("testgroup")
+        # first execution, no rename expected
+        mock_rename.assert_not_called()
+        # we shouldn't be comapring with existing users if the config is invalid
+        mock_get_gr_users.assert_not_called()
+        # no users to configure
+        mock_conf_user.assert_not_called()
+        # we should enter blocked state
+        self.assertIsInstance(self.harness.model.unit.status, BlockedStatus)
+
+    @patch("os.makedirs")
+    def test_empty_group_config(self, _):
+        self.harness.update_config({"group": "", "users": "test;;"})
+        self.assertIsInstance(self.harness.model.unit.status, BlockedStatus)
diff --git a/tests/test_local_users.py b/tests/test_local_users.py
new file mode 100644
index 0000000..b53720f
--- /dev/null
+++ b/tests/test_local_users.py
@@ -0,0 +1,154 @@
+# Copyright 2021 Canonical
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import unittest
+from collections import namedtuple
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
+
+from lib import local_users
+
+
+class TestLocalUsers(unittest.TestCase):
+    @patch("os.chmod")
+    def test_set_ssh_authorized_key_update(self, _):
+        testuser = local_users.User(
+            "testuser", ["Test User", "", "", "", ""], "ssh-rsa ABC testuser@testhost"
+        )
+
+        testuser2 = local_users.User(
+            "testuser", ["Test User", "", "", "", ""], "ssh-rsa XYZ testuser@testhost"
+        )
+
+        with TemporaryDirectory() as fake_home:
+            testfile_path = os.path.join(fake_home, "testuser", ".ssh", "authorized_keys")
+            with patch("lib.local_users.HOME_DIR_PATH", fake_home):
+                local_users.set_ssh_authorized_key(testuser)
+                with open(testfile_path, "r") as f:
+                    keys = f.readlines()
+                    self.assertIn("ssh-rsa ABC testuser@testhost # charm-local-users\n", keys)
+
+                # update the key
+                local_users.set_ssh_authorized_key(testuser2)
+                with open(testfile_path, "r") as f:
+                    keys = f.readlines()
+                    self.assertIn("ssh-rsa XYZ testuser@testhost # charm-local-users\n", keys)
+                    self.assertNotIn("ssh-rsa ABC testuser@testhost # charm-local-users\n", keys)
+
+    def test_parse_gecos(self):
+        test_cases = [
+            ("", ["", "", "", "", ""]),
+            ("Test User", ["Test User", "", "", "", ""]),
+            ("Test,,,,", ["Test", "", "", "", ""]),
+            ("Test,,", ["Test", "", "", "", ""]),
+            ("Test,,+0123456789", ["Test", "", "+0123456789", "", ""]),
+            (",,,", ["", "", "", "", ""]),
+            (",,,,,,ignored", ["", "", "", "", ""]),
+            (",,,,'other' field", ["", "", "", "", "'other' field"]),
+        ]
+        for tc in test_cases:
+            result = local_users.parse_gecos(tc[0])
+            self.assertEqual(result, tc[1])
+
+    def test_get_gecos(self):
+        testcases = [
+            (
+                # standard account
+                "testuser",
+                b"testuser:x:1000:1000:Test User,,,:/home/testuser:/usr/bin/bash",
+                ["Test User", "", "", "", ""],
+            ),
+            (
+                # system account, empty GECOS
+                "svcaccount",
+                b"svcaccount:x:999:999::/var/lib/svcaccount:/bin/false",
+                ["", "", "", "", ""],
+            ),
+            (
+                # all fields
+                "testuser",
+                b"testuser:x:1000:1000:Test User,ACME,+123,+321,test:/home/testuser:/usr/bin/bash",
+                ["Test User", "ACME", "+123", "+321", "test"],
+            ),
+            (
+                # one field, no commas
+                "testuser",
+                b"testuser:x:1000:1000:Test:home/testuser:/usr/bin/bash",
+                ["Test", "", "", "", ""],
+            ),
+        ]
+        for tc in testcases:
+            with patch("subprocess.check_output") as mock_cmd:
+                mock_cmd.return_value = tc[1]
+                result = local_users.get_gecos(tc[0])
+                self.assertEqual(result, tc[2])
+
+    def test_update_gecos(self):
+        # ensure that only fields that changed are passed to chfn
+        testcase = namedtuple("testcase", ["prev", "updated", "expected", "should_call"])
+
+        # NOTE: order of chfn flags in GECOS fields order is: -f -r -w -h -o
+        testcases = [
+            testcase(["", "", "", "", ""], ["", "", "", "", ""], [], False),
+            testcase(["A", "B", "0", "1", "2"], ["A", "B", "0", "1", "2"], [], False),
+            testcase(
+                ["A", "B", "0", "1", "2"],
+                ["A", "X", "9", "1", "2"],
+                ["-r", "X", "-w", "9"],
+                True,
+            ),
+            testcase(["A", "", "", "", ""], ["B", "", "", "", ""], ["-f", "B"], True),
+        ]
+        for tc in testcases:
+            with patch("lib.local_users.get_gecos") as mock_prev, patch(
+                "subprocess.check_call"
+            ) as mock_call:
+                u = local_users.User("test", tc.updated, "")
+                mock_prev.return_value = tc.prev
+                local_users.update_gecos(u)
+                if tc.should_call:
+                    expected_cmd = ["chfn"] + tc.expected + ["test"]
+                    mock_call.assert_called_once_with(expected_cmd)
+                else:
+                    mock_call.assert_not_called()
+
+    def test_rename_group(self):
+        with patch("subprocess.check_call") as mock_call:
+            # the old and the new name are identical - ensure groupmod not called
+            local_users.rename_group("test1", "test1")
+            mock_call.assert_not_called()
+
+            # ensure it is called when renaming test1 to test2
+            local_users.rename_group("test1", "test2")
+            mock_call.assert_called_once_with(["groupmod", "-n", "test2", "test1"])
+
+    def test_get_user_membership(self):
+        with patch("subprocess.check_output") as mock_call:
+            expected = ["test", "group1", "group2", "group3"]
+            mock_call.return_value = b"test : test group1 group2 group3"
+            result = local_users.get_user_membership("test")
+            self.assertListEqual(result, expected)
+
+    def test_get_group_users(self):
+        with patch("subprocess.check_output") as mock_call:
+            mock_call.return_value = b"acme:x:1001:test1,test2,test3"
+            result = local_users.get_group_users("acme")
+            self.assertListEqual(result, ["test1", "test2", "test3"])
+
+    def test_get_group_users_empty(self):
+        with patch("subprocess.check_output") as mock_call:
+            mock_call.return_value = b"acme:x:1001:"
+            result = local_users.get_group_users("acme")
+            self.assertListEqual(result, [])