[improvement] Add Swapper #86

Closes #86

.travis.yml

@@ -15,6 +15,8 @@ addons:
 env:
   - DJANGO="django>=2.2,<3.0"
   - DJANGO="django>=3.0,<3.1"
+  - SAMPLE_APP=1 DJANGO="django>=2.2,<3.0"
+  - SAMPLE_APP=1 DJANGO="django>=3.0,<3.1"
 
 branches:
   only:
@@ -32,7 +34,21 @@ install:
   - python setup.py -q develop
 
 before_script:
-  - openwisp-utils-qa-checks --migrations-to-ignore 4 --migration-path ./django_x509/migrations/ --migration-module django_x509
+  - |
+    if [[ $SAMPLE_APP != "1" ]]; then
+      openwisp-utils-qa-checks --migrations-to-ignore 4 \
+                               --migration-path ./django_x509/migrations/ \
+                               --migration-module django_x509
+    else
+      openwisp-utils-qa-checks --skip-isort \
+                               --skip-flake8 \
+                               --skip-checkmigrations \
+                               --skip-checkendline \
+                               --skip-checkcommit \
+                               --migration-path ./tests/openwisp2/sample_x509/migrations/ \
+                               --migration-module sample_x509
+    fi
+
 
 script:
   - jslint django_x509/static/django-x509/js/*.js

README.rst

@@ -327,6 +327,11 @@ be protected with authentication or not.
 Extending django-x509
 ---------------------
 
+The django app ``tests/openwisp2/sample_x509/`` adds some changes on
+top of the ``django-x509`` module with the sole purpose of testing the
+module's extensibility. It can be used as a sample for extending
+``django-x509`` functionality in your own application.
+
 *django-x509* provides a set of models and admin classes which can be imported,
 extended and reused by third party apps.
 
@@ -394,6 +399,17 @@ Add ``openwisp_utils.loaders.DependencyLoader`` to ``TEMPLATES`` in your ``setti
         }
     ]
 
+5. Add swapper configurations
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Add the following to your ``settings.py``:
+
+.. code-block:: python
+
+    # Setting models for swapper module
+    DJANGO_X509_CA_MODEL = 'YOUR_MODULE_NAME.Ca'
+    DJANGO_X509_CERT_MODEL = 'YOUR_MODULE_NAME.Cert'
+
 Extending models
 ~~~~~~~~~~~~~~~~
 

django_x509/admin.py

@@ -1,7 +1,10 @@
 from django.contrib import admin
+from swapper import load_model
 
 from .base.admin import AbstractCaAdmin, AbstractCertAdmin
-from .models import Ca, Cert
+
+Ca = load_model('django_x509', 'Ca')
+Cert = load_model('django_x509', 'Cert')
 
 
 class CertAdmin(AbstractCertAdmin):

django_x509/base/admin.py

@@ -81,14 +81,14 @@ def get_context(self, data, ca_count=0, cert_count=0):
                 'title': _('Renew selected CAs'),
                 'ca_count': ca_count,
                 'cert_count': cert_count,
-                'cancel_url': 'django_x509_ca_changelist',
+                'cancel_url': f'{self.opts.app_label}_ca_changelist',
                 'action': 'renew_ca'
             })
         else:
             context.update({
                 'title': _('Renew selected certs'),
                 'cert_count': cert_count,
-                'cancel_url': 'django_x509_cert_changelist',
+                'cancel_url': f'{self.opts.app_label}_cert_changelist',
                 'action': 'renew_cert'
             })
         context.update({
@@ -211,7 +211,7 @@ class Media:
 
     def ca_url(self, obj):
         url = reverse('admin:{0}_ca_change'.format(self.opts.app_label), args=[obj.ca.pk])
-        return format_html("<a href='{url}'>{text}</a>",
+        return format_html('<a href="{url}">{text}</a>',
                            url=url,
                            text=obj.ca.name)
     ca_url.short_description = 'CA'

django_x509/base/models.py

@@ -3,6 +3,7 @@
 from datetime import datetime, timedelta
 
 import OpenSSL
+import swapper
 from django.core.exceptions import ValidationError
 from django.db import models
 from django.utils import timezone
@@ -153,7 +154,7 @@ class BaseX509(models.Model):
     modified = AutoLastModifiedField(_('modified'), editable=True)
     passphrase = models.CharField(max_length=64,
                                   blank=True,
-                                  help_text=_("Passphrase for the private key, if present"))
+                                  help_text=_('Passphrase for the private key, if present'))
 
     class Meta:
         abstract = True
@@ -239,9 +240,9 @@ def _validate_pem(self):
                     kwargs['passphrase'] = getattr(self, 'passphrase').encode('utf8')
                 load_pem(*args, **kwargs)
             except OpenSSL.crypto.Error as e:
-                error = 'OpenSSL error: <br>{0}'.format(str(e.args[0]).replace('), ', '), <br>').strip("[]"))
-                if "bad decrypt" in error:
-                    error = "<b>Incorrect Passphrase</b> <br>" + error
+                error = 'OpenSSL error: <br>{0}'.format(str(e.args[0]).replace('), ', '), <br>').strip('[]'))
+                if 'bad decrypt' in error:
+                    error = '<b>Incorrect Passphrase</b> <br>' + error
                     errors['passphrase'] = ValidationError(_(mark_safe(error)))
                     continue
                 errors[field] = ValidationError(_(mark_safe(error)))
@@ -284,13 +285,13 @@ def _generate(self):
         cert.set_pubkey(key)
         cert = self._add_extensions(cert)
         cert.sign(issuer_key, str(self.digest))
-        self.certificate = crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode("utf-8")
+        self.certificate = crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode('utf-8')
         key_args = (crypto.FILETYPE_PEM, key)
         key_kwargs = {}
         if self.passphrase:
             key_kwargs['passphrase'] = self.passphrase.encode('utf-8')
             key_kwargs['cipher'] = 'DES-EDE3-CBC'
-        self.private_key = crypto.dump_privatekey(*key_args, **key_kwargs).decode("utf-8")
+        self.private_key = crypto.dump_privatekey(*key_args, **key_kwargs).decode('utf-8')
 
     def _fill_subject(self, subject):
         """
@@ -491,7 +492,9 @@ class AbstractCert(BaseX509):
     """
     Abstract Cert model
     """
-    ca = models.ForeignKey('django_x509.Ca', on_delete=models.CASCADE, verbose_name=_('CA'))
+    ca = models.ForeignKey(swapper.get_model_name('django_x509', 'Ca'),
+                           on_delete=models.CASCADE,
+                           verbose_name=_('CA'))
     revoked = models.BooleanField(_('revoked'),
                                   default=False)
     revoked_at = models.DateTimeField(_('revoked at'),

django_x509/models.py

@@ -1,3 +1,5 @@
+import swapper
+
 from .base.models import AbstractCa, AbstractCert
 
 
@@ -7,6 +9,7 @@ class Ca(AbstractCa):
     """
     class Meta(AbstractCa.Meta):
         abstract = False
+        swappable = swapper.swappable_setting('django_x509', 'Ca')
 
 
 class Cert(AbstractCert):
@@ -15,3 +18,4 @@ class Cert(AbstractCert):
     """
     class Meta(AbstractCert.Meta):
         abstract = False
+        swappable = swapper.swappable_setting('django_x509', 'Cert')

django_x509/tests/__init__.py

@@ -1,45 +0,0 @@
-"""
-test utilities shared among test classes
-these mixins are reused also in openwisp2
-change with care.
-"""
-
-
-class TestX509Mixin(object):
-    def _create_ca(self, **kwargs):
-        options = dict(name='Test CA',
-                       key_length='2048',
-                       digest='sha256',
-                       country_code='IT',
-                       state='RM',
-                       city='Rome',
-                       organization_name='OpenWISP',
-                       email='test@test.com',
-                       common_name='openwisp.org',
-                       extensions=[])
-        options.update(kwargs)
-        ca = self.ca_model(**options)
-        ca.full_clean()
-        ca.save()
-        return ca
-
-    def _create_cert(self, **kwargs):
-        options = dict(name='TestCert',
-                       ca=None,
-                       key_length='2048',
-                       digest='sha256',
-                       country_code='IT',
-                       state='RM',
-                       city='Rome',
-                       organization_name='Test',
-                       email='test@test.com',
-                       common_name='openwisp.org',
-                       extensions=[])
-        options.update(kwargs)
-        # auto create CA if not supplied
-        if not options.get('ca'):
-            options['ca'] = self._create_ca()
-        cert = self.cert_model(**options)
-        cert.full_clean()
-        cert.save()
-        return cert

django_x509/tests/base/__init__.py

@@ -0,0 +1,56 @@
+from django.contrib.messages.storage.fallback import FallbackStorage
+from django.http import HttpRequest
+
+
+class MessagingRequest(HttpRequest):
+    session = 'session'
+
+    def __init__(self):
+        super().__init__()
+        self._messages = FallbackStorage(self)
+
+    def get_messages(self):
+        return getattr(self._messages, '_queued_messages')
+
+    def get_message_strings(self):
+        return [str(m) for m in self.get_messages()]
+
+
+class TestX509Mixin(object):
+    def _create_ca(self, **kwargs):
+        options = dict(name='Test CA',
+                       key_length='2048',
+                       digest='sha256',
+                       country_code='IT',
+                       state='RM',
+                       city='Rome',
+                       organization_name='OpenWISP',
+                       email='test@test.com',
+                       common_name='openwisp.org',
+                       extensions=[])
+        options.update(kwargs)
+        ca = self.ca_model(**options)
+        ca.full_clean()
+        ca.save()
+        return ca
+
+    def _create_cert(self, **kwargs):
+        options = dict(name='TestCert',
+                       ca=None,
+                       key_length='2048',
+                       digest='sha256',
+                       country_code='IT',
+                       state='RM',
+                       city='Rome',
+                       organization_name='Test',
+                       email='test@test.com',
+                       common_name='openwisp.org',
+                       extensions=[])
+        options.update(kwargs)
+        # auto create CA if not supplied
+        if not options.get('ca'):
+            options['ca'] = self._create_ca()
+        cert = self.cert_model(**options)
+        cert.full_clean()
+        cert.save()
+        return cert