Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def check(reload):
result, reloaded = gramex.cache.open(
path, 'template', _reload_status=True, autoescape=None)
eq_(reloaded, reload)
ok_(isinstance(result, Template))
first_line = result.generate(name='高').decode('utf-8').split('\n')[0]
eq_(first_line, '1: name=高')
def check(reload):
result, reloaded = gramex.cache.open(path, 'json', _reload_status=True,
object_pairs_hook=OrderedDict)
eq_(reloaded, reload)
ok_(isinstance(result, OrderedDict))
eq_(result, expected)
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import os
import gramex.cache
import pandas as pd
from pandas.util.testing import assert_frame_equal as afe
from . import TestGramex, folder
class TestFilterHandler(TestGramex):
sales_file = os.path.join(folder, 'sales.xlsx')
sales = gramex.cache.open(sales_file, 'xlsx')
census = gramex.cache.open(sales_file, 'xlsx', sheet_name='census')
def test_filters(self):
def eqframe(result, expected, **kwargs):
'''Same as assert_frame_equal or afe, but does not compare index'''
actual = pd.DataFrame(result)
expected.index = actual.index
afe(actual, expected, **kwargs)
def unique_of(data, cols):
return data.groupby(cols).size().reset_index().drop(0, 1)
self.check('/filters/sales?_c=city')
result = self.get('/filters/sales', params={'_c': ['city']}).json()
expected = unique_of(self.sales, 'city')
eqframe(result['city'], expected, check_like=True)
def check(reload):
result, reloaded = gramex.cache.open(path, 'md', _reload_status=True, encoding='utf-8')
eq_(reloaded, reload)
ok_(isinstance(result, six.text_type))
eq_(result, expected)
return d['a'].sum()
data = gramex.cache.open(path, 'csv', transform=transform2, _cache=cache)
eq_(data, pd.read_csv(path)['a'].sum()) # noqa - ignore encoding
cache_key = (path, 'csv', hashfn(transform2), frozenset([]))
self.assertIn(cache_key, cache)
# Check that non-callable transforms are ignored but used as cache key
data = gramex.cache.open(path, 'csv', transform='ignore', _cache=cache)
assert_frame_equal(data, pd.read_csv(path)) # noqa - ignore encoding
cache_key = (path, 'csv', hashfn('ignore'), frozenset([]))
self.assertIn(cache_key, cache)
# Check that temporary caches are hashed by function
v = 1
data = gramex.cache.open(path, 'csv', lambda x: v, _cache=cache)
eq_(data, 1)
v = 2
data = gramex.cache.open(path, 'csv', lambda x: v, _cache=cache)
eq_(data, 2)
def check(reload):
result, reloaded = gramex.cache.open(path, 'bin', _reload_status=True)
eq_(reloaded, reload)
eq_(result, expected)
def test_open_yaml(self):
path = os.path.join(cache_folder, 'data.yaml')
with io.open(path, encoding='utf-8') as handle:
expected = yaml.load(handle, Loader=AttrDictYAMLLoader)
def check(reload):
result, reloaded = gramex.cache.open(path, 'yaml', _reload_status=True,
Loader=AttrDictYAMLLoader)
eq_(reloaded, reload)
ok_(isinstance(result, AttrDict))
eq_(result, expected)
self.check_file_cache(path, check)
eq_(gramex.cache.open(path), gramex.cache.open(path, 'yaml'))
def error(*args, **kwargs):
tmpl = gramex.cache.open(error_config['path'], 'template', **template_kwargs)
return tmpl.generate(*args, **kwargs)
def load_component(page, **kwargs):
'''return generateed template'''
return gramex.cache.open(page, 'template', rel=True).generate(**kwargs)