How to use amplify - 10 common examples

def setup_method(self, method):
        super(MySQLMetricsCollectorTestCase, self).setup_method(method)
        context._setup_object_tank()

        self.mysql_obj = MySQLObject(
            local_id=123,
            pid=2,
            cmd='/usr/sbin/mysqld --basedir=/usr',
            conf_path='/etc/mysql/my.cnf',
        )

        context.objects.register(self.mysql_obj)

def teardown_method(self, method):
        context._setup_object_tank()
        super(PHPFPMPoolMetricsCollectorTestCase, self).teardown_method(method)

def test_init_ipv4(self):
        del context.app_config['mysql']['unix_socket']
        context.app_config['mysql']['host'] = '10.10.10.10'
        context.app_config['mysql']['port'] = '3306'

        assert_that(isinstance(context.app_config['mysql']['port'], str), equal_to(True))

        mysql_obj = MySQLObject(
            local_id=123,
            pid=2,
            cmd='/usr/sbin/mysqld --basedir=/usr '
                '--datadir=/var/lib/mysql --plugin-dir=/usr/lib/mysql/plugin '
                '--user=mysql --log-error=/var/log/mysql/error.log '
                '--pid-file=/var/run/mysqld/mysqld.pid --socket=/var/run/mysqld/mysqld.sock --port=3306',
            conf_path='/etc/mysql/my.cnf'
        )
        assert_that(mysql_obj, not_none())

        assert_that(mysql_obj.connection_args, equal_to(

def nginx_installed():
    try:
        subp.call('/usr/sbin/nginx -V')
        return True
    except:
        return False

def teardown_method(self, method):
        # use check=False because sometimes this returns code 123 on test-plus for some reason
        subp.call('pgrep nginx | sudo xargs kill -9', check=False)
        subp.call('rm -f %s' % self.conf_path, check=False)
        super(RealNginxTempFileTestCase, self).teardown_method(method)

def test_default_interface(self):
        container = SystemManager()
        container._discover_objects()

        os_obj = container.objects.objects[container.objects.objects_by_type[container.type][0]]
        collector = SystemCommonMetaCollector(object=os_obj)
        collector.collect()

        default_from_netstat, _ = subp.call(
            'netstat -nr | egrep -i "^0.0.0.0|default" | head -1 | sed "s/.*[ ]\([^ ][^ ]*\)$/\\1/"'
        )[0]

        default_interface = os_obj.metad.current['network']['default']

        assert_that(default_interface, equal_to(default_from_netstat))

def reload_nginx(self):
        subp.call('service nginx reload', check=False)
        time.sleep(0.5)

def teardown_method(self, method):
        from amplify.ext.configurator.config import ConfiguratorConfig
        ConfiguratorConfig.config = copy.deepcopy(self.original_config)
        context.app_config.remove(ConfiguratorConfig())
        super(BaseConfiguratorTestCase, self).teardown_method(method)

def test_freeze_api_url(self):
        # check that if api_url is not set it will not prevent agent from setting api_url from cloud
        context.app_config['cloud']['api_url'] = ''
        context.setup(app='test', app_config=context.app_config.default)
        assert_that(context.freeze_api_url, equal_to(False))

        # check that an api_url from our receiver's domain will not prevent agent from setting api_url from cloud
        context.app_config['cloud']['api_url'] = 'https://receiver.amplify.nginx.com:443/1.1'
        context.setup(app='test', app_config=context.app_config.default)
        assert_that(context.freeze_api_url, equal_to(False))

        # check that a custom api_url will prevent agent from setting api_url from cloud
        context.app_config['cloud']['api_url'] = 'http://some.other.domain/endpoint/'
        context.setup(app='test', app_config=context.app_config.default)
        assert_that(context.freeze_api_url, equal_to(True))

def teardown_method(self, method):
        context.app_config['credentials']['imagename'] = None
        context.app_config['credentials']['uuid'] = DEFAULT_UUID
        context.setup(app='test', app_config=context.app_config.default)

        context.objects = None
        context._setup_object_tank()
        super(ContainerSystemManagerTestCase, self).teardown_method(method)