add test case, update password if changed

docker/docker-prepare.sh

@@ -9,16 +9,13 @@ wait_for_postgres() {
 	host="${PAPERLESS_DBHOST}"
 	port="${PAPERLESS_DBPORT}"
 
-	if [[ -z $port ]] ;
-	then
+	if [[ -z $port ]]; then
 		port="5432"
 	fi
 
-	while !</dev/tcp/$host/$port ;
-	do
+	while ! </dev/tcp/$host/$port; do
 
-		if [ $attempt_num -eq $max_attempts ]
-		then
+		if [ $attempt_num -eq $max_attempts ]; then
 			echo "Unable to connect to database."
 			exit 1
 		else
@@ -31,13 +28,7 @@ wait_for_postgres() {
 	done
 }
 
-
 migrations() {
-	if [[ -n "${PAPERLESS_DBHOST}" ]]
-	then
-		wait_for_postgres
-	fi
-
 	(
 		# flock is in place to prevent multiple containers from doing migrations
 		# simultaneously. This also ensures that the db is ready when the command
@@ -45,24 +36,37 @@ migrations() {
 		flock 200
 		echo "Apply database migrations..."
 		python3 manage.py migrate
-	)  200>/usr/src/paperless/data/migration_lock
+	) 200>/usr/src/paperless/data/migration_lock
 }
 
 search_index() {
 	index_version=1
 	index_version_file=/usr/src/paperless/data/.index_version
 
-	if [[ (! -f "$index_version_file") || $(< $index_version_file) != "$index_version" ]]; then
+	if [[ (! -f "$index_version_file") || $(<$index_version_file) != "$index_version" ]]; then
 		echo "Search index out of date. Updating..."
 		python3 manage.py document_index reindex
 		echo $index_version | tee $index_version_file >/dev/null
 	fi
 }
 
-do_work() {
-	migrations;
-	
-	search_index;
+superuser() {
+	if [[ -n "${PAPERLESS_ADMIN_USER}" ]]; then
+		sudo -HEu paperless python3 manage.py manage_superuser
+	fi
 }
 
-do_work;
+do_work() {
+	if [[ -n "${PAPERLESS_DBHOST}" ]]; then
+		wait_for_postgres
+	fi
+
+	migrations
+
+	search_index
+
+	superuser
+
+}
+
+do_work

src/documents/management/commands/manage_superuser.py

@@ -1,9 +1,13 @@
+import logging
 import os
 
 from django.contrib.auth.models import User
 from django.core.management.base import BaseCommand, CommandError
 
 
+logger = logging.getLogger("paperless.management.superuser")
+
+
 class Command(BaseCommand):
 
     help = """
@@ -12,29 +16,25 @@ class Command(BaseCommand):
 
     def handle(self, *args, **options):
 
-        PAPERLESS_ADMIN_USER = os.getenv('PAPERLESS_ADMIN_USER')
-        PAPERLESS_ADMIN_MAIL = os.getenv('PAPERLESS_ADMIN_MAIL', 'root@localhost')
-        PAPERLESS_ADMIN_PASSWORD = os.getenv('PAPERLESS_ADMIN_PASSWORD')
+        username = os.getenv('PAPERLESS_ADMIN_USER')
+        if not username:
+            return
 
-        # If PAPERLESS_ADMIN_USER env variable is set
-        if PAPERLESS_ADMIN_USER:
-            try:
-                # Check if user exists already, leave as is if it does
-                if User.objects.filter(
-                    username=PAPERLESS_ADMIN_USER).exists():
-                    self.stdout.write(
-                        f'The user "{PAPERLESS_ADMIN_USER}" already exists! Leaving user as is.')
-                elif PAPERLESS_ADMIN_PASSWORD:
-                    # Create superuser based on env variables
-                    User.objects.create_superuser(
-                        PAPERLESS_ADMIN_USER, PAPERLESS_ADMIN_MAIL, PAPERLESS_ADMIN_PASSWORD)
-                    self.stdout.write(
-                        f'Created superuser "{PAPERLESS_ADMIN_USER}" with provided password.')
-                else:
-                    self.stdout.write(
-                        f'Did not create superuser "{PAPERLESS_ADMIN_USER}".')
-                    self.stdout.write(
-                        'Make sure you specified "PAPERLESS_ADMIN_PASSWORD" in your "docker-compose.env" file.')
-            except Exception as error:
-                self.stdout.write(
-                    f'Exception occured while creating superuser: {error}')
+        mail = os.getenv('PAPERLESS_ADMIN_MAIL', 'root@localhost')
+        password = os.getenv('PAPERLESS_ADMIN_PASSWORD')
+
+        # Check if user exists already, leave as is if it does
+        if User.objects.filter(username=username).exists():
+            user: User = User.objects.get_by_natural_key(username)
+            user.set_password(password)
+            user.save()
+        elif password:
+            # Create superuser based on env variables
+            User.objects.create_superuser(username, mail, password)
+            self.stdout.write(
+                f'Created superuser "{username}" with provided password.')
+        else:
+            self.stdout.write(
+                f'Did not create superuser "{username}".')
+            self.stdout.write(
+                'Make sure you specified "PAPERLESS_ADMIN_PASSWORD" in your "docker-compose.env" file.')

src/documents/tests/test_management_superuser.py

@@ -0,0 +1,66 @@
+import os
+import shutil
+from unittest import mock
+
+from django.contrib.auth.models import User
+from django.core.management import call_command
+from django.test import TestCase
+
+from documents.management.commands.document_thumbnails import _process_document
+from documents.models import Document, Tag, Correspondent, DocumentType
+from documents.tests.utils import DirectoriesMixin
+
+
+class TestManageSuperUser(DirectoriesMixin, TestCase):
+
+    def reset_environment(self):
+        if "PAPERLESS_ADMIN_USER" in os.environ:
+            del os.environ["PAPERLESS_ADMIN_USER"]
+        if "PAPERLESS_ADMIN_PASSWORD" in os.environ:
+            del os.environ["PAPERLESS_ADMIN_PASSWORD"]
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.reset_environment()
+
+    def tearDown(self) -> None:
+        super().tearDown()
+        self.reset_environment()
+
+    def test_no_user(self):
+        call_command("manage_superuser")
+
+        # just the consumer user.
+        self.assertEqual(User.objects.count(), 1)
+        self.assertTrue(User.objects.filter(username="consumer").exists())
+
+    def test_create(self):
+        os.environ["PAPERLESS_ADMIN_USER"] = "new_user"
+        os.environ["PAPERLESS_ADMIN_PASSWORD"] = "123456"
+
+        call_command("manage_superuser")
+
+        user: User = User.objects.get_by_natural_key("new_user")
+        self.assertTrue(user.check_password("123456"))
+
+    def test_update(self):
+        os.environ["PAPERLESS_ADMIN_USER"] = "new_user"
+        os.environ["PAPERLESS_ADMIN_PASSWORD"] = "123456"
+
+        call_command("manage_superuser")
+
+        os.environ["PAPERLESS_ADMIN_USER"] = "new_user"
+        os.environ["PAPERLESS_ADMIN_PASSWORD"] = "more_secure_pwd_7645"
+
+        call_command("manage_superuser")
+
+        user: User = User.objects.get_by_natural_key("new_user")
+        self.assertTrue(user.check_password("more_secure_pwd_7645"))
+
+    def test_no_password(self):
+        os.environ["PAPERLESS_ADMIN_USER"] = "new_user"
+
+        call_command("manage_superuser")
+
+        with self.assertRaises(User.DoesNotExist):
+            User.objects.get_by_natural_key("new_user")