Python django.core.management 模块，call_command() 实例源码

我们从Python开源项目中，提取了以下47个代码示例，用于说明如何使用django.core.management.call_command()。

项目：DjangoCMS 作者：farhan711 | 项目源码 | 文件源码

def test_copy_filled_placeholder(self):
        """
        If an existing title in the target language has plugins in a placeholder
        that placeholder is skipped
        """
        site = 1
        number_start_plugins = CMSPlugin.objects.all().count()

        # create an empty title language
        root_page = Page.objects.on_site(site).get_home()
        create_title("de", "root page de", root_page)
        ph = root_page.placeholders.get(slot="body")
        add_plugin(ph, "TextPlugin", "de", body="Hello World")

        out = StringIO()
        management.call_command(
            'cms', 'copy', 'lang', '--from-lang=en', '--to-lang=de', interactive=False, stdout=out
        )

        self.assertEqual(CMSPlugin.objects.filter(language='en').count(), number_start_plugins)
        # one placeholder (with 7 plugins) is skipped, so the difference must be 6
        self.assertEqual(CMSPlugin.objects.filter(language='de').count(), number_start_plugins-6)

项目：tts-bug-bounty-dashboard 作者：18F | 项目源码 | 文件源码

def call_runscheduler(loops=1, mock_call_command=None):
    ctx = {'sleep_count': 0}

    def fake_sleep(seconds):
        ctx['sleep_count'] += 1
        if ctx['sleep_count'] > loops:
            raise KeyboardInterrupt()

    if mock_call_command is None:
        mock_call_command = mock.MagicMock()

    with mock.patch.object(runscheduler, 'call_command', mock_call_command):
        with mock.patch.object(runscheduler, 'logger') as mock_logger:
            with mock.patch('time.sleep', fake_sleep):
                with pytest.raises(KeyboardInterrupt):
                    call_command('runscheduler')
            return mock_call_command, mock_logger

项目：wagtail-bakery 作者：moorinteractive | 项目源码 | 文件源码

def test_page_has_redirect_response(site):
    redirect_page = RedirectPage.objects.create(
        depth=1,
        path='0002',
        title='Page',
        slug='page',
    )

    # Make redirect page the root page
    site.root_page = redirect_page
    site.save()

    # Build static files
    management.call_command('build', '--skip-static', '--skip-media')
    assert os.path.exists(os.path.join(settings.BUILD_DIR, 'index.html'))

    # Check if meta tag is present
    content = open(os.path.join(settings.BUILD_DIR, 'index.html')).read()
    assert '<meta http-equiv="refresh" content="1; url=http://www.example.com/">' in content # noqa

项目：django-gitversions 作者：michaelkuty | 项目源码 | 文件源码

def test_04_versioner(self):

        # remove local files
        #versioner.backend.destroy()

        management.call_command('gitversions',
                                format='json',
                                indent=4)

        # check local uncomited changes
        self.assertEqual(versioner.backend.check(), True)

        versioner.backend.commit('Initial Commit')

        self.assertEqual(versioner.backend.check(), False)

        # cleanup

项目：vaultier 作者：Movile | 项目源码 | 文件源码

def handle(self, *args, **options):
        print ">>> Initializing your database"
        try:
            management.call_command('syncdb')
            management.call_command('migrate')
            try:
                # do we need cache table?
                cache.get('', None)
            except ProgrammingError:
                # yes we do
                management.call_command('createcachetable', 'vaultier_cache')

            # public static files
            management.call_command('collectstatic', interactive=False)
        except OperationalError as e:
            msg = ">>> Your DB is not configured correctly: {}"
            print msg.format(e.message)
        else:
            if options.get('no_statistics'):
                task_statistics_collector.delay()
            print (">>> DB is initialized, you can now try to run Vaultier "
                   "using 'vaultier runserver'")

项目：oim-cms 作者：parksandwildlife | 项目源码 | 文件源码

def user_logged_in_handler(sender, request, user, **kwargs):
    logging.debug('user_logged_in_handler')
    request.session.save()
    usersession, created = UserSession.objects.get_or_create(user=user, session_id=request.session.session_key)
    usersession.ip = get_ip(request)
    if DepartmentUser.objects.filter(email__iexact=user.email).exists():
        logging.debug('user_logged_in_handler departmentuser {}'.format(user.email))
        usersession.department_user = DepartmentUser.objects.filter(email__iexact=user.email)[0]
        if (user.username != usersession.department_user.username):
            test = get_user_model().objects.filter(username=usersession.department_user.username)
            if test.exists():
                test.delete()
            user.username = usersession.department_user.username
            user.save()
    usersession.save()
    logging.debug('user_logged_in_handler saving stuff')
    management.call_command("clearsessions", verbosity=0)

项目：tissuelab 作者：VirtualPlants | 项目源码 | 文件源码

def test_clearsessions_command(self):
        """
        Test clearsessions command for clearing expired sessions.
        """
        self.assertEqual(0, Session.objects.count())

        # One object in the future
        self.session['foo'] = 'bar'
        self.session.set_expiry(3600)
        self.session.save()

        # One object in the past
        other_session = self.backend()
        other_session['foo'] = 'bar'
        other_session.set_expiry(-3600)
        other_session.save()

        # Two sessions are in the database before clearsessions...
        self.assertEqual(2, Session.objects.count())
        management.call_command('clearsessions')
        # ... and one is deleted.
        self.assertEqual(1, Session.objects.count())

项目：openkamer 作者：openkamer | 项目源码 | 文件源码

def create_json_dump():
        filepath = os.path.join(settings.DBBACKUP_STORAGE_OPTIONS['location'], 'openkamer-' + str(datetime.date.today()) + '.json')
        filepath_compressed = filepath + '.gz'
        with open(filepath, 'w') as fileout:
            management.call_command(
                'dumpdata',
                '--all',
                '--natural-foreign',
                '--exclude', 'auth.permission',
                '--exclude', 'contenttypes',
                'person',
                'parliament',
                'government',
                'document',
                'stats',
                'website',
                stdout=fileout
            )
        with open(filepath, 'rb') as f_in:
            with gzip.open(filepath_compressed, 'wb') as f_out:
                shutil.copyfileobj(f_in, f_out)
        os.remove(filepath)
        BackupDaily.remove_old_json_dumps(days_old=30)