测试Django自定义管理命令

Question

我正在努力解决与django自定义管理命令相关的问题。这是用于从json文件加载数据的自定义管理命令代码：

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))

当我运行它时：

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

一切都很好，并且表中有数据。

下一步是创建测试。我尝试了这个：

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())

当我运行它时：

coverage run --source='.' manage test

我收到此错误：

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'...

Answer 1

我找到了。这完全是关于移动：
self.stdin = options.get('stdin', sys.stdin) # Used for testing
到方法而不是作为 的属性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))

并修改测试：

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())