add user home dir

src/jupyterhub_config/custom_spawner.py

@@ -1,7 +1,9 @@
 import fcntl
 import os
+import pwd
 import subprocess
 import tempfile
+from pathlib import Path
 
 from jupyterhub.spawner import SimpleLocalProcessSpawner
 
@@ -20,10 +22,18 @@ def start(self):
         """
 
         username = self.user.name
+        global_home = Path(os.environ['JUPYTERHUB_USER_HOME'])
+        user_dir = global_home / username
 
         # Ensure the system user exists
         self._ensure_system_user(username, group='jupyterhub')
 
+        # Ensure the user directory exists and has correct permissions
+        self._ensure_user_directory(user_dir, username)
+
+        # Ensure the user's Jupiter directory exists
+        self._ensure_user_jupyter_directory(user_dir)
+
         return super().start()
 
     def _ensure_system_user(self, username: str, group: str = None):
@@ -68,3 +78,43 @@ def _ensure_system_user(self, username: str, group: str = None):
 
             finally:
                 fcntl.flock(lock, fcntl.LOCK_UN)
+
+    def _ensure_user_directory(self, user_dir: Path, username: str):
+        """
+        Ensure the user's home directory exists and is correctly owned and permissioned.
+        """
+        if not user_dir.exists():
+            self.log.info(f'Creating user directory for {username}')
+            user_dir.mkdir(parents=True)
+
+            # Get the Jupyter user's UID and GID
+            user_info = pwd.getpwnam(username)
+            uid = user_info.pw_uid
+            gid = user_info.pw_gid
+
+            # Change the directory's ownership to the user
+            os.chown(user_dir, uid, gid)
+
+            # Set directory permissions to 750: Owner (rwx), Group (r-x), Others (---)
+            os.chmod(user_dir, 0o750)
+
+        else:
+            self.log.info(f'Reusing user directory for {username}')
+
+    def _ensure_user_jupyter_directory(self, user_dir: Path):
+        """
+        Create the user's Jupyter directory and subdirectories if they do not exist. And set the
+        environment variables for Jupyter to use these directories.
+        """
+
+        jupyter_dir = user_dir / '.jupyter'
+        jupyter_runtime_dir = jupyter_dir / 'runtime'
+        juputer_data_dir = jupyter_dir / 'data'
+
+        jupyter_dir.mkdir(parents=True, exist_ok=True)
+        jupyter_runtime_dir.mkdir(parents=True, exist_ok=True)
+        juputer_data_dir.mkdir(parents=True, exist_ok=True)
+
+        self.environment['JUPYTER_CONFIG_DIR'] = str(jupyter_dir)
+        self.environment['JUPYTER_RUNTIME_DIR'] = str(jupyter_runtime_dir)
+        self.environment['JUPYTER_DATA_DIR'] = str(juputer_data_dir)

test/src/jupyterhub_config/custom_spawner_test.py

@@ -1,6 +1,7 @@
 import logging
 import subprocess
 import unittest
+from pathlib import Path
 from subprocess import CalledProcessError
 from unittest.mock import patch, MagicMock
 
@@ -99,3 +100,86 @@ def test_ensure_system_user_error(mock_run):
     ]
 
     mock_run.assert_has_calls(expected_calls, any_order=False)
+
+
+@patch('pwd.getpwnam')
+@patch('pathlib.Path.exists')
+@patch('pathlib.Path.mkdir')
+@patch('os.chown')
+@patch('os.chmod')
+def test_ensure_user_directory_with_logging(mock_chmod, mock_chown, mock_mkdir, mock_exists, mock_getpwnam, caplog):
+    username = 'testuser'
+    user_dir = Path('/home/testuser')
+
+    # Mock directory existence check (directory does not exist)
+    mock_exists.return_value = False
+
+    # Mock pwd.getpwnam to return a mock user info
+    mock_user_info = MagicMock()
+    mock_user_info.pw_uid = 1000
+    mock_user_info.pw_gid = 1000
+    mock_getpwnam.return_value = mock_user_info
+
+    with caplog.at_level(logging.INFO):
+        spawner = VirtualEnvSpawner()
+        spawner._ensure_user_directory(user_dir, username)
+
+    # Assert that mkdir, chown, and chmod were called with correct parameters
+    mock_mkdir.assert_called_once_with(parents=True)
+    mock_chown.assert_called_once_with(user_dir, 1000, 1000)
+    mock_chmod.assert_called_once_with(user_dir, 0o750)
+
+    assert f'Creating user directory for {username}' in caplog.text
+
+
+@patch('os.chown')
+@patch('os.chmod')
+@patch('pathlib.Path.mkdir')
+@patch('pathlib.Path.exists')
+def test_ensure_user_directory_reuse_existing(mock_exists, mock_mkdir, mock_chown, mock_chmod, caplog):
+    username = 'testuser'
+    user_dir = Path('/home/testuser')
+
+    # Mock directory existence check (directory already exists)
+    mock_exists.return_value = True
+
+    with caplog.at_level(logging.INFO):
+        spawner = VirtualEnvSpawner()
+        spawner._ensure_user_directory(user_dir, username)
+
+    # Assert that mkdir was not called since directory already exists
+    mock_mkdir.assert_not_called()
+
+    # Assert that chown and chmod were not called since directory already exists
+    mock_chown.assert_not_called()
+    mock_chmod.assert_not_called()
+
+    # Assert that the correct log message was created
+    assert f'Reusing user directory for {username}' in caplog.text
+
+
+@patch('pathlib.Path.mkdir')
+def test_ensure_user_jupyter_directory(mock_mkdir):
+    user_dir = Path('/home/testuser')
+
+    spawner = VirtualEnvSpawner()
+    spawner._ensure_user_jupyter_directory(user_dir)
+
+    # Assert that mkdir was called with the correct parameters
+    expected_calls = [
+        unittest.mock.call(parents=True, exist_ok=True),  # .jupyter
+        unittest.mock.call(parents=True, exist_ok=True),  # .jupyter/runtime
+        unittest.mock.call(parents=True, exist_ok=True)  # .jupyter/data
+    ]
+
+    mock_mkdir.assert_has_calls(expected_calls, any_order=False)
+
+    # Expected directories
+    jupyter_dir = user_dir / '.jupyter'
+    jupyter_runtime_dir = jupyter_dir / 'runtime'
+    juputer_data_dir = jupyter_dir / 'data'
+
+    # Assert the JUPYTER environment variables are set correctly
+    assert spawner.environment['JUPYTER_CONFIG_DIR'] == str(jupyter_dir)
+    assert spawner.environment['JUPYTER_RUNTIME_DIR'] == str(jupyter_runtime_dir)
+    assert spawner.environment['JUPYTER_DATA_DIR'] == str(juputer_data_dir)