django - 测试 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())

bu 当我运行它时:

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