Issue #19, added tests for discos-keygen

Author: giuseppe-carboni

.coveragerc

@@ -1,7 +1,6 @@
 [run]
 concurrency = thread
 source = discos_client
-omit = discos_client/cli.py
 
 [paths]
 discos_client =

.gitignore

@@ -182,9 +182,9 @@ cython_debug/
 .abstra/
 
 # Visual Studio Code
-#  Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
+#  Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
 #  that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
-#  and can be added to the global gitignore or merged into this file. However, if you prefer, 
+#  and can be added to the global gitignore or merged into this file. However, if you prefer,
 #  you could uncomment the following to ignore the entire vscode folder
 # .vscode/
 
@@ -206,4 +206,4 @@ marimo/_static/
 marimo/_lsp/
 __marimo__/
 
-
+**.swp

discos_client/cli.py discos_client/scripts.pydiscos_client/cli.py renamed to discos_client/scripts.py

@@ -1,57 +1,64 @@
 import os
-import sys
 from pathlib import Path
 from argparse import ArgumentParser
 from platformdirs import user_config_dir
 from zmq.auth import create_certificates
 
-base_config = Path(user_config_dir("discos"))
-target_dir = base_config / "rpc" / "client"
-KEY_FILENAME = "identity"
-full_path_public = target_dir / f"{KEY_FILENAME}.key"
-full_path_secret = target_dir / f"{KEY_FILENAME}.key_secret"
+
+def get_config_paths():
+    base_config = Path(user_config_dir("discos"))
+    config_dir = base_config / "rpc" / "client"
+    public = config_dir / "identity.key"
+    secret = config_dir / "identity.key_secret"
+    return config_dir, public, secret
 
 
 def create_discos_keys(overwrite):
+    config_dir, public, secret = get_config_paths()
 
-    if full_path_secret.exists() and not overwrite:
+    if secret.exists() and not overwrite:
         print("Kept previously created key pair. "
               "Use --overwrite to replace it.\n")
-        return
+        return 0
 
     try:
-        target_dir.mkdir(parents=True, exist_ok=True)
+        config_dir.mkdir(parents=True, exist_ok=True)
     except OSError as e:
         print(f"Error creating the configuration directory: {e}")
-        sys.exit(1)
+        return 1
 
-    create_certificates(str(target_dir), KEY_FILENAME)
+    create_certificates(str(config_dir), "identity")
 
     if os.name == 'posix':
-        full_path_secret.chmod(0o600)
-        (target_dir / f"{KEY_FILENAME}.key").chmod(0o644)
-    print(f"Key pair created in: '{target_dir}'.")
+        public.chmod(0o644)
+        secret.chmod(0o600)
+    print(f"Key pair created in: '{config_dir}'.")
+    return 0
 
 
 def print_discos_keys():
-    if not full_path_public.exists():
+    _, public, _ = get_config_paths()
+
+    if not public.exists():
         print("No key was generated yet.")
-        return
+        return 0
 
-    with open(full_path_public, "r", encoding="utf-8") as f:
+    with open(public, "r", encoding="utf-8") as f:
         print(f.read())
-    print(f"\nPath of the public key file: {full_path_public}")
-    print(f"Remember to never share the '{KEY_FILENAME}.key_secret' file with "
+
+    print(f"\nPath of the public key file: {public}")
+    print("Remember to never share the 'identity.key_secret' file with "
           "anyone.")
     print(
         "In order to be authorized to send command to any of the telescopes, "
-        f"remember to send a copy of the '{KEY_FILENAME}.key' file to the "
+        "remember to send a copy of the 'identity.key' file to the "
         "DISCOS team, asking for authorization. Your request will be taken "
         "into consideration and you will hear back from the team."
     )
+    return 0
 
 
-def main():
+def keygen():
     parser = ArgumentParser(
         "DISCOS CURVE key pairs generator."
     )
@@ -69,5 +76,9 @@ def main():
     args = parser.parse_args()
 
     if not args.show_only:
-        create_discos_keys(args.overwrite)
-    print_discos_keys()
+        return_code = create_discos_keys(args.overwrite)
+
+        if return_code != 0:
+            return return_code
+
+    return print_discos_keys()

pyproject.toml

@@ -23,7 +23,7 @@ dependencies = [
 packages = ["discos_client"]
 
 [project.scripts]
-discos-keygen = "discos_client.cli:main"
+discos-keygen = "discos_client.scripts:keygen"
 
 [tool.setuptools.package-data]
 discos_client = ["schemas/**", "servers/**"]

tests/test_scripts.py

@@ -0,0 +1,98 @@
+import unittest
+import sys
+import shutil
+import tempfile
+from io import StringIO
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+from platformdirs import user_config_dir
+from discos_client import scripts
+
+
+class TestKeygen(unittest.TestCase):
+
+    def setUp(self):
+        self.test_dir = tempfile.mkdtemp()
+        self.test_path = Path(self.test_dir)
+        self.mock_target_dir = self.test_path / "rpc" / "client"
+        self.mock_public = self.mock_target_dir / "identity.key"
+        self.mock_secret = self.mock_target_dir / "identity.key_secret"
+
+    def tearDown(self):
+        shutil.rmtree(self.test_dir)
+
+    def test_correct_paths(self):
+        config_dir, public, secret = scripts.get_config_paths()
+        expected_config_dir = \
+            Path(user_config_dir("discos")) / "rpc" / "client"
+        expected_public = expected_config_dir / "identity.key"
+        expected_secret = expected_config_dir / "identity.key_secret"
+        self.assertEqual(config_dir, expected_config_dir)
+        self.assertEqual(public, expected_public)
+        self.assertEqual(secret, expected_secret)
+
+    @patch("discos_client.scripts.get_config_paths")
+    @patch("sys.stdout", new_callable=StringIO)
+    @patch.object(sys, "argv", ["discos-keygen"])
+    def test_keygen(self, mock_stdout, mock_paths):
+        mock_paths.return_value = (
+            self.mock_target_dir,
+            self.mock_public,
+            self.mock_secret
+        )
+        rc = scripts.keygen()
+        self.assertEqual(rc, 0)
+        self.assertTrue(self.mock_public.exists())
+        self.assertTrue(self.mock_secret.exists())
+        output = mock_stdout.getvalue()
+        self.assertIn("Key pair created in", output)
+
+    @patch("discos_client.scripts.get_config_paths")
+    @patch("sys.stdout", new_callable=StringIO)
+    @patch.object(sys, "argv", ["discos-keygen"])
+    def test_keygen_no_overwrite(self, mock_stdout, mock_paths):
+        mock_paths.return_value = (
+            self.mock_target_dir,
+            self.mock_public,
+            self.mock_secret
+        )
+        self.assertEqual(scripts.keygen(), 0)
+        self.assertTrue(self.mock_public.exists())
+        self.assertTrue(self.mock_secret.exists())
+        output = mock_stdout.getvalue()
+        self.assertIn("Key pair created in", output)
+        self.assertEqual(scripts.keygen(), 0)
+        output = mock_stdout.getvalue()
+        self.assertIn("Kept previously created key pair", output)
+
+    @patch("discos_client.scripts.get_config_paths")
+    @patch("sys.stdout", new_callable=StringIO)
+    def test_print_keys(self, mock_stdout, mock_paths):
+        mock_paths.return_value = (
+            self.mock_target_dir,
+            self.mock_public,
+            self.mock_secret
+        )
+        scripts.print_discos_keys()
+        output = mock_stdout.getvalue()
+        self.assertIn("No key was generated yet.", output)
+
+    @patch("discos_client.scripts.get_config_paths")
+    @patch("sys.stdout", new_callable=StringIO)
+    @patch.object(sys, "argv", ["discos-keygen"])
+    def test_mkdir_error(self, mock_stdout, mock_paths):
+        mock_target_dir = MagicMock()
+        mock_target_dir.mkdir.side_effect = OSError("Test error")
+        mock_paths.return_value = (
+            mock_target_dir,
+            self.mock_public,
+            self.mock_secret
+        )
+        rc = scripts.keygen()
+        self.assertEqual(rc, 1)
+        output = mock_stdout.getvalue()
+        self.assertIn("Error creating the configuration directory", output)
+
+
+if __name__ == "__main__":
+    unittest.main()