Using Mock with Django

Django has lots of great tools for testing. All of them involve creating a test database before the tests are run. This can be slow.

Python Unittests are another important part of testing. They often are very fast. But what if you want to do unit tests that involve Django model instances? This is where mock comes in. Mock lets you create objects that function like model instances without constructing a database. This Celerity Blog post is a great way to get started.

One problem you may encounter while using mock.patch is this error:

AppRegistryNotReady: Apps aren't loaded yet.

This can be solved by adding this to your test:

import django
django.setup()

You might need to setup some paths and specify a settings file. I do not have to do that because I run the tests from within PyCharm and it does some of that stuff automatically.

Here is a sample of how you might test a Django request that accesses a Django model instance:

import unittest
import json

from django.http import JsonResponse

import mock

def get_contract_items_for_ticket(request):
    """
    AJAX function for loading contract item select based on job
    :param request: Django request object
    :return:
    """
    # Import here to avoid circular imports
    from data.models import Job

    job = Job.objects.get(id=int(request.POST['ticket_job_id']))
    ci_choices = get_contract_items(job.number)
    response = {'choices': ci_choices}
    return JsonResponse(response)


class MyTest(unittest.TestCase):
    import django
    django.setup()

    def setUp(self):
        # Setup code here

    def test(self):
        job_w_contract = mock.Mock()
        job_w_contract.number = '2222-22'
        job_w_contract.id = 22

        job_wo_contract = mock.Mock()
        job_wo_contract.number = '1111-11'
        job_wo_contract.id = 1

        # Function for mocking the .get() method
        def get_job(id=None):
            if id == job_w_contract.id:
                return job_w_contract
            else:
                return job_wo_contract

        # Mock the import statement and test the function
        with mock.patch('data.models.Job') as my_model_mock:
            my_model_mock.objects = mock.Mock()  # mocks Job.objects

            # Make Job.objects.get() use get_job()
            conf = {'get.side_effect': get_job}
            my_model_mock.objects.configure_mock(**conf)

            # Make request object for a job with contract items
            mock_request = mock.Mock()
            mock_request.POST = {'ticket_job_id': str(job_w_contract.id)}

            # Do a request that returns choices
            response = get_contract_items_for_ticket(mock_request)
            choices = json.loads(response.content)['choices']
            self.assertTrue(len(choices) > 1)

            # Do a request that does not return choices
            mock_request.POST = {'ticket_job_id': str(job_wo_contract.id)}
            response = get_contract_items_for_ticket(mock_request)
            choices = json.loads(response.content)['choices']
            self.assertTrue(len(choices) == 1)

Migrating django_manage from Ansible 1.9 to Ansible 2.5

Here is one of the errors (to help you get here from Google):

"msg": "\n:stderr: /usr/bin/env: python\r: No such file or directory\n"

I had an old Django manage.py file that worked for Ansible 1.9. It did not have a shebang (e.g. #!/usr/bin/env python). The error it was generating was:

"msg": "[Errno 8] Exec format error", "rc": 8

Reading the docs, I found this key tidbit:

As of ansible 2.x, your manage.py application must be executable (rwxr-xr-x), and must have a valid shebang, i.e. “#!/usr/bin/env python”, for invoking the appropriate Python interpreter.

That seemed easy enough. But I was not sure what shebang to use since I was using a virtualenv. The quickest way to find out seemed to be use nano to edit manage.py on the server, run the Ansible task that was failing. Repeat until it works.

Then things got weird. Every reasonable shebang lead to the error I started this post with. Also, for no obvious reason, nano started warning me about the file being in DOS format, which is strange since this project has always been in Linux. Maybe when I was cutting and pasting some shebangs into the file I ended up with some DOS whitespace. Running the command “dos2unix manage.py” fixed it.

The shebang that worked was:

#!/usr/bin/env python

UPDATE

It appears that the DOS format chars entered manage.py a long time ago. Maybe when the project was created. I ended up fixing the file in my repo.

 

Receive Text Messages on a Website Using Twilio

In a nutshell:

  1. Setup and get a phone number on Twilio
  2. Configure Twilio
  3. When some one sends a text to the phone number, Twilio packs info about the text in an HTTP POST and posts to your website at the URL you provided when configuring Twilio

Here is a sample of the POST:

{
    u'Body': [u'Hello world!'],
    u'MessageSid': [u'SMxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'],
    u'FromZip': [u'53706'],
    u'SmsStatus': [u'received'],
    u'SmsMessageSid': [u'SMxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'],
    u'AccountSid': [u'AC63444b3f5817d72cbbadb35a71bdd2e9'],
    u'FromCity': [u'MADISON'],
    u'ApiVersion': [u'2010-04-01'],
    u'To': [u'+16089999999'],
    u'From': [u'+16081234567'],
    u'NumMedia': [u'0'],
    u'ToZip': [u'53703'],
    u'ToCountry': [u'US'],
    u'NumSegments': [u'1'],
    u'ToState': [u'WI'],
    u'SmsSid': [u'SMxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'],
    u'ToCity': [u'MADISON'],
    u'FromState': [u'WI'],
    u'FromCountry': [u'US']
}

That should get you started. See the Twilio documentation for details.

Monitoring Django RQ

In my use case, users used a form to put a long running process on the queue. I wanted to make a page that would allow each use to see the status of the jobs they queued. This turned out to be slightly more difficult than it should be.

The first step involved saving the job information to the user’s session:

from datetime import datetime

from django.conf import settings
from django.views.generic import FormView
from django.http import HttpResponseRedirect
from django.core.urlresolvers import reverse

import pytz
import django_rq

class MyView(FormView):
    def form_valid(self, form):
        queue = django_rq.get_queue('low')
        job = queue.enqueue(
            generate_report_func,
            int(form.cleaned_data['year']),
            email_list=form.cleaned_data['recipients']
        )

        # Save queued job to session as a list so that order is preserved
        now = pytz.timezone(settings.TIME_ZONE).localize(datetime.now())
        enqueued_jobs = self.request.session.get('enqueued_jobs', [])
        enqueued_jobs.append({
            'job_id': job._id,
            'name': 'Revenue and Open POs Report {}'.format(form.cleaned_data['year']),
            'started': now.isoformat(),
            'started_for_display': now.strftime('%Y-%m-%d %H:%M'),
            'queue': 'low'
        })
        self.request.session['enqueued_jobs'] = enqueued_jobs
        self.request.session.modified = True
        return HttpResponseRedirect(reverse('on_queue'))

 
Here is the code for the view that allows the user to see the status of each job:

import django_rq
from redis import Redis
from rq.registry import StartedJobRegistry

from django.views.generic import TemplateView

from dateutil.parser import parse


class OnQueueView(TemplateView):
    template_name = 'on_queue.html'

    def get_context_data(self, **kwargs):
        kwargs = super(OnQueueView, self).get_context_data(**kwargs)

        # Make a list of queued jobs
        queue = django_rq.get_queue('low')
        job_status = {x._id: x.status for x in queue.jobs}

        # Make a list of running jobs
        redis_conn = Redis()
        registry = StartedJobRegistry('low', connection=redis_conn)
        for job_id in registry.get_job_ids():
            job_status[job_id] = 'running'

        # Make a list of failed jobs
        for job_id in django_rq.get_failed_queue().job_ids:
            job_status[job_id] = 'failed'

        # Insert status into list of jobs, remove old jobs
        all_jobs = self.request.session.get('enqueued_jobs', [])
        kwargs['jobs'] = []
        now = pytz.timezone(settings.TIME_ZONE).localize(datetime.now())
        self.request.session.modified = False
        for job in all_jobs:
            dt = (now - parse(job['started'])).total_seconds()
            if dt < 3600 * 4:
                if job['job_id'] in job_status:
                    job['status'] = job_status[job['job_id']]
                else:
                    job['status'] = 'completed'
                kwargs['jobs'].append(job)
                self.request.session.modified = True

        if self.request.session.modified:
            self.request.session['enqueued_jobs'] = kwargs['jobs']

        return kwargs

Django Sessions in StaticLiveServerTestCase

This is for Django 1.8.

Here was the scenario. I have a Django site where most of the pages require the user to login. I have a form that gets some initial values from Django sessions. The form has lots of javascript, so live testing seemed best. The problem I could not set the session values in the test code.

Initially, this seemed pretty easy because in the test code, there was a variable called:

self.client.session

However, changes to the session were not present when the test code called the view. Careful inspection showed that the session id’s in self.client.session and view.request.session were different.

Adding the following method to StaticLiveServerTestCase solved the problem:

    def save_to_session(self, key, value):
        cookies = self.selenium.get_cookies()
        session_key = None
        for cookie in cookies:
            if cookie[u'name'] == u'sessionid':
                session_key = cookie[u'value']
                break

        if session_key:
            from django.contrib.sessions.backends.cached_db import SessionStore
            s = SessionStore(session_key)
            s[key] = value
            s.save()

 

Testing Django When Using @cached_property

I use the @cached_property decorator quite a bit. It’s pretty straightforward. Usually, it’s set it and forget it. As noted in the docs, it persists as long as the instance persists.

However, it can cause problems during testing if you want to change it’s value. Here is how to change the value (from SO):

class SomeClass(object):

    @cached_property
    def expensive_property(self):
         return datetime.now()

obj = SomeClass()
print obj.expensive_property
print obj.expensive_property # outputs the same value as before
del obj.expensive_property
print obj.expensive_property # outputs new value