Added base64 IO support.
This commit is contained in:
parent
30c5945f1c
commit
19597e565e
|
@ -53,6 +53,11 @@ class benedict(IODict, KeypathDict, ParseDict):
|
|||
def fromkeys(cls, sequence, value=None):
|
||||
return KeypathDict.fromkeys(sequence, value)
|
||||
|
||||
@staticmethod
|
||||
@benediction
|
||||
def from_base64(s, **kwargs):
|
||||
return IODict.from_base64(s, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
@benediction
|
||||
def from_json(s, **kwargs):
|
||||
|
|
|
@ -49,6 +49,11 @@ class IODict(dict):
|
|||
|
||||
@staticmethod
|
||||
def _from_any_data_string(s, **kwargs):
|
||||
try:
|
||||
d = IODict.from_base64(s, **kwargs)
|
||||
return d
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
d = IODict.from_json(s, **kwargs)
|
||||
return d
|
||||
|
@ -70,6 +75,11 @@ class IODict(dict):
|
|||
except ValueError:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def from_base64(s, **kwargs):
|
||||
return IODict._decode(s,
|
||||
decoder=io_util.decode_base64, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
def from_json(s, **kwargs):
|
||||
return IODict._decode(s,
|
||||
|
@ -90,6 +100,11 @@ class IODict(dict):
|
|||
return IODict._decode(s,
|
||||
decoder=io_util.decode_yaml, **kwargs)
|
||||
|
||||
def to_base64(self, filepath=None, **kwargs):
|
||||
return IODict._encode(self,
|
||||
encoder=io_util.encode_base64,
|
||||
filepath=filepath, **kwargs)
|
||||
|
||||
def to_json(self, filepath=None, **kwargs):
|
||||
return IODict._encode(self,
|
||||
encoder=io_util.encode_json,
|
||||
|
|
|
@ -1,5 +1,6 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import base64
|
||||
import errno
|
||||
import json
|
||||
import os
|
||||
|
@ -8,6 +9,12 @@ import xmltodict
|
|||
import toml
|
||||
import yaml
|
||||
|
||||
def decode_base64(s, **kwargs):
|
||||
decode_func = kwargs.pop('through', decode_json)
|
||||
b = base64.b64decode(s)
|
||||
s = b.decode('utf-8')
|
||||
return decode_func(s, **kwargs)
|
||||
|
||||
|
||||
def decode_json(s, **kwargs):
|
||||
data = json.loads(s, **kwargs)
|
||||
|
@ -31,6 +38,13 @@ def decode_yaml(s, **kwargs):
|
|||
return data
|
||||
|
||||
|
||||
def encode_base64(d, **kwargs):
|
||||
encode_func = kwargs.pop('through', encode_json)
|
||||
data = base64.b64encode(
|
||||
encode_func(d, **kwargs).encode('utf-8')).decode('utf-8')
|
||||
return data
|
||||
|
||||
|
||||
def encode_json(d, **kwargs):
|
||||
data = json.dumps(d, **kwargs)
|
||||
return data
|
||||
|
|
|
@ -0,0 +1,2 @@
|
|||
Lorem ipsum consectetur sint id aute officia sed excepteur consectetur labore laboris dolore in labore consequat ut in eu ut deserunt.
|
||||
Elit aliqua velit aliquip voluptate consequat reprehenderit occaecat dolor ut esse aute laboris cillum fugiat esse est laborum.
|
|
@ -0,0 +1 @@
|
|||
eyJhIjogMSwgImIiOiAyLCAiYyI6IDN9
|
|
@ -455,6 +455,17 @@ class BenedictTestCase(unittest.TestCase):
|
|||
self.assertEqual(b, r)
|
||||
self.assertEqual(type(b), benedict)
|
||||
|
||||
def test_from_base64(self):
|
||||
j = 'eyJhIjogMSwgImIiOiAyLCAiYyI6IDN9'
|
||||
# static method
|
||||
d = benedict.from_base64(j)
|
||||
self.assertTrue(isinstance(d, benedict))
|
||||
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
|
||||
# constructor
|
||||
d = benedict(j)
|
||||
self.assertTrue(isinstance(d, benedict))
|
||||
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
|
||||
|
||||
def test_from_json(self):
|
||||
j = '{"a": 1, "b": 2, "c": 3}'
|
||||
# static method
|
||||
|
|
|
@ -23,6 +23,117 @@ class IODictTestCase(unittest.TestCase):
|
|||
dir_path = os.path.dirname(os.path.realpath(__file__))
|
||||
return os.path.join(dir_path, 'output/{}'.format(filepath))
|
||||
|
||||
# BASE64
|
||||
|
||||
def test_from_base64_with_valid_data(self):
|
||||
j = 'eyJhIjogMSwgImIiOiAyLCAiYyI6IDN9'
|
||||
# j = '{"a": 1, "b": 2, "c": 3}'
|
||||
# static method
|
||||
d = IODict.from_base64(j)
|
||||
self.assertTrue(isinstance(d, dict))
|
||||
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
|
||||
# constructor
|
||||
d = IODict(j)
|
||||
self.assertTrue(isinstance(d, dict))
|
||||
self.assertEqual(d, { 'a': 1, 'b': 2, 'c': 3, })
|
||||
|
||||
def test_from_base64_with_invalid_data(self):
|
||||
j = 'Lorem ipsum est in ea occaecat nisi officia.'
|
||||
# static method
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(j)
|
||||
# constructor
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict(j)
|
||||
|
||||
def test_from_base64_with_valid_file_valid_content(self):
|
||||
filepath = self.input_path('valid-content.base64')
|
||||
# static method
|
||||
d = IODict.from_base64(filepath)
|
||||
self.assertTrue(isinstance(d, dict))
|
||||
# constructor
|
||||
d = IODict(filepath)
|
||||
self.assertTrue(isinstance(d, dict))
|
||||
|
||||
def test_from_base64_with_valid_file_valid_content_invalid_format(self):
|
||||
filepath = self.input_path('valid-content.json')
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(filepath)
|
||||
filepath = self.input_path('valid-content.toml')
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(filepath)
|
||||
filepath = self.input_path('valid-content.xml')
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(filepath)
|
||||
filepath = self.input_path('valid-content.yml')
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(filepath)
|
||||
|
||||
def test_from_base64_with_valid_file_invalid_content(self):
|
||||
filepath = self.input_path('invalid-content.base64')
|
||||
# static method
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(filepath)
|
||||
# constructor
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict(filepath)
|
||||
|
||||
def test_from_base64_with_invalid_file(self):
|
||||
filepath = self.input_path('invalid-file.base64')
|
||||
# static method
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(filepath)
|
||||
# constructor
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict(filepath)
|
||||
|
||||
# def test_from_base64_with_valid_url_valid_content(self):
|
||||
# url = 'https://raw.githubusercontent.com/fabiocaccamo/python-benedict/master/tests/input/valid-content.base64'
|
||||
# # static method
|
||||
# d = IODict.from_base64(url)
|
||||
# self.assertTrue(isinstance(d, dict))
|
||||
# # constructor
|
||||
# d = IODict(url)
|
||||
# self.assertTrue(isinstance(d, dict))
|
||||
|
||||
def test_from_base64_with_valid_url_invalid_content(self):
|
||||
url = 'https://github.com/fabiocaccamo/python-benedict'
|
||||
# static method
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(url)
|
||||
# constructor
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict(url)
|
||||
|
||||
def test_from_base64_with_invalid_url(self):
|
||||
url = 'https://github.com/fabiocaccamo/python-benedict-invalid'
|
||||
# static method
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict.from_base64(url)
|
||||
# constructor
|
||||
with self.assertRaises(ValueError):
|
||||
d = IODict(url)
|
||||
|
||||
def test_to_base64(self):
|
||||
d = IODict({
|
||||
'a': 1,
|
||||
'b': 2,
|
||||
'c': 3,
|
||||
})
|
||||
s = d.to_base64(sort_keys=True)
|
||||
self.assertEqual(s, 'eyJhIjogMSwgImIiOiAyLCAiYyI6IDN9')
|
||||
|
||||
def test_to_base64_file(self):
|
||||
d = IODict({
|
||||
'a': 1,
|
||||
'b': 2,
|
||||
'c': 3,
|
||||
})
|
||||
filepath = self.output_path('test_to_base64_file.base64')
|
||||
s = d.to_base64(filepath=filepath, sort_keys=True)
|
||||
self.assertTrue(d, os.path.isfile(filepath))
|
||||
self.assertEqual(d, IODict.from_base64(filepath))
|
||||
|
||||
# JSON
|
||||
|
||||
def test_from_json_with_valid_data(self):
|
||||
|
|
Loading…
Reference in New Issue