Testing django custom management command

Question

I'm struggling with an issue related to a django custom management-command. Here's the custom management-command code to load data from json file:

import time
import json
import sys
from cycu.models import hfModel
from django.db import IntegrityError
from django.core.management import BaseCommand
from django.utils.text import slugify
from django.contrib.gis.geos import Point


class Command(BaseCommand):
    stdin = sys.stdin

    def handle(self, *args, **options):
        start_time = time.time()
        self.stdout.write("Loading Facilities...")
        data = json.load(self.stdin)

        if data:
            hfModel.objects.all().delete()
        instances = []

        for d in data["features"]:
            try:
                if d["properties"]["statename"] == 'Kano':
                    a = hfModel(
                        id=int(d["properties"]["id"]),
                        wardname=d["properties"]["wardname"],
                        wardcode=d["properties"]["wardcode"],
                        lganame=d["properties"]["lganame"],
                        lgacode=int(d["properties"]["lgacode"]),
                        zonename=d["properties"]["zone"],
                        statename=d["properties"]["statename"],
                        source=d["properties"]["source"],
                        ownership=d["properties"]["ownership"],
                        category=d["properties"]["category"],
                        primary_name=d["properties"]["primary_name"],
                        hthfa_code=d["properties"]["hthfa_code"],
                        hthfa_code_slug=slugify(d["properties"]["hthfa_code"]),
                        masterlist_type=d["properties"]["masterlist_type"],
                        point=Point(d["geometry"]["coordinates"])
                    )
                    instances.append(a)
            except IntegrityError as errorDesc:
                print errorDesc  # i.e. to catch duplicates
            except KeyError as keyErrDesc:
                print keyErrDesc  # i.e. filed not exist
                pass
        hfModel.objects.bulk_create(instances)
        print("Facilities loaded in %s seconds." % (time.time() - start_time))

When I run it:

curl 'www.test.org/test.json' | python manage.py load_facilities

everything is ok and have the table has data.

Next step is to create a test. I tried this:

import json
from django.core.management import call_command
from django.test import TestCase
from django.utils.six import StringIO


class AllTheDudesTest(TestCase):
    def test_command_output(self):
        with open('hfs.json', 'r') as data_file:
            json_test_in = json.load(data_file)
        out = StringIO()
        call_command('allTheDudes', stdin=json_test_in, verbosity=3)
        self.assertIn('Facilities loaded in', out.getvalue())

bu when I run it:

coverage run --source='.' manage test

I get this error:

Creating test database for alias 'default'...
Loading Facilities...
E
======================================================================
ERROR: test_command_output (cycu.tests.AllTheDudesTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/jj/projects_dev/ferdas/cycu/tests.py", line 28, in test_command_output
  File "/home/jj/venvs/django_19/local/lib/python2.7/site-packages/django/core/management/__init__.py", line 119, in call_command
    return command.execute(*args, **defaults)
  File "/home/jj/venvs/django_19/local/lib/python2.7/site-packages/django/core/management/base.py", line 399, in execute
    output = self.handle(*args, **options)
  File "/home/jj/projects_dev/ferdas/cycu/management/commands/allTheDudes.py", line 17, in handle
    data = json.load(sys.stdin)
  File "/usr/lib/python2.7/json/__init__.py", line 290, in load
    **kw)
  File "/usr/lib/python2.7/json/__init__.py", line 338, in loads
    return _default_decoder.decode(s)
  File "/usr/lib/python2.7/json/decoder.py", line 366, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/lib/python2.7/json/decoder.py", line 384, in raw_decode
    raise ValueError("No JSON object could be decoded")
ValueError: No JSON object could be decoded

----------------------------------------------------------------------
Ran 1 test in 508.243s

FAILED (errors=1)
Destroying test database for alias 'default'...

Your `hfs.json` is not the correct json format. You need to figure that out. How did you get the `hfs.json` file? — Shang Wang, Dec 22 '15 at 14:17
Yeah it is b/c I'm able to load this file by custom command but I can't test it.... — grillazz, Dec 22 '15 at 14:39
Plus when I add I.e. pprint I'm able to print while json file — grillazz, Dec 22 '15 at 14:40
Then I'm wondering if your `www.test.org/test.json` and `jfs.json` are the same file? — Shang Wang, Dec 22 '15 at 14:49
Hmm, how did you download the file? Did you use wget? Something like `wget 'www.test.org/test.json' -O hfs.json`. — Shang Wang, Dec 22 '15 at 15:08
Or you should directly get the json data: `response = urllib.urlopen("www.test.org/test.json")` and `data = json.loads(response.read())`. — Shang Wang, Dec 22 '15 at 15:09
Could you run call_command in the with statement and pass the file handler `data_file` to stdin instead ? no json could be decoded -> stdin is empty, just a wild guess :) — jpic, Dec 22 '15 at 15:09
@jpic you mean: def test_all_the_dudes(self): out = StringIO() with open('test.json', 'rb') as data_file: command = allTheDudes.Command() command.execute( stdin=data_file, stdout=StringIO(), interactive=False, ) self.assertIn('Facilities loaded in', out.getvalue()) — grillazz, Dec 22 '15 at 15:17
or `class AllTheDudesTest(TestCase): def test_command_output(self): with open('hfs.json', 'r') as data_file: out = StringIO() call_command('allTheDudes', stdin=data_file, verbosity=3) self.assertIn('Facilities loaded in', out.getvalue())` — grillazz, Dec 22 '15 at 15:23

score 7 · Accepted Answer · edited Dec 11 '16 at 17:17

I found it. It was all about moving:
self.stdin = options.get('stdin', sys.stdin) # Used for testing
to the method instead of as an attribute of class Command(BaseCommand):

import time
import json
import sys
from cycu.models import hfModel
from django.db import IntegrityError
from django.core.management import BaseCommand
from django.utils.text import slugify
from django.contrib.gis.geos import Point


class Command(BaseCommand):

    def handle(self, *args, **options):
        stdin = options.get('stdin', sys.stdin)  # Used for testing
        start_time = time.time()
        self.stdout.write("Loading Health Facilities...")
        data = json.load(stdin)

        if data:
            hfModel.objects.all().delete()
        instances = []

        for d in data["features"]:
            try:
                if d["properties"]["statename"] == 'Kano':
                    a = hfModel(
                        id=int(d["properties"]["id"]),
                        wardname=d["properties"]["wardname"],
                        wardcode=d["properties"]["wardcode"],
                        lganame=d["properties"]["lganame"],
                        lgacode=int(d["properties"]["lgacode"]),
                        zonename=d["properties"]["zone"],
                        statename=d["properties"]["statename"],
                        source=d["properties"]["source"],
                        ownership=d["properties"]["ownership"],
                        category=d["properties"]["category"],
                        primary_name=d["properties"]["primary_name"],
                        hthfa_code=d["properties"]["hthfa_code"],
                        hthfa_code_slug=slugify(d["properties"]["hthfa_code"]),
                        masterlist_type=d["properties"]["masterlist_type"],
                        point=Point(d["geometry"]["coordinates"])
                    )
                    instances.append(a)
            except IntegrityError as errorDesc:
                print errorDesc  # i.e. to catch duplicates
            except KeyError as keyErrDesc:
                print keyErrDesc  # i.e. filed not exist
                pass
        hfModel.objects.bulk_create(instances)
        self.stdout.write("Health Facilities loaded in %s seconds." % (time.time() - start_time))

and modifying the test as well:

from django.core.management import call_command
from django.test import TestCase
from django.utils.six import StringIO


class AllTheDudesTest(TestCase):
    def test_command_output(self):
        with open('hfs.json', 'r') as data_file:
            out = StringIO()
            call_command('allTheDudes', stdin=data_file, stdout=out)
            self.assertIn('Health Facilities loaded in', out.getvalue())

Testing django custom management command

1 Answers1