Best Python code snippet using autotest_python
test_tasks.py
Source:test_tasks.py  
1import shlex2import sys3import mock4import six5import unittest2 as unittest6from fabric import api as fab, state7from fabric.contrib import console8from fabric.main import load_tasks_from_module, is_task_module, is_task_object9import fabricio10import fabricio.tasks11from fabricio import docker, tasks, utils12from tests import docker_build_args_parser, SucceededResult13class TestContainer(docker.Container):14    image = docker.Image('test')15class TestCase(unittest.TestCase):16    def test_infrastructure(self):17        class AbortException(Exception):18            pass19        def task():20            pass21        cases = dict(22            default=dict(23                decorator=tasks.infrastructure,24                expected_infrastructure='task',25            ),26            invoked=dict(27                decorator=tasks.infrastructure(),28                expected_infrastructure='task',29            ),30        )31        with fab.settings(abort_on_prompts=True, abort_exception=AbortException):32            with mock.patch.object(fab, 'abort', side_effect=AbortException):33                for case, data in cases.items():34                    with self.subTest(case=case):35                        decorator = data['decorator']36                        infrastructure = decorator(task)37                        self.assertTrue(is_task_object(infrastructure.confirm))38                        self.assertTrue(is_task_object(infrastructure.default))39                        fab.execute(infrastructure.confirm)40                        self.assertEqual(data['expected_infrastructure'], fab.env.infrastructure)41                        fab.env.infrastructure = None42                        with mock.patch.object(console, 'confirm', side_effect=[True, False]):43                            fab.execute(infrastructure.default)44                            self.assertEqual(data['expected_infrastructure'], fab.env.infrastructure)45                            with self.assertRaises(AbortException):46                                fab.execute(infrastructure.default)47    def test_infrastructure_details(self):48        class Infrastructure(tasks.Infrastructure):49            """50            inf {name}51            """52        @Infrastructure53        def infrastructure(foo='bar'):54            """55            func56            """57        self.assertListEqual(58            ['inf', 'infrastructure', 'func', 'Arguments:', "foo='bar'"],59            infrastructure.__details__().split(),60        )61    def test_get_task_name(self):62        cases = dict(63            case1=dict(64                command='command1',65                expected='layer1.layer2.command1',66            ),67            case2=dict(68                command='command2',69                expected='layer1.layer2.command2',70            ),71            case3=dict(72                command='command3',73                expected='layer1.command3',74            ),75        )76        commands = dict(77            layer1=dict(78                layer2=dict(79                    command1='command1',80                    command2='command2',81                ),82                command3='command3',83            ),84        )85        with utils.patch(state, 'commands', commands):86            for case, data in cases.items():87                with self.subTest(case=case):88                    self.assertEqual(89                        tasks.get_task_name(data['command']),90                        data['expected'],91                    )92class TasksTestCase(unittest.TestCase):93    def test_tasks(self):94        class TestTasks(tasks.Tasks):95            @fab.task(default=True, aliases=['foo', 'bar'])96            def default(self):97                pass98            @fab.task(name='name', alias='alias')99            def task(self):100                pass101            @fab.task102            @fab.serial103            def serial(self):104                pass105            @fab.task106            @fab.parallel107            def parallel(self):108                pass109        roles = ['role_1', 'role_2']110        hosts = ['host_1', 'host_2']111        tasks_list = TestTasks(roles=roles, hosts=hosts)112        self.assertTrue(is_task_module(tasks_list))113        self.assertTrue(tasks_list.default.is_default)114        self.assertListEqual(['foo', 'bar'], tasks_list.default.aliases)115        self.assertEqual('name', tasks_list.task.name)116        self.assertListEqual(['alias'], tasks_list.task.aliases)117        for task in tasks_list:118            self.assertListEqual(roles, task.roles)119            self.assertListEqual(hosts, task.hosts)120        docstring, new_style, classic, default = load_tasks_from_module(tasks_list)121        self.assertIsNone(docstring)122        self.assertIn('default', new_style)123        self.assertIn('alias', new_style)124        self.assertIn('foo', new_style)125        self.assertIn('bar', new_style)126        self.assertIn('name', new_style)127        self.assertDictEqual({}, classic)128        self.assertIs(tasks_list.default, default)129        self.assertIn('serial', tasks_list.serial.wrapped.__dict__)130        self.assertTrue(tasks_list.serial.wrapped.serial)131        self.assertIn('parallel', tasks_list.parallel.wrapped.__dict__)132        self.assertTrue(tasks_list.parallel.wrapped.parallel)133    def test_super(self):134        class TestTasks1(tasks.Tasks):135            @fab.task136            def task(self):137                pass138            @fab.task139            def task2(self):140                pass141            @fab.task142            def task3(self, argument):143                pass144        class TestTasks2(TestTasks1):145            @fab.task146            def task(self):147                super(TestTasks2, self).task()148        roles = ['role_1', 'role_2']149        hosts = ['host_1', 'host_2']150        tasks_list = TestTasks2(roles=roles, hosts=hosts)151        self.assertListEqual(roles, tasks_list.task.roles)152        self.assertListEqual(hosts, tasks_list.task.hosts)153        self.assertIsNone(getattr(super(TestTasks2, tasks_list).task, 'roles', None))154        self.assertIsNone(getattr(super(TestTasks2, tasks_list).task, 'hosts', None))155        with fab.settings(fab.hide('everything')):156            fab.execute(tasks_list.task)157            # check if there is enough arguments passed to methods158            tasks_list.task()159            tasks_list.task2()160            tasks_list.task3('argument')161            super(TestTasks2, tasks_list).task()162            super(TestTasks2, tasks_list).task3('argument')163class DockerTasksTestCase(unittest.TestCase):164    maxDiff = None165    def setUp(self):166        self.stderr, sys.stderr = sys.stderr, six.StringIO()167        self.fab_settings = fab.settings(fab.hide('everything'))168        self.fab_settings.__enter__()169    def tearDown(self):170        self.fab_settings.__exit__(None, None, None)171        sys.stderr = self.stderr172    def test_commands_list(self):173        cases = dict(174            default=dict(175                init_kwargs=dict(service='service'),176                expected_commands={'deploy'},177            ),178            prepare_tasks=dict(179                init_kwargs=dict(service='service', registry='registry', prepare_command=True, push_command=True, upgrade_command=True),180                expected_commands={'deploy', 'prepare', 'push', 'upgrade'},181            ),182            migrate_tasks=dict(183                init_kwargs=dict(service='service', migrate_commands=True),184                expected_commands={'deploy', 'migrate', 'migrate-back'},185            ),186            backup_tasks=dict(187                init_kwargs=dict(service='service', backup_commands=True),188                expected_commands={'deploy', 'backup', 'restore'},189            ),190            revert_task=dict(191                init_kwargs=dict(service='service', revert_command=True),192                expected_commands={'deploy', 'revert'},193            ),194            pull_task=dict(195                init_kwargs=dict(service='service', pull_command=True),196                expected_commands={'deploy', 'pull'},197            ),198            update_task=dict(199                init_kwargs=dict(service='service', update_command=True),200                expected_commands={'deploy', 'update'},201            ),202            destroy_task=dict(203                init_kwargs=dict(service='service', destroy_command=True),204                expected_commands={'deploy', 'destroy'},205            ),206            complex=dict(207                init_kwargs=dict(service='service', backup_commands=True, migrate_commands=True, registry='registry', revert_command=True, update_command=True, pull_command=True, destroy_command=True),208                expected_commands={'pull', 'deploy', 'update', 'backup', 'restore', 'migrate', 'migrate-back', 'revert', 'destroy'},209            ),210            task_mode=dict(211                init_kwargs=dict(service='service', backup_commands=True, migrate_commands=True, registry='registry', revert_command=True, update_command=True, pull_command=True),212                expected_commands={'pull', 'rollback', 'update', 'backup', 'restore', 'migrate', 'migrate-back', 'prepare', 'push', 'revert', 'upgrade', 'deploy', 'destroy', 'deploy'},213                env=dict(tasks='task'),214            ),215        )216        for case, data in cases.items():217            with self.subTest(case=case):218                with mock.patch.dict(fab.env, data.get('env', {})):219                    tasks_list = tasks.DockerTasks(**data['init_kwargs'])220                docstring, new_style, classic, default = load_tasks_from_module(tasks_list)221                self.assertSetEqual(set(new_style), data['expected_commands'])222    @mock.patch.object(docker.Container, 'destroy')223    @mock.patch.object(console, 'confirm', return_value=True)224    @mock.patch.dict(fab.env, dict(tasks='task'))225    def test_destroy(self, confirm, destroy):226        service = docker.Container(name='name')227        tasks_list = tasks.DockerTasks(service=service)228        cases = dict(229            explicit=dict(230                execute=tasks_list.destroy,231                expected_calls=[mock.call.destroy('args', kwargs='kwargs')],232            ),233            default=dict(234                execute=tasks_list.destroy.default,235                expected_calls=[236                    mock.call.confirm(mock.ANY, default=mock.ANY),237                    mock.call.destroy('args', kwargs='kwargs'),238                ],239            ),240            confirm=dict(241                execute=tasks_list.destroy.confirm,242                expected_calls=[mock.call.destroy('args', kwargs='kwargs')],243            ),244        )245        calls = mock.Mock()246        calls.attach_mock(destroy, 'destroy')247        calls.attach_mock(confirm, 'confirm')248        for case, data in cases.items():249            with self.subTest(case):250                calls.reset_mock()251                fab.execute(data['execute'], 'args', kwargs='kwargs')252                self.assertListEqual(data['expected_calls'], calls.mock_calls)253    @mock.patch.dict(fab.env, dict(tasks='task'))254    def test_destroy_details(self):255        class Service(docker.Container):256            def destroy(self, foo='bar'):257                """258                service doc259                """260        class DockerTasks(tasks.DockerTasks):261            class DestroyTask(tasks.DockerTasks.DestroyTask):262                """263                tasks doc264                """265        self.assertEqual(266            "\ntasks doc\n\nservice doc\n\nArguments: self, foo='bar'",267            DockerTasks(Service()).destroy.__details__(),268        )269    @mock.patch.multiple(TestContainer, revert=mock.DEFAULT, migrate_back=mock.DEFAULT)270    def test_rollback(self, revert, migrate_back):271        tasks_list = tasks.DockerTasks(service=TestContainer(), hosts=['host'])272        rollback = mock.Mock()273        rollback.attach_mock(migrate_back, 'migrate_back')274        rollback.attach_mock(revert, 'revert')275        revert.return_value = True276        # with migrate_back disabled277        tasks_list.rollback.name = '{0}__migrate_disabled'.format(self)278        fab.execute(tasks_list.rollback, migrate_back='no')279        migrate_back.assert_not_called()280        revert.assert_called_once()281        rollback.reset_mock()282        # default case283        tasks_list.rollback.name = '{0}__default'.format(self)284        fab.execute(tasks_list.rollback)285        self.assertListEqual(286            [mock.call.migrate_back(), mock.call.revert()],287            rollback.mock_calls,288        )289        rollback.reset_mock()290    @mock.patch.multiple(docker.Container, backup=mock.DEFAULT, migrate=mock.DEFAULT, update=mock.DEFAULT)291    @mock.patch.multiple(fabricio, run=mock.DEFAULT, local=mock.DEFAULT)292    @mock.patch.object(fab, 'remote_tunnel', return_value=mock.MagicMock())293    def test_deploy(self, remote_tunnel, run, local, backup, migrate, update):294        cases = dict(295            # TODO empty registry296            # TODO empty account297            default=dict(298                deploy_kwargs=dict(),299                init_kwargs=dict(),300                expected_calls=[301                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),302                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),303                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),304                    mock.call.migrate(tag=None, registry=None, account=None),305                    mock.call.update(force=False, tag=None, registry=None, account=None),306                ],307                image_registry=None,308            ),309            custom_image_registry=dict(310                deploy_kwargs=dict(),311                init_kwargs=dict(),312                expected_calls=[313                    mock.call.run('docker tag registry:5000/test:latest fabricio-temp-image:test && docker rmi registry:5000/test:latest', ignore_errors=True, use_cache=False),314                    mock.call.run('docker pull registry:5000/test:latest', quiet=False, use_cache=False, ignore_errors=False),315                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),316                    mock.call.migrate(tag=None, registry=None, account=None),317                    mock.call.update(force=False, tag=None, registry=None, account=None),318                ],319                image_registry='registry:5000',320            ),321            custom_image_registry_with_ssh_tunnel_deprecated=dict(322                deploy_kwargs=dict(),323                init_kwargs=dict(ssh_tunnel_port=1234),324                expected_calls=[325                    mock.call.run('docker tag localhost:1234/test:latest fabricio-temp-image:test && docker rmi localhost:1234/test:latest', ignore_errors=True, use_cache=False),326                    mock.call.run('docker pull localhost:1234/test:latest', quiet=False, use_cache=False, ignore_errors=False),327                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),328                    mock.call.migrate(tag=None, registry='localhost:1234', account=None),329                    mock.call.update(force=False, tag=None, registry='localhost:1234', account=None),330                ],331                image_registry='registry:5000',332            ),333            custom_registry=dict(334                deploy_kwargs=dict(),335                init_kwargs=dict(registry='host:5000'),336                expected_calls=[337                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),338                    mock.call.local('docker pull test:latest', quiet=False, use_cache=True, ignore_errors=False),339                    mock.call.local('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=True),340                    mock.call.local('docker tag test:latest host:5000/test:latest', use_cache=True),341                    mock.call.local('docker push host:5000/test:latest', quiet=False, use_cache=True),342                    mock.call.local('docker rmi host:5000/test:latest', use_cache=True),343                    mock.call.run('docker tag host:5000/test:latest fabricio-temp-image:test && docker rmi host:5000/test:latest', ignore_errors=True, use_cache=False),344                    mock.call.run('docker pull host:5000/test:latest', quiet=False, use_cache=False, ignore_errors=False),345                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),346                    mock.call.migrate(tag=None, registry='host:5000', account=None),347                    mock.call.update(force=False, tag=None, registry='host:5000', account=None),348                ],349                image_registry=None,350            ),351            custom_registry_no_image=dict(352                deploy_kwargs=dict(),353                init_kwargs=dict(registry='host:5000'),354                expected_calls=[355                    mock.call.migrate(tag=None, registry='host:5000', account=None),356                    mock.call.update(force=False, tag=None, registry='host:5000', account=None),357                ],358                image_registry=None,359                service=docker.Container(name='name'),360            ),361            custom_registry_with_ssh_tunnel_deprecated=dict(362                deploy_kwargs=dict(),363                init_kwargs=dict(registry='host:5000', ssh_tunnel_port=1234),364                expected_calls=[365                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),366                    mock.call.local('docker pull test:latest', quiet=False, use_cache=True, ignore_errors=False),367                    mock.call.local('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=True),368                    mock.call.local('docker tag test:latest host:5000/test:latest', use_cache=True),369                    mock.call.local('docker push host:5000/test:latest', quiet=False, use_cache=True),370                    mock.call.local('docker rmi host:5000/test:latest', use_cache=True),371                    mock.call.run('docker tag localhost:1234/test:latest fabricio-temp-image:test && docker rmi localhost:1234/test:latest', ignore_errors=True, use_cache=False),372                    mock.call.run('docker pull localhost:1234/test:latest', quiet=False, use_cache=False, ignore_errors=False),373                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),374                    mock.call.migrate(tag=None, registry='localhost:1234', account=None),375                    mock.call.update(force=False, tag=None, registry='localhost:1234', account=None),376                ],377                image_registry=None,378            ),379            custom_registry_with_ssh_tunnel_no_image_deprecated=dict(380                deploy_kwargs=dict(),381                init_kwargs=dict(registry='host:5000', ssh_tunnel_port=1234),382                expected_calls=[383                    mock.call.migrate(tag=None, registry='localhost:1234', account=None),384                    mock.call.update(force=False, tag=None, registry='localhost:1234', account=None),385                ],386                image_registry=None,387                service=docker.Container(name='name'),388            ),389            custom_account=dict(390                deploy_kwargs=dict(),391                init_kwargs=dict(account='account'),392                expected_calls=[393                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),394                    mock.call.local('docker pull test:latest', quiet=False, use_cache=True, ignore_errors=False),395                    mock.call.local('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=True),396                    mock.call.local('docker tag test:latest account/test:latest', use_cache=True),397                    mock.call.local('docker push account/test:latest', quiet=False, use_cache=True),398                    mock.call.local('docker rmi account/test:latest', use_cache=True),399                    mock.call.run('docker tag account/test:latest fabricio-temp-image:test && docker rmi account/test:latest', ignore_errors=True, use_cache=False),400                    mock.call.run('docker pull account/test:latest', quiet=False, use_cache=False, ignore_errors=False),401                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),402                    mock.call.migrate(tag=None, registry=None, account='account'),403                    mock.call.update(force=False, tag=None, registry=None, account='account'),404                ],405                image_registry=None,406            ),407            custom_account_no_image=dict(408                deploy_kwargs=dict(),409                init_kwargs=dict(account='account'),410                expected_calls=[411                    mock.call.migrate(tag=None, registry=None, account='account'),412                    mock.call.update(force=False, tag=None, registry=None, account='account'),413                ],414                image_registry=None,415                service=docker.Container(name='name'),416            ),417            custom_account_with_ssh_tunnel_deprecated=dict(418                deploy_kwargs=dict(),419                init_kwargs=dict(account='account', ssh_tunnel_port=1234),420                expected_calls=[421                    mock.call.local('docker tag registry:5000/test:latest fabricio-temp-image:test && docker rmi registry:5000/test:latest', ignore_errors=True, use_cache=True),422                    mock.call.local('docker pull registry:5000/test:latest', quiet=False, use_cache=True, ignore_errors=False),423                    mock.call.local('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=True),424                    mock.call.local('docker tag registry:5000/test:latest registry:5000/account/test:latest', use_cache=True),425                    mock.call.local('docker push registry:5000/account/test:latest', quiet=False, use_cache=True),426                    mock.call.local('docker rmi registry:5000/account/test:latest', use_cache=True),427                    mock.call.run('docker tag localhost:1234/account/test:latest fabricio-temp-image:test && docker rmi localhost:1234/account/test:latest', ignore_errors=True, use_cache=False),428                    mock.call.run('docker pull localhost:1234/account/test:latest', quiet=False, use_cache=False, ignore_errors=False),429                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),430                    mock.call.migrate(tag=None, registry='localhost:1234', account='account'),431                    mock.call.update(force=False, tag=None, registry='localhost:1234', account='account'),432                ],433                image_registry='registry:5000',434            ),435            custom_registry_and_image_registry=dict(436                deploy_kwargs=dict(),437                init_kwargs=dict(registry='host:4000'),438                expected_calls=[439                    mock.call.local('docker tag registry:5000/test:latest fabricio-temp-image:test && docker rmi registry:5000/test:latest', ignore_errors=True, use_cache=True),440                    mock.call.local('docker pull registry:5000/test:latest', quiet=False, use_cache=True, ignore_errors=False),441                    mock.call.local('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=True),442                    mock.call.local('docker tag registry:5000/test:latest host:4000/test:latest', use_cache=True),443                    mock.call.local('docker push host:4000/test:latest', quiet=False, use_cache=True),444                    mock.call.local('docker rmi host:4000/test:latest', use_cache=True),445                    mock.call.run('docker tag host:4000/test:latest fabricio-temp-image:test && docker rmi host:4000/test:latest', ignore_errors=True, use_cache=False),446                    mock.call.run('docker pull host:4000/test:latest', quiet=False, use_cache=False, ignore_errors=False),447                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),448                    mock.call.migrate(tag=None, registry='host:4000', account=None),449                    mock.call.update(force=False, tag=None, registry='host:4000', account=None),450                ],451                image_registry='registry:5000',452            ),453            custom_registry_and_image_registry_with_ssh_tunnel_deprecated=dict(454                deploy_kwargs=dict(),455                init_kwargs=dict(registry='host:4000', ssh_tunnel_port=1234),456                expected_calls=[457                    mock.call.local('docker tag registry:5000/test:latest fabricio-temp-image:test && docker rmi registry:5000/test:latest', ignore_errors=True, use_cache=True),458                    mock.call.local('docker pull registry:5000/test:latest', quiet=False, use_cache=True, ignore_errors=False),459                    mock.call.local('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=True),460                    mock.call.local('docker tag registry:5000/test:latest host:4000/test:latest', use_cache=True),461                    mock.call.local('docker push host:4000/test:latest', quiet=False, use_cache=True),462                    mock.call.local('docker rmi host:4000/test:latest', use_cache=True),463                    mock.call.run('docker tag localhost:1234/test:latest fabricio-temp-image:test && docker rmi localhost:1234/test:latest', ignore_errors=True, use_cache=False),464                    mock.call.run('docker pull localhost:1234/test:latest', quiet=False, use_cache=False, ignore_errors=False),465                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),466                    mock.call.migrate(tag=None, registry='localhost:1234', account=None),467                    mock.call.update(force=False, tag=None, registry='localhost:1234', account=None),468                ],469                image_registry='registry:5000',470            ),471            forced=dict(472                deploy_kwargs=dict(force='yes'),473                init_kwargs=dict(),474                expected_calls=[475                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),476                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),477                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),478                    mock.call.migrate(tag=None, registry=None, account=None),479                    mock.call.update(force=True, tag=None, registry=None, account=None),480                ],481                image_registry=None,482            ),483            explicit_not_forced=dict(484                deploy_kwargs=dict(force='no'),485                init_kwargs=dict(),486                expected_calls=[487                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),488                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),489                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),490                    mock.call.migrate(tag=None, registry=None, account=None),491                    mock.call.update(force=False, tag=None, registry=None, account=None),492                ],493                image_registry=None,494            ),495            custom_tag=dict(496                deploy_kwargs=dict(tag='tag'),497                init_kwargs=dict(),498                expected_calls=[499                    mock.call.run('docker tag test:tag fabricio-temp-image:test && docker rmi test:tag', ignore_errors=True, use_cache=False),500                    mock.call.run('docker pull test:tag', quiet=False, use_cache=False, ignore_errors=False),501                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),502                    mock.call.migrate(tag='tag', registry=None, account=None),503                    mock.call.update(force=False, tag='tag', registry=None, account=None),504                ],505                image_registry=None,506            ),507            backup_enabled=dict(508                deploy_kwargs=dict(backup='yes'),509                init_kwargs=dict(),510                expected_calls=[511                    mock.call.backup(),512                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),513                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),514                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),515                    mock.call.migrate(tag=None, registry=None, account=None),516                    mock.call.update(force=False, tag=None, registry=None, account=None),517                ],518                image_registry=None,519            ),520            explicit_backup_disabled=dict(521                deploy_kwargs=dict(backup='no'),522                init_kwargs=dict(),523                expected_calls=[524                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),525                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),526                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),527                    mock.call.migrate(tag=None, registry=None, account=None),528                    mock.call.update(force=False, tag=None, registry=None, account=None),529                ],530                image_registry=None,531            ),532            skip_migrations=dict(533                deploy_kwargs=dict(migrate='no'),534                init_kwargs=dict(),535                expected_calls=[536                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),537                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),538                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),539                    mock.call.update(force=False, tag=None, registry=None, account=None),540                ],541                image_registry=None,542            ),543            explicit_migrate=dict(544                deploy_kwargs=dict(migrate='yes'),545                init_kwargs=dict(),546                expected_calls=[547                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),548                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),549                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),550                    mock.call.migrate(tag=None, registry=None, account=None),551                    mock.call.update(force=False, tag=None, registry=None, account=None),552                ],553                image_registry=None,554            ),555            complex=dict(556                deploy_kwargs=dict(force=True, backup=True, tag='tag'),557                init_kwargs=dict(registry='host:4000', account='account'),558                expected_calls=[559                    mock.call.local('docker tag registry:5000/test:tag fabricio-temp-image:test && docker rmi registry:5000/test:tag', ignore_errors=True, use_cache=True),560                    mock.call.local('docker pull registry:5000/test:tag', quiet=False, use_cache=True, ignore_errors=False),561                    mock.call.local('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=True),562                    mock.call.local('docker tag registry:5000/test:tag host:4000/account/test:tag', use_cache=True),563                    mock.call.local('docker push host:4000/account/test:tag', quiet=False, use_cache=True),564                    mock.call.local('docker rmi host:4000/account/test:tag', use_cache=True),565                    mock.call.backup(),566                    mock.call.run('docker tag host:4000/account/test:tag fabricio-temp-image:test && docker rmi host:4000/account/test:tag', ignore_errors=True, use_cache=False),567                    mock.call.run('docker pull host:4000/account/test:tag', quiet=False, use_cache=False, ignore_errors=False),568                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),569                    mock.call.migrate(tag='tag', registry='host:4000', account='account'),570                    mock.call.update(force=True, tag='tag', registry='host:4000', account='account'),571                ],572                image_registry='registry:5000',573            ),574        )575        deploy = mock.Mock()576        deploy.attach_mock(backup, 'backup')577        deploy.attach_mock(migrate, 'migrate')578        deploy.attach_mock(update, 'update')579        deploy.attach_mock(run, 'run')580        deploy.attach_mock(local, 'local')581        update.return_value = False582        run.return_value = SucceededResult()583        local.return_value = SucceededResult()584        for case, data in cases.items():585            with self.subTest(case=case):586                deploy.reset_mock()587                tasks_list = tasks.DockerTasks(588                    service=data.get('service') or docker.Container(589                        name='name',590                        image=docker.Image('test', registry=data['image_registry']),591                    ),592                    hosts=['host'],593                    **data['init_kwargs']594                )595                tasks_list.deploy.name = '{0}__{1}'.format(self, case)596                fab.execute(tasks_list.deploy, **data['deploy_kwargs'])597                self.assertListEqual(data['expected_calls'], deploy.mock_calls)598    def test_delete_dangling_images_deprecated(self):599        cases = dict(600            windows=dict(601                os_name='nt',602                expected_command="for /F %i in ('docker images --filter \"dangling=true\" --quiet') do @docker rmi %i",603            ),604            posix=dict(605                os_name='posix',606                expected_command='for img in $(docker images --filter "dangling=true" --quiet); do docker rmi "$img"; done',607            ),608        )609        for case, data in cases.items():610            with self.subTest(case=case):611                with mock.patch.object(fabricio, 'local') as local:612                    with mock.patch('os.name', data['os_name']):613                        tasks_list = tasks.DockerTasks(service=docker.Container(name='name'))614                        tasks_list.delete_dangling_images()615                        local.assert_called_once_with(616                            data['expected_command'],617                            ignore_errors=True,618                        )619class ImageBuildDockerTasksTestCase(unittest.TestCase):620    maxDiff = None621    def setUp(self):622        self.fab_settings = fab.settings(fab.hide('everything'))623        self.fab_settings.__enter__()624    def tearDown(self):625        self.fab_settings.__exit__(None, None, None)626    @mock.patch.multiple(docker.Container, backup=mock.DEFAULT, migrate=mock.DEFAULT, update=mock.DEFAULT)627    @mock.patch.multiple(fabricio, run=mock.DEFAULT)628    @mock.patch.object(fab, 'remote_tunnel', return_value=mock.MagicMock())629    def test_deploy(self, remote_tunnel, run, backup, migrate, update):630        cases = dict(631            default=dict(632                deploy_kwargs=dict(),633                init_kwargs=dict(),634                expected_calls=[635                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),636                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),637                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),638                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),639                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),640                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),641                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),642                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),643                    mock.call.migrate(tag=None, registry=None, account=None),644                    mock.call.update(force=False, tag=None, registry=None, account=None),645                ],646                image_registry=None,647            ),648            custom_image_registry=dict(649                deploy_kwargs=dict(),650                init_kwargs=dict(),651                expected_calls=[652                    mock.call.local('docker inspect --type image registry:5000/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),653                    mock.call.local('docker tag registry:5000/test:latest fabricio-temp-image:test && docker rmi registry:5000/test:latest', ignore_errors=True, use_cache=True),654                    mock.call.local('docker build --tag=registry:5000/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),655                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),656                    mock.call.local('docker push registry:5000/test:latest', quiet=False, use_cache=True),657                    mock.call.run('docker tag registry:5000/test:latest fabricio-temp-image:test && docker rmi registry:5000/test:latest', ignore_errors=True, use_cache=False),658                    mock.call.run('docker pull registry:5000/test:latest', quiet=False, use_cache=False, ignore_errors=False),659                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),660                    mock.call.migrate(tag=None, registry=None, account=None),661                    mock.call.update(force=False, tag=None, registry=None, account=None),662                ],663                image_registry='registry:5000',664            ),665            custom_image_registry_with_ssh_tunnel_deprecated=dict(666                deploy_kwargs=dict(),667                init_kwargs=dict(ssh_tunnel_port=1234),668                expected_calls=[669                    mock.call.local('docker inspect --type image registry:5000/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),670                    mock.call.local('docker tag registry:5000/test:latest fabricio-temp-image:test && docker rmi registry:5000/test:latest', ignore_errors=True, use_cache=True),671                    mock.call.local('docker build --tag=registry:5000/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),672                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),673                    mock.call.local('docker push registry:5000/test:latest', quiet=False, use_cache=True),674                    mock.call.run('docker tag localhost:1234/test:latest fabricio-temp-image:test && docker rmi localhost:1234/test:latest', ignore_errors=True, use_cache=False),675                    mock.call.run('docker pull localhost:1234/test:latest', quiet=False, use_cache=False, ignore_errors=False),676                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),677                    mock.call.migrate(tag=None, registry='localhost:1234', account=None),678                    mock.call.update(force=False, tag=None, registry='localhost:1234', account=None),679                ],680                image_registry='registry:5000',681            ),682            custom_registry=dict(683                deploy_kwargs=dict(),684                init_kwargs=dict(registry='host:5000'),685                expected_calls=[686                    mock.call.local('docker inspect --type image host:5000/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),687                    mock.call.local('docker tag host:5000/test:latest fabricio-temp-image:test && docker rmi host:5000/test:latest', ignore_errors=True, use_cache=True),688                    mock.call.local('docker build --tag=host:5000/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),689                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),690                    mock.call.local('docker push host:5000/test:latest', quiet=False, use_cache=True),691                    mock.call.run('docker tag host:5000/test:latest fabricio-temp-image:test && docker rmi host:5000/test:latest', ignore_errors=True, use_cache=False),692                    mock.call.run('docker pull host:5000/test:latest', quiet=False, use_cache=False, ignore_errors=False),693                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),694                    mock.call.migrate(tag=None, registry='host:5000', account=None),695                    mock.call.update(force=False, tag=None, registry='host:5000', account=None),696                ],697                image_registry=None,698            ),699            custom_registry_with_ssh_tunnel_deprecated=dict(700                deploy_kwargs=dict(),701                init_kwargs=dict(registry='host:5000', ssh_tunnel_port=1234),702                expected_calls=[703                    mock.call.local('docker inspect --type image host:5000/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),704                    mock.call.local('docker tag host:5000/test:latest fabricio-temp-image:test && docker rmi host:5000/test:latest', ignore_errors=True, use_cache=True),705                    mock.call.local('docker build --tag=host:5000/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),706                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),707                    mock.call.local('docker push host:5000/test:latest', quiet=False, use_cache=True),708                    mock.call.run('docker tag localhost:1234/test:latest fabricio-temp-image:test && docker rmi localhost:1234/test:latest', ignore_errors=True, use_cache=False),709                    mock.call.run('docker pull localhost:1234/test:latest', quiet=False, use_cache=False, ignore_errors=False),710                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),711                    mock.call.migrate(tag=None, registry='localhost:1234', account=None),712                    mock.call.update(force=False, tag=None, registry='localhost:1234', account=None),713                ],714                image_registry=None,715            ),716            custom_account=dict(717                deploy_kwargs=dict(),718                init_kwargs=dict(account='account'),719                expected_calls=[720                    mock.call.local('docker inspect --type image account/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),721                    mock.call.local('docker tag account/test:latest fabricio-temp-image:test && docker rmi account/test:latest', ignore_errors=True, use_cache=True),722                    mock.call.local('docker build --tag=account/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),723                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),724                    mock.call.local('docker push account/test:latest', quiet=False, use_cache=True),725                    mock.call.run('docker tag account/test:latest fabricio-temp-image:test && docker rmi account/test:latest', ignore_errors=True, use_cache=False),726                    mock.call.run('docker pull account/test:latest', quiet=False, use_cache=False, ignore_errors=False),727                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),728                    mock.call.migrate(tag=None, registry=None, account='account'),729                    mock.call.update(force=False, tag=None, registry=None, account='account'),730                ],731                image_registry=None,732            ),733            custom_account_with_ssh_tunnel_deprecated=dict(734                deploy_kwargs=dict(),735                init_kwargs=dict(account='account', ssh_tunnel_port=1234),736                expected_calls=[737                    mock.call.local('docker inspect --type image host:5000/account/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),738                    mock.call.local('docker tag host:5000/account/test:latest fabricio-temp-image:test && docker rmi host:5000/account/test:latest', ignore_errors=True, use_cache=True),739                    mock.call.local('docker build --tag=host:5000/account/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),740                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),741                    mock.call.local('docker push host:5000/account/test:latest', quiet=False, use_cache=True),742                    mock.call.run('docker tag localhost:1234/account/test:latest fabricio-temp-image:test && docker rmi localhost:1234/account/test:latest', ignore_errors=True, use_cache=False),743                    mock.call.run('docker pull localhost:1234/account/test:latest', quiet=False, use_cache=False, ignore_errors=False),744                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),745                    mock.call.migrate(tag=None, registry='localhost:1234', account='account'),746                    mock.call.update(force=False, tag=None, registry='localhost:1234', account='account'),747                ],748                image_registry='host:5000',749            ),750            custom_registry_and_image_registry=dict(751                deploy_kwargs=dict(),752                init_kwargs=dict(registry='host:4000'),753                expected_calls=[754                    mock.call.local('docker inspect --type image host:4000/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),755                    mock.call.local('docker tag host:4000/test:latest fabricio-temp-image:test && docker rmi host:4000/test:latest', ignore_errors=True, use_cache=True),756                    mock.call.local('docker build --tag=host:4000/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),757                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),758                    mock.call.local('docker push host:4000/test:latest', quiet=False, use_cache=True),759                    mock.call.run('docker tag host:4000/test:latest fabricio-temp-image:test && docker rmi host:4000/test:latest', ignore_errors=True, use_cache=False),760                    mock.call.run('docker pull host:4000/test:latest', quiet=False, use_cache=False, ignore_errors=False),761                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),762                    mock.call.migrate(tag=None, registry='host:4000', account=None),763                    mock.call.update(force=False, tag=None, registry='host:4000', account=None),764                ],765                image_registry='registry:5000',766            ),767            custom_registry_and_image_registry_with_ssh_tunnel_deprecated=dict(768                deploy_kwargs=dict(),769                init_kwargs=dict(registry='host:4000', ssh_tunnel_port=1234),770                expected_calls=[771                    mock.call.local('docker inspect --type image host:4000/test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),772                    mock.call.local('docker tag host:4000/test:latest fabricio-temp-image:test && docker rmi host:4000/test:latest', ignore_errors=True, use_cache=True),773                    mock.call.local('docker build --tag=host:4000/test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),774                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),775                    mock.call.local('docker push host:4000/test:latest', quiet=False, use_cache=True),776                    mock.call.run('docker tag localhost:1234/test:latest fabricio-temp-image:test && docker rmi localhost:1234/test:latest', ignore_errors=True, use_cache=False),777                    mock.call.run('docker pull localhost:1234/test:latest', quiet=False, use_cache=False, ignore_errors=False),778                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),779                    mock.call.migrate(tag=None, registry='localhost:1234', account=None),780                    mock.call.update(force=False, tag=None, registry='localhost:1234', account=None),781                ],782                image_registry='registry:5000',783            ),784            forced=dict(785                deploy_kwargs=dict(force='yes'),786                init_kwargs=dict(),787                expected_calls=[788                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),789                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),790                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),791                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),792                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),793                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),794                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),795                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),796                    mock.call.migrate(tag=None, registry=None, account=None),797                    mock.call.update(force=True, tag=None, registry=None, account=None),798                ],799                image_registry=None,800            ),801            explicit_not_forced=dict(802                deploy_kwargs=dict(force='no'),803                init_kwargs=dict(),804                expected_calls=[805                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),806                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),807                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),808                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),809                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),810                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),811                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),812                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),813                    mock.call.migrate(tag=None, registry=None, account=None),814                    mock.call.update(force=False, tag=None, registry=None, account=None),815                ],816                image_registry=None,817            ),818            custom_build_path=dict(819                deploy_kwargs=dict(),820                init_kwargs=dict(build_path='foo'),821                expected_calls=[822                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),823                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),824                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 foo', quiet=False, use_cache=True),825                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),826                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),827                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),828                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),829                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),830                    mock.call.migrate(tag=None, registry=None, account=None),831                    mock.call.update(force=False, tag=None, registry=None, account=None),832                ],833                image_registry=None,834            ),835            custom_tag=dict(836                deploy_kwargs=dict(tag='tag'),837                init_kwargs=dict(),838                expected_calls=[839                    mock.call.local('docker inspect --type image test:tag', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),840                    mock.call.local('docker tag test:tag fabricio-temp-image:test && docker rmi test:tag', ignore_errors=True, use_cache=True),841                    mock.call.local('docker build --tag=test:tag --pull=1 --force-rm=1 .', quiet=False, use_cache=True),842                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),843                    mock.call.local('docker push test:tag', quiet=False, use_cache=True),844                    mock.call.run('docker tag test:tag fabricio-temp-image:test && docker rmi test:tag', ignore_errors=True, use_cache=False),845                    mock.call.run('docker pull test:tag', quiet=False, use_cache=False, ignore_errors=False),846                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),847                    mock.call.migrate(tag='tag', registry=None, account=None),848                    mock.call.update(force=False, tag='tag', registry=None, account=None),849                ],850                image_registry=None,851            ),852            backup_enabled=dict(853                deploy_kwargs=dict(backup='yes'),854                init_kwargs=dict(),855                expected_calls=[856                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),857                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),858                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),859                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),860                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),861                    mock.call.backup(),862                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),863                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),864                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),865                    mock.call.migrate(tag=None, registry=None, account=None),866                    mock.call.update(force=False, tag=None, registry=None, account=None),867                ],868                image_registry=None,869            ),870            explicit_backup_disabled=dict(871                deploy_kwargs=dict(backup='no'),872                init_kwargs=dict(),873                expected_calls=[874                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),875                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),876                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),877                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),878                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),879                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),880                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),881                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),882                    mock.call.migrate(tag=None, registry=None, account=None),883                    mock.call.update(force=False, tag=None, registry=None, account=None),884                ],885                image_registry=None,886            ),887            skip_migrations=dict(888                deploy_kwargs=dict(migrate='no'),889                init_kwargs=dict(),890                expected_calls=[891                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),892                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),893                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),894                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),895                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),896                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),897                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),898                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),899                    mock.call.update(force=False, tag=None, registry=None, account=None),900                ],901                image_registry=None,902            ),903            explicit_migrate=dict(904                deploy_kwargs=dict(migrate='yes'),905                init_kwargs=dict(),906                expected_calls=[907                    mock.call.local('docker inspect --type image test:latest', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),908                    mock.call.local('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=True),909                    mock.call.local('docker build --tag=test:latest --pull=1 --force-rm=1 .', quiet=False, use_cache=True),910                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),911                    mock.call.local('docker push test:latest', quiet=False, use_cache=True),912                    mock.call.run('docker tag test:latest fabricio-temp-image:test && docker rmi test:latest', ignore_errors=True, use_cache=False),913                    mock.call.run('docker pull test:latest', quiet=False, use_cache=False, ignore_errors=False),914                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),915                    mock.call.migrate(tag=None, registry=None, account=None),916                    mock.call.update(force=False, tag=None, registry=None, account=None),917                ],918                image_registry=None,919            ),920            complex=dict(921                deploy_kwargs=dict(force=True, backup=True, tag='tag'),922                init_kwargs=dict(registry='host:4000', build_path='foo', account='account'),923                expected_calls=[924                    mock.call.local('docker inspect --type image host:4000/account/test:tag', capture=True, use_cache=True, abort_exception=docker.ImageNotFoundError),925                    mock.call.local('docker tag host:4000/account/test:tag fabricio-temp-image:test && docker rmi host:4000/account/test:tag', ignore_errors=True, use_cache=True),926                    mock.call.local('docker build --tag=host:4000/account/test:tag --pull=1 --force-rm=1 foo', quiet=False, use_cache=True),927                    mock.call.local('docker rmi fabricio-temp-image:test old_parent_id', ignore_errors=True, use_cache=True),928                    mock.call.local('docker push host:4000/account/test:tag', quiet=False, use_cache=True),929                    mock.call.backup(),930                    mock.call.run('docker tag host:4000/account/test:tag fabricio-temp-image:test && docker rmi host:4000/account/test:tag', ignore_errors=True, use_cache=False),931                    mock.call.run('docker pull host:4000/account/test:tag', quiet=False, use_cache=False, ignore_errors=False),932                    mock.call.run('docker rmi fabricio-temp-image:test', ignore_errors=True, use_cache=False),933                    mock.call.migrate(tag='tag', registry='host:4000', account='account'),934                    mock.call.update(force=True, tag='tag', registry='host:4000', account='account'),935                ],936                image_registry='registry:5000',937            ),938        )939        with mock.patch.object(fabricio, 'local') as local:940            deploy = mock.Mock()941            deploy.attach_mock(backup, 'backup')942            deploy.attach_mock(migrate, 'migrate')943            deploy.attach_mock(update, 'update')944            deploy.attach_mock(run, 'run')945            deploy.attach_mock(local, 'local')946            update.return_value = False947            run.return_value = SucceededResult()948            local.return_value = SucceededResult('[{"Parent": "old_parent_id"}]')949            for case, data in cases.items():950                with self.subTest(case=case):951                    deploy.reset_mock()952                    tasks_list = tasks.ImageBuildDockerTasks(953                        service=docker.Container(954                            name='name',955                            image=docker.Image('test', registry=data['image_registry']),956                        ),957                        hosts=['host'],958                        **data['init_kwargs']959                    )960                    tasks_list.deploy.name = '{0}__{1}'.format(self, case)961                    fab.execute(tasks_list.deploy, **data['deploy_kwargs'])962                    self.assertListEqual(data['expected_calls'], deploy.mock_calls)963    def test_prepare(self):964        cases = dict(965            default=dict(966                kwargs=dict(),967                expected_docker_build_command={968                    'executable': ['docker', 'build'],969                    'tag': 'image:latest',970                    'pull': True,971                    'force-rm': True,972                    'path': '.',973                },974            ),975            no_force_rm=dict(976                kwargs={'force-rm': 'no'},977                expected_docker_build_command={978                    'executable': ['docker', 'build'],979                    'tag': 'image:latest',980                    'pull': True,981                    'path': '.',982                    'force-rm': 0,983                },984            ),985            no_pull=dict(986                kwargs={'pull': 'no'},987                expected_docker_build_command={988                    'executable': ['docker', 'build'],989                    'tag': 'image:latest',990                    'force-rm': True,991                    'path': '.',992                    'pull': 0,993                },994            ),995            complex=dict(996                kwargs={'pull': 'no', 'force-rm': 'no', 'custom': 'custom', 'custom-bool': 'yes'},997                expected_docker_build_command={998                    'executable': ['docker', 'build'],999                    'tag': 'image:latest',1000                    'custom-bool': 1,1001                    'custom': 'custom',1002                    'path': '.',1003                    'pull': 0,1004                    'force-rm': 0,1005                },1006            ),1007        )1008        expected_calls = [1009            mock.call('docker inspect --type image image:latest', use_cache=True, capture=True, abort_exception=docker.ImageNotFoundError),1010            mock.call('docker tag image:latest fabricio-temp-image:image && docker rmi image:latest', use_cache=True, ignore_errors=True),1011            mock.call(mock.ANY, quiet=False, use_cache=True),  # docker build1012            mock.call('docker rmi fabricio-temp-image:image old_parent_id', use_cache=True, ignore_errors=True),1013        ]1014        def test_docker_build_command(command, **kwargs):1015            command = command.split(';', 1)[-1].strip()1016            if not command.startswith('docker build'):1017                return SucceededResult('[{"Parent": "old_parent_id"}]')1018            args = shlex.split(command)1019            actual_docker_build_command = vars(docker_build_args_parser.parse_args(args))1020            self.assertDictEqual(1021                expected_docker_build_command,1022                actual_docker_build_command,1023            )1024        for case, data in cases.items():1025            with self.subTest(case=case):1026                with mock.patch.object(fabricio, 'local', side_effect=test_docker_build_command) as local:1027                    expected_docker_build_command = data['expected_docker_build_command']1028                    tasks_list = tasks.ImageBuildDockerTasks(1029                        service=docker.Container(name='name', image='image'),1030                        hosts=['host'],1031                    )1032                    fab.execute(tasks_list.prepare, **data['kwargs'])1033                    self.assertListEqual(local.mock_calls, expected_calls)1034class SshTunnelTestCase(unittest.TestCase):1035    def test_init(self):1036        cases = dict(1037            full=dict(1038                mapping='bind_address:1111:host:2222',1039                expected_bind_address='bind_address',1040                expected_port=1111,1041                expected_host='host',1042                expected_host_port=2222,1043            ),1044            single_port_int=dict(1045                mapping=1111,1046                expected_bind_address='127.0.0.1',1047                expected_port=1111,1048                expected_host='localhost',1049                expected_host_port=1111,1050            ),1051            single_port_str=dict(1052                mapping='1111',1053                expected_bind_address='127.0.0.1',1054                expected_port=1111,1055                expected_host='localhost',1056                expected_host_port=1111,1057            ),1058            double_ports=dict(1059                mapping='1111:2222',1060                expected_bind_address='127.0.0.1',1061                expected_port=1111,1062                expected_host='localhost',1063                expected_host_port=2222,1064            ),1065            double_ports_with_host=dict(1066                mapping='1111:host:2222',1067                expected_bind_address='127.0.0.1',1068                expected_port=1111,1069                expected_host='host',1070                expected_host_port=2222,1071            ),1072        )1073        for case, data in cases.items():1074            with self.subTest(case):1075                tunnel = tasks.SshTunnel(data['mapping'])1076                self.assertEqual(tunnel.bind_address, data['expected_bind_address'])1077                self.assertEqual(tunnel.port, data['expected_port'])1078                self.assertEqual(tunnel.host, data['expected_host'])1079                self.assertEqual(tunnel.host_port, data['expected_host_port'])1080    def test_init_none_mapping(self):...vk.py
Source:vk.py  
1import typing2import typing as ty3from dev_up import models4from dev_up.base.category import APICategory5class VKAPICategory(APICategory):6    async def experts_get_info(7            self,8            user_id: int,9            *,10            access_token: ty.Optional[str] = None,11            raise_error: bool = None,12            raw_response: bool = None,13            use_cache: bool = None14    ) -> models.VKExpertsGetInfoResponse:15        """ÐолÑÑÐ°ÐµÑ Ð¸Ð½ÑоÑмаÑио о полÑзоваÑеле в пÑогÑамме VK Experts16        :param user_id: VK ID полÑзоваÑелÑ17        :param access_token: Ñокен18        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`19        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ20        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ21        """22        return await self.api.make_request(23            method="vk.expertsGetInfo",24            data=dict(user_id=user_id),25            access_token=access_token,26            request_model=models.VKExpertsGetInfoRequest,27            response_model=models.VKExpertsGetInfoResponse,28            raise_error=raise_error,29            raw_response=raw_response,30            use_cache=use_cache31        )32    async def get_apps(33            self,34            user_id: int,35            *,36            access_token: ty.Optional[str] = None,37            raise_error: bool = None,38            raw_response: bool = None,39            use_cache: bool = None40    ) -> models.VKGetAppsResponse:41        """ÐолÑÑÐ°ÐµÑ ÑпиÑок пÑиложений полÑзоваÑелÑ42        :param user_id: VK ID полÑзоваÑелÑ43        :param access_token: Ñокен44        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`45        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ46        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ47        """48        return await self.api.make_request(49            method="vk.getApps",50            data=dict(user_id=user_id),51            access_token=access_token,52            request_model=models.VKExpertsGetInfoRequest,53            response_model=models.VKGetAppsResponse,54            raise_error=raise_error,55            raw_response=raw_response,56            use_cache=use_cache57        )58    async def get_groups(59            self,60            user_id: int,61            *,62            access_token: ty.Optional[str] = None,63            raise_error: bool = None,64            raw_response: bool = None,65            use_cache: bool = None66    ) -> models.VKGetGroupsResponse:67        """ÐолÑÑÐ°ÐµÑ ÑпиÑок гÑÑпп полÑзоваÑелÑ68        :param user_id: VK ID полÑзоваÑелÑ69        :param access_token: Ñокен70        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`71        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ72        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ73        """74        return await self.api.make_request(75            method="vk.getGroups",76            data=dict(user_id=user_id),77            access_token=access_token,78            request_model=models.VKGetGroupsRequest,79            response_model=models.VKGetGroupsResponse,80            raise_error=raise_error,81            raw_response=raw_response,82            use_cache=use_cache83        )84    async def get_sticker_info(85            self,86            sticker_id: int,87            *,88            access_token: ty.Optional[str] = None,89            raise_error: bool = None,90            raw_response: bool = None,91            use_cache: bool = None92    ) -> models.VKGetStickerInfoResponse:93        """ÐолÑÑÐ°ÐµÑ Ð¸Ð½ÑоÑмаÑÐ¸Ñ Ð¾ ÑÑикеÑе и ÑÑикеÑ-паке94        :param sticker_id: ID ÑÑикеÑа95        :param access_token: Ñокен96        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`97        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ98        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ99        """100        return await self.api.make_request(101            method="vk.getStickerInfo",102            data=dict(sticker_id=sticker_id),103            access_token=access_token,104            request_model=models.VKGetStickerInfoRequest,105            response_model=models.VKGetStickerInfoResponse,106            raise_error=raise_error,107            raw_response=raw_response,108            use_cache=use_cache109        )110    async def get_stickers(111            self,112            user_id: int,113            *,114            access_token: ty.Optional[str] = None,115            raise_error: bool = None,116            raw_response: bool = None,117            use_cache: bool = None118    ) -> models.VKGetStickersResponse:119        """ÐолÑÑÐ°ÐµÑ ÑпиÑок ÑÑикеÑов полÑзоваÑелÑ120        :param user_id: VK ID полÑзоваÑелÑ121        :param access_token: Ñокен122        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`123        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ124        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ125        """126        return await self.api.make_request(127            method="vk.getStickers",128            data=dict(user_id=user_id),129            access_token=access_token,130            request_model=models.VKGetStickersRequest,131            response_model=models.VKGetStickersResponse,132            raise_error=raise_error,133            raw_response=raw_response,134            use_cache=use_cache135        )136    async def group_get_managers(137            self,138            group_id: typing.Union[str, int],139            *,140            access_token: ty.Optional[str] = None,141            raise_error: bool = None,142            raw_response: bool = None,143            use_cache: bool = None144    ) -> models.VKGroupGetManagersResponse:145        """ÐолÑÑÐ°ÐµÑ Ð¸Ð½ÑоÑмаÑÐ¸Ñ Ð¾Ð± админиÑÑÑаÑоÑаÑ
 гÑÑппÑ146        :param group_id: VK ID гÑÑппÑ147        :param access_token: Ñокен148        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`149        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ150        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ151        """152        return await self.api.make_request(153            method="vk.groupGetManagers",154            data=dict(group_id=group_id),155            access_token=access_token,156            request_model=models.VKGroupGetManagersRequest,157            response_model=models.VKGroupGetManagersResponse,158            raise_error=raise_error,159            raw_response=raw_response,160            use_cache=use_cache161        )162    async def search_audio(163            self,164            q: str,165            *,166            access_token: ty.Optional[str] = None,167            raise_error: bool = None,168            raw_response: bool = None,169            use_cache: bool = None170    ) -> models.VKSearchAudioResponse:171        """ÐоиÑк аÑдиозапиÑей172        :param q: поиÑÐºÐ¾Ð²Ð°Ñ ÑÑÑока173        :param access_token: Ñокен174        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`175        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ176        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ177        """178        return await self.api.make_request(179            method="vk.searchAudio",180            data=dict(q=q),181            access_token=access_token,182            request_model=models.VKSearchAudioRequest,183            response_model=models.VKSearchAudioResponse,184            raise_error=raise_error,185            raw_response=raw_response,186            use_cache=use_cache187        )188    async def search_playlists(189            self,190            q: str,191            *,192            access_token: ty.Optional[str] = None,193            raise_error: bool = None,194            raw_response: bool = None,195            use_cache: bool = None196    ) -> models.VKSearchPlaylistsResponse:197        """ÐоиÑк плейлиÑÑов198        :param q: поиÑÐºÐ¾Ð²Ð°Ñ ÑÑÑока199        :param access_token: Ñокен200        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`201        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ202        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ203        """204        return await self.api.make_request(205            method="vk.searchPlaylists",206            data=dict(q=q),207            access_token=access_token,208            request_model=models.VKSearchPlaylistsRequest,209            response_model=models.VKSearchPlaylistsResponse,210            raise_error=raise_error,211            raw_response=raw_response,212            use_cache=use_cache213        )214    async def set_steps(215            self,216            vk_me_access_token: str,217            *,218            access_token: ty.Optional[str] = None,219            raise_error: bool = None,220            raw_response: bool = None,221            use_cache: bool = None222    ) -> models.VKSetStepsResponse:223        """УÑÑÐ°Ð½Ð°Ð²Ð»Ð¸Ð²Ð°ÐµÑ ÐºÐ¾Ð»Ð¸ÑеÑÑво Ñагов224        :param vk_me_access_token: Ñокен пÑÐ¸Ð»Ð¾Ð¶ÐµÐ½Ð¸Ñ VK ME225        :param access_token: Ñокен226        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`227        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ228        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ229        """230        return await self.api.make_request(231            method="vk.setSteps",232            data=dict(access_token=vk_me_access_token),233            access_token=access_token,234            request_model=models.VKSetStepsRequest,235            response_model=models.VKSetStepsResponse,236            raise_error=raise_error,237            raw_response=raw_response,238            use_cache=use_cache239        )240    async def testers_get_info(241            self,242            user_id: int,243            *,244            access_token: ty.Optional[str] = None,245            raise_error: bool = None,246            raw_response: bool = None,247            use_cache: bool = None248    ) -> models.VKSearchPlaylistsResponse:249        """ÐолÑÑÐ°ÐµÑ Ð¸Ð½ÑоÑмаÑÐ¸Ñ Ð¾ полÑзоваÑеле в пÑогÑамме VK Testers250        :param user_id: VK ID полÑзоваÑелÑ251        :param access_token: Ñокен252        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`253        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ254        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ255        """256        return await self.api.make_request(257            method="vk.testersGetInfo",258            data=dict(user_id=user_id),259            access_token=access_token,260            request_model=models.VKSearchPlaylistsRequest,261            response_model=models.VKSearchPlaylistsResponse,262            raise_error=raise_error,263            raw_response=raw_response,264            use_cache=use_cache265        )266    async def user_get_subscriptions(267            self,268            user_id: int,269            *,270            access_token: ty.Optional[str] = None,271            raise_error: bool = None,272            raw_response: bool = None,273            use_cache: bool = None274    ) -> models.VKUserGetSubscriptionsResponse:275        """ÐолÑÑÐ°ÐµÑ Ð¸Ð½ÑоÑмаÑÐ¸Ñ Ð¾Ð± подпиÑкаÑ
 полÑзоваÑелÑ276        :param user_id: VK ID полÑзоваÑелÑ277        :param access_token: Ñокен278        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`279        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ280        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ281        """282        return await self.api.make_request(283            method="vk.userGetSubscriptions",284            data=dict(user_id=user_id),285            access_token=access_token,286            request_model=models.VKUserGetSubscriptionsRequest,287            response_model=models.VKUserGetSubscriptionsResponse,288            raise_error=raise_error,289            raw_response=raw_response,290            use_cache=use_cache...utils.py
Source:utils.py  
1import typing as ty2from dev_up import models3from dev_up.base.category import APICategory4class UtilsAPICategory(APICategory):5    async def check_link(6            self,7            url: str,8            *,9            access_token: ty.Optional[str] = None,10            raise_error: bool = None,11            raw_response: bool = None,12            use_cache: bool = None13    ) -> models.UtilsCheckLinkResponse:14        """ÐолÑÑÐ°ÐµÑ Ð°Ð´ÑеÑ, на коÑоÑÑй Ð²ÐµÐ´ÐµÑ ÑокÑаÑÐµÐ½Ð½Ð°Ñ ÑÑÑлка15        :param url: пÑовеÑÑÐµÐ¼Ð°Ñ ÑÑÑлка16        :param access_token: Ñокен17        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`18        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ19        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ20        """21        return await self.api.make_request(22            method="utils.checkLink",23            data=dict(url=url),24            access_token=access_token,25            request_model=models.UtilsCheckLinkRequest,26            response_model=models.UtilsCheckLinkResponse,27            raise_error=raise_error,28            raw_response=raw_response,29            use_cache=use_cache30        )31    async def create_short_link(32            self,33            url: str,34            *,35            access_token: ty.Optional[str] = None,36            raise_error: bool = None,37            raw_response: bool = None,38            use_cache: bool = None39    ) -> models.UtilsCreateShortLinkResponse:40        """СокÑаÑение ÑÑÑлок41        :param url: ÑÑÑлка42        :param access_token: Ñокен43        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`44        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ45        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ46        """47        return await self.api.make_request(48            method="utils.createShortLink",49            data=dict(url=url),50            access_token=access_token,51            request_model=models.UtilsCreateShortLinkRequest,52            response_model=models.UtilsCreateShortLinkResponse,53            raise_error=raise_error,54            raw_response=raw_response,55            use_cache=use_cache56        )57    async def get_server_time(58            self,59            *,60            access_token: ty.Optional[str] = None,61            raise_error: bool = None,62            raw_response: bool = None,63            use_cache: bool = None64    ) -> models.UtilsGetServerTimeResponse:65        """ÐозвÑаÑÐ°ÐµÑ ÑекÑÑее вÑÐµÐ¼Ñ Ð½Ð° ÑеÑвеÑе66        :param access_token: Ñокен67        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`68        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ69        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ70        """71        return await self.api.make_request(72            method="utils.getServerTime",73            data=dict(),74            access_token=access_token,75            request_model=models.UtilsGetServerTimeRequest,76            response_model=models.UtilsGetServerTimeResponse,77            raise_error=raise_error,78            raw_response=raw_response,79            use_cache=use_cache80        )81    async def get_web_info(82            self,83            address: str,84            *,85            access_token: ty.Optional[str] = None,86            raise_error: bool = None,87            raw_response: bool = None,88            use_cache: bool = None89    ) -> models.UtilsGetWebInfoResponse:90        """ÐнÑоÑмаÑÐ¸Ñ Ð¾ ÑеÑвеÑе91        :param address: URL или IP ÑеÑвеÑа92        :param access_token: Ñокен93        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`94        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ95        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ96        """97        return await self.api.make_request(98            method="utils.getWebInfo",99            data=dict(address=address),100            access_token=access_token,101            request_model=models.UtilsGetWebInfoRequest,102            response_model=models.UtilsGetWebInfoResponse,103            raise_error=raise_error,104            raw_response=raw_response,105            use_cache=use_cache106        )107    async def md5_generate(108            self,109            text: str,110            *,111            access_token: ty.Optional[str] = None,112            raise_error: bool = None,113            raw_response: bool = None,114            use_cache: bool = None115    ) -> models.UtilsMD5GenerateResponse:116        """ÐолÑÑиÑÑ Ñ
ÑÑ md5 Ð¾Ñ ÑекÑÑа117        :param text: знаÑение Ð¾Ñ ÐºÐ¾ÑоÑого необÑ
одимо полÑÑиÑÑ Ñ
ÑÑ118        :param access_token: Ñокен119        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`120        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ121        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ122        """123        return await self.api.make_request(124            method="utils.md5Generate",125            data=dict(text=text),126            access_token=access_token,127            request_model=models.UtilsMD5GenerateRequest,128            response_model=models.UtilsMD5GenerateResponse,129            raise_error=raise_error,130            raw_response=raw_response,131            use_cache=use_cache132        )133    async def notifications_links(134            self,135            code: str,136            status: models.UtilsNotificationsLinksStatus,137            *,138            access_token: ty.Optional[str] = None,139            raise_error: bool = None,140            raw_response: bool = None,141            use_cache: bool = None142    ) -> models.UtilsNotificationsLinksResponse:143        """ÐолÑÑиÑÑ Ñ
ÑÑ md5 Ð¾Ñ ÑекÑÑа144        :param code: код ÑÑÑлки145        :param status: ÑÑаÑÑÑ Ð¾Ð¿Ð¾Ð²ÐµÑений146        :param access_token: Ñокен147        :param raise_error: вÑзÑваÑÑ Ð¸ÑклÑÑение `DevUpResponseException`148        :param raw_response: возвÑаÑаÑÑ Ð½ÐµÐ¾Ð±ÑабоÑаннÑй оÑвеÑ149        :param use_cache: иÑполÑзоваÑÑ ÐºÑÑ150        """151        return await self.api.make_request(152            method="utils.notificationsLinks",153            data=dict(code=code, status=status),154            access_token=access_token,155            request_model=models.UtilsNotificationsLinksRequest,156            response_model=models.UtilsNotificationsLinksResponse,157            raise_error=raise_error,158            raw_response=raw_response,159            use_cache=use_cache...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
