Python fabric.api 模块，env() 实例源码

def test_pull_image(self):
        cases = dict(
            no_errors=dict(
                side_effect=(SucceededResult(), ),
                expected_pull_error=None,
            ),
            error=dict(
                side_effect=(RuntimeError(), ),
                expected_pull_error=True,
            ),
        )
        for case, test_data in cases.items():
            with self.subTest(case=case):
                service = docker.Service(name='service', image='image')
                with mock.patch.object(
                    fabricio,
                    'run',
                    side_effect=test_data['side_effect']
                ):
                    service.pull_image()
                    self.assertEqual(
                        test_data['expected_pull_error'],
                        service.pull_errors.get(fab.env.host),
                    )

def deploy():
    with cd("/data/stregsystem"):
        sudo("systemctl stop apache2.service")
        with settings(sudo_user='stregsystem'):
            sudo("git pull --ff-only")
            with prefix("source /data/stregsystem/env/bin/activate"):
                sudo("pip install -rrequirements.txt")
                sudo("python manage.py collectstatic --noinput")
                sudo("python manage.py migrate")
        sudo("systemctl start apache2.service")

def localhost():
    api.env.hosts = ['localhost']
    global RUNCMD
    RUNCMD = api.local

def package_common_lib():
    RUNCMD("fpm"
           " -s python"      # read from standard Python setup.py
           " --python-pip pip3 --python-bin python3"
           " --python-package-name-prefix python3"
           " -t %(pkgfmt)s"  # set output package format
           " -f"             # overwrite any existing file
           " -p ~/endaga-packages"  # put output in endaga-packages
           " %(fpm_flags)s"  # pass in extra flags
           " ~/common/setup.py" %
           api.env)

def add_mobileStaticResources_parser(parser):
    """
    ?????????
    """
    sub_parser = parser.add_parser("mobileStaticResources", help="????????")

    sub_parser.add_argument(
        "-g", 
        "--game", 
        dest="game", 
        required=True, 
        help="game, eg: gcmob"
    )
    sub_parser.add_argument(
        "-l", 
        "--region", 
        dest="language", 
        required=True, 
        help="region, eg: cn, vn, ft"
    )
    sub_parser.add_argument(
        '--env',
        type=str,
        dest='game_env',
        choices=['prod', 'test'],
        help='Production Environment or Test Environment?'
    )
    sub_parser.add_argument(
        '-t',
        type=str,
        dest='component',
        required=True,
        help='component name, eg: activity_post'
    )
    sub_parser.set_defaults(func=main)

def _format_target():
    return '{app_name}-{env}'.format(**env)

def test_is_manager_returns_false_if_pull_error(self, *args):
        with mock.patch.object(fabricio, 'run') as run:
            service = docker.Service(name='service')
            service.pull_errors[fab.env.host] = True
            self.assertFalse(service.is_manager())
            run.assert_not_called()

def test_is_manager_raises_error_if_all_pulls_failed(self, *args):
        with mock.patch.object(fabricio, 'run') as run:
            service = docker.Service(name='service')
            service.pull_errors[fab.env.host] = True
            with self.assertRaises(docker.ServiceError):
                service.is_manager()
            run.assert_not_called()

def test_infrastructure(self):
        class AbortException(Exception):
            pass

        def task():
            pass

        cases = dict(
            default=dict(
                decorator=tasks.infrastructure,
                expected_infrastructure='task',
            ),
            invoked=dict(
                decorator=tasks.infrastructure(),
                expected_infrastructure='task',
            ),
            with_custom_name=dict(
                decorator=tasks.infrastructure(name='infrastructure'),
                expected_infrastructure='infrastructure',
            ),
        )

        with fab.settings(abort_on_prompts=True, abort_exception=AbortException):
            with mock.patch.object(fab, 'abort', side_effect=AbortException):
                for case, data in cases.items():
                    with self.subTest(case=case):
                        decorator = data['decorator']
                        infrastructure = decorator(task)

                        self.assertTrue(is_task_object(infrastructure.confirm))
                        self.assertTrue(is_task_object(infrastructure.default))

                        fab.execute(infrastructure.confirm)
                        self.assertEqual(data['expected_infrastructure'], fab.env.infrastructure)

                        fab.env.infrastructure = None
                        with mock.patch.object(console, 'confirm', side_effect=[True, False]):
                            fab.execute(infrastructure.default)
                            self.assertEqual(data['expected_infrastructure'], fab.env.infrastructure)
                            with self.assertRaises(AbortException):
                                fab.execute(infrastructure.default)

def _run_task(task, args=None, kwargs=None):
    from fabfile import targets
    if targets:
        for target in targets:
            env = _build_env(target)
            if env:
                with fab_settings(** env):
                    _run_task_core(task, args, kwargs)
    else:
        _run_task_core(task, args, kwargs)

def env():
    print('creating env')
    run('virtualenv env')

def newsetup():
    execute(dependencies)
    execute(install_mysql)
    execute(install_nginx)
    execute(env)

def test_options(self):
        cases = dict(
            default=dict(
                kwargs=dict(),
                expected={},
            ),
            custom=dict(
                kwargs=dict(options=dict(foo='bar')),
                expected={
                    'foo': 'bar',
                },
            ),
            collision=dict(
                kwargs=dict(options=dict(execute='execute')),
                expected={
                    'execute': 'execute',
                },
            ),
            override=dict(
                kwargs=dict(options=dict(env='custom_env')),
                expected={
                    'env': 'custom_env',
                },
            ),
            complex=dict(
                kwargs=dict(options=dict(
                    env='custom_env',
                    user=lambda service: 'user',
                    foo='foo',
                    bar=lambda service: 'bar',
                )),
                expected={
                    'env': 'custom_env',
                    'user': 'user',
                    'foo': 'foo',
                    'bar': 'bar',
                },
            ),
        )
        for case, data in cases.items():
            with self.subTest(case=case):
                container = TestContainer(**data['kwargs'])
                self.assertDictEqual(data['expected'], dict(container.options))