Produits Entr'ouvert » OLAP / Business Intelligence pour Publik

      install_requires=['requests', 'psycopg2', 'isodate', 'six', 'cached-property'],

@pytest.fixture

def olap_cmd(wcs, tmpdir, postgres_db):

    config_ini = tmpdir / 'config.ini'

    model_dir = tmpdir / 'model_dir'

    model_dir.mkdir()

    with config_ini.open('w') as fd:

        fd.write(u'''

[wcs-olap]

cubes_model_dirs = {model_dir}

pg_dsn = {dsn}

[{wcs.url}]

orig = olap

key = olap

schema = olap

'''.format(wcs=wcs, model_dir=model_dir, dsn=postgres_db.dsn))

    from wcs_olap import cmd

    import sys

    def f(no_log_errors=True):

        old_argv = sys.argv

        try:

            sys.argv = ['', str(config_ini)]

            if no_log_errors:

                sys.argv.insert(1, '--no-log-errors')

            cmd.main2()

        finally:

            sys.argv = old_argv

    f.model_dir = model_dir

    return f

import pytest

import requests

import pathlib2

import mock

def test_wcs_fixture(wcs, postgres_db, tmpdir, olap_cmd, caplog):

    olap_cmd()

    with (olap_cmd.model_dir / 'olap.model').open() as fd, \

            (pathlib2.Path(__file__).parent / 'olap.model').open() as fd2:

def test_requests_exception(wcs, postgres_db, tmpdir, olap_cmd, caplog):

    with mock.patch('requests.get', side_effect=requests.RequestException('wat!')):

        with pytest.raises(SystemExit):

            olap_cmd(no_log_errors=False)

    assert 'wat!' in caplog.text

def test_requests_not_ok(wcs, postgres_db, tmpdir, olap_cmd, caplog):

    with mock.patch('requests.get') as mocked_get:

        mocked_get.return_value.ok = False

        mocked_get.return_value.status_code = 401

        mocked_get.return_value.text = '{"err": 1, "err_desc": "invalid signature"}'

        with pytest.raises(SystemExit):

            olap_cmd(no_log_errors=False)

    assert 'invalid signature' in caplog.text

def test_requests_not_json(wcs, postgres_db, tmpdir, olap_cmd, caplog):

    with mock.patch('requests.get') as mocked_get:

        mocked_get.return_value.ok = True

        mocked_get.return_value.json.side_effect = ValueError('invalid JSON')

        with pytest.raises(SystemExit):

            olap_cmd(no_log_errors=False)

    assert 'Invalid JSON content' in caplog.text

	mock

from cached_property import cached_property

        # keep at end of __init__ to prevent leak if __init__ raises

        self.connection = psycopg2.connect(dsn=pg_dsn)

        self.connection.autocommit = True

        self.cur = self.connection.cursor()

        try:

            self.has_jsonb = self.detect_jsonb()

            if self.has_jsonb:

                cube['json_field'] = 'json_data'

        except Exception:

            self.connection.close()

            raise

    @cached_property

    def formdefs(self):

        return self.api.formdefs

    @cached_property

    def roles(self):

        return self.api.roles

    @cached_property

    def categories(self):

        return self.api.categories

                    self.logger.error(u'failed to retrieve formdef %s, %s', formdef.slug, e)

            # keep temporary schema alive for debugging

import six

def exception_to_text(e):

    try:

        return six.text_type(e)

    except Exception:

        pass

    try:

        return six.text_type(e.decode('utf8'))

    except Exception:

        pass

    try:

        return six.text_type(repr(e))

    except Exception:

        pass

    try:

        args = e.args

        try:

            content = six.text_type(repr(args)) if args != [] else ''

        except Exception:

            content = '<exception-while-rendering-args>'

    except AttributeError:

        content = ''

    return u'%s(%s)' % (e.__class__.__name__, content)

    def __init__(self, message, **kwargs):

        super(WcsApiError, self).__init__(message)

        self.kwargs = kwargs

    def __str__(self):

        kwargs = self.kwargs.copy()

        if 'exception' in kwargs:

            kwargs['exception'] = exception_to_text(kwargs['exception'])

        return '%s: %s' % (self.args[0], ' '.join('%s=%s' % (key, value) for key, value in kwargs.items()))

        except requests.RequestException as e:

            raise WcsApiError('GET request failed', url=signed_url, exception=e)

            if not response.ok:

                try:

                    text = response.text

                except UnicodeError:

                    text = '<undecodable>' + repr(response.content)

                raise WcsApiError('GET response is not 200',

                                  url=signed_url,

                                  status_code=response.status_code,

                                  content=text)

            except ValueError as e:

                raise WcsApiError('Invalid JSON content', url=signed_url, exception=e)