import os, sys, io, re, unittest, json, enum, textwrap, collections as cs
import yaml
try: import pyaml
except ImportError:
sys.path.insert(1, os.path.join(__file__, *['..']*3))
import pyaml
try: import unidecode
except ImportError: unidecode = None
large_yaml = br'''
### Default (baseline) configuration parameters.
### DO NOT ever change this config, use -c commandline option instead!
# Note that this file is YAML, so YAML types can be used here, see http://yaml.org/type/
# For instance, large number can be specified as "10_000_000" or "!!float 10e6".
source:
# Path or glob pattern (to match path) to backup, required
path: # example: /srv/backups/weekly.*
queue:
# Path to intermediate backup queue-file (list of paths to upload), required
path: # example: /srv/backups/queue.txt
# Don't rebuild queue-file if it's newer than source.path
check_mtime: true
entry_cache:
# Path to persistent db (sqlite) of remote directory nodes, required
path: # example: /srv/backups/dentries.sqlite
# How to pick a path among those matched by "path" glob
pick_policy: alphasort_last # only one supported
destination:
# URL of Tahoe-LAFS node webapi
url: http://localhost:3456/uri
result: # what to do with a cap (URI) of a resulting tree (with full backup)
print_to_stdout: true
# Append the entry to the specified file (creating it, if doesn't exists)
# Example entry: "2012-10-10T23:12:43.904543 /srv/backups/weekly.2012-10-10 URI:DIR2-CHK:..."
append_to_file: # example: /srv/backups/lafs_caps
# Append the entry to specified tahoe-lafs directory (i.e. put it into that dir)
append_to_lafs_dir: # example: URI:DIR2:...
encoding:
xz:
enabled: true
options: # see lzma.LZMAOptions, empty = module defaults
min_size: 5120 # don't compress files smaller than 5 KiB (unless overidden in "path_filter")
path_filter:
# List of include/exclude regexp path-rules, similar to "filter" section below.
# Same as with "filter", rules can be tuples with '+' or '-' (implied for strings) as first element.
# '+' will indicate that file is compressible, if it's size >= "min_size" option.
# Unlike "filter", first element of rule-tuple can also be a number,
# overriding "min_size" parameter for matched (by that rule) paths.
# If none of the patterns match path, file is handled as if it was matched by '+' rule.
- '\.(gz|bz2|t[gb]z2?|xz|lzma|7z|zip|rar)$'
- '\.(rpm|deb|iso)$'
- '\.(jpe?g|gif|png|mov|avi|ogg|mkv|webm|mp[34g]|flv|flac|ape|pdf|djvu)$'
- '\.(sqlite3?|fossil|fsl)$'
- '\.git/objects/[0-9a-f]+/[0-9a-f]+$'
# - [500, '\.(txt|csv|log|md|rst|cat|(ba|z|k|c|fi)?sh|env)$']
# - [500, '\.(cgi|py|p[lm]|php|c|h|[ce]l|lisp|hs|patch|diff|xml|xsl|css|x?html[45]?|js)$']
# - [500, '\.(co?nf|cfg?|li?st|ini|ya?ml|jso?n|vg|tab)(\.(sample|default|\w+-new))?$']
# - [500, '\.(unit|service|taget|mount|desktop|rules|rc|menu)$']
# - [2000, '^/etc/']
http:
request_pool_options:
maxPersistentPerHost: 10
cachedConnectionTimeout: 600
retryAutomatically: true
ca_certs_files: /etc/ssl/certs/ca-certificates.crt # can be a list
debug_requests: false # insecure! logs will contain tahoe caps
filter:
# Either tuples like "[action ('+' or '-'), regexp]" or just exclude-patterns (python
# regexps) to match relative (to source.path, starting with "/") paths to backup.
# Patterns are matched against each path in order they're listed here.
# Leaf directories are matched with the trailing slash
# (as with rsync) to be distinguishable from files with the same name.
# If path doesn't match any regexp on the list, it will be included.
#
# Examples:
# - ['+', '/\.git/config$'] # backup git repository config files
# - '/\.git/' # *don't* backup any repository objects
# - ['-', '/\.git/'] # exactly same thing as above (redundant)
# - '/(?i)\.?svn(/.*|ignore)$' # exclude (case-insensitive) svn (or .svn) paths and ignore-lists
- '/(CVS|RCS|SCCS|_darcs|\{arch\})/$'
- '/\.(git|hg|bzr|svn|cvs)(/|ignore|attributes|tags)?$'
- '/=(RELEASE-ID|meta-update|update)$'
operation:
queue_only: false # only generate upload queue file, don't upload anything
reuse_queue: false # don't generate upload queue file, use existing one as-is
disable_deduplication: false # make no effort to de-duplicate data (should still work on tahoe-level for files)
# Rate limiting might be useful to avoid excessive cpu/net usage on nodes,
# and especially when uploading to rate-limited api's (like free cloud storages).
# Only used when uploading objects to the grid, not when building queue file.
# Format of each value is "interval[:burst]", where "interval" can be specified as rate (e.g. "1/3e5").
# Simple token bucket algorithm is used. Empty values mean "no limit".
# Examples:
# "objects: 1/10:50" - 10 objects per second, up to 50 at once (if rate was lower before).
# "objects: 0.1:50" - same as above.
# "objects: 10:20" - 1 object in 10 seconds, up to 20 at once.
# "objects: 5" - make interval between object uploads equal 5 seconds.
# "bytes: 1/3e6:50e6" - 3 MB/s max, up to 50 MB/s if connection was underutilized before.
rate_limit:
bytes: # limit on rate of *file* bytes upload, example: 1/3e5:20e6
objects: # limit on rate of uploaded objects, example: 10:50
logging: # see http://docs.python.org/library/logging.config.html
# "custom" level means WARNING/DEBUG/NOISE, depending on CLI options
warnings: true # capture python warnings
sql_queries: false # log executed sqlite queries (very noisy, caps will be there)
version: 1
formatters:
basic:
format: '%(asctime)s :: %(name)s :: %(levelname)s: %(message)s'
datefmt: '%Y-%m-%d %H:%M:%S'
handlers:
console:
class: logging.StreamHandler
stream: ext://sys.stderr
formatter: basic
level: custom
debug_logfile:
class: logging.handlers.RotatingFileHandler
filename: /srv/backups/debug.log
formatter: basic
encoding: utf-8
maxBytes: 5242880 # 5 MiB
backupCount: 2
level: NOISE
loggers:
twisted:
handlers: [console]
level: 0
root:
level: custom
handlers: [console]
'''
class test_const(enum.IntEnum):
dispatch = 2455
heartbeat = 123
data = dict(
path='/some/path',
query_dump=cs.OrderedDict([
('key1', 'тест1'),
('key2', 'тест2'),
('key3', 'тест3'),
('последний', None) ]),
ids=cs.OrderedDict(),
a=[1,None,'asd', 'не-ascii'], b=3.5, c=None, d=test_const.dispatch,
asd=cs.OrderedDict([('b', 1), ('a', 2)]) )
data['query_dump_clone'] = data['query_dump']
data['ids']['id в уникоде'] = [4, 5, 6]
data['ids']['id2 в уникоде'] = data['ids']['id в уникоде']
data["'asd'\n!\0\1"] = cs.OrderedDict([('b', 1), ('a', 2)])
data_str_multiline = dict(cert=(
'-----BEGIN CERTIFICATE-----\n'
'MIIDUjCCAjoCCQD0/aLLkLY/QDANBgkqhkiG9w0BAQUFADBqMRAwDgYDVQQKFAdm\n'
'Z19jb3JlMRYwFAYDVQQHEw1ZZWthdGVyaW5idXJnMR0wGwYDVQQIExRTdmVyZGxv\n'
'dnNrYXlhIG9ibGFzdDELMAkGA1UEBhMCUlUxEjAQBgNVBAMTCWxvY2FsaG9zdDAg\n'
'Fw0xMzA0MjQwODUxMTRaGA8yMDUzMDQxNDA4NTExNFowajEQMA4GA1UEChQHZmdf\n'
'Y29yZTEWMBQGA1UEBxMNWWVrYXRlcmluYnVyZzEdMBsGA1UECBMUU3ZlcmRsb3Zz\n'
'a2F5YSBvYmxhc3QxCzAJBgNVBAYTAlJVMRIwEAYDVQQDEwlsb2NhbGhvc3QwggEi\n'
'MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCnZr3jbhfb5bUhORhmXOXOml8N\n'
'fAli/ak6Yv+LRBtmOjke2gFybPZFuXYr0lYGQ4KgarN904vEg7WUbSlwwJuszJxQ\n'
'Lz3xSDqQDqF74m1XeBYywZQIywKIbA/rfop3qiMeDWo3WavYp2kaxW28Xd/ZcsTd\n'
'bN/eRo+Ft1bor1VPiQbkQKaOOi6K8M9a/2TK1ei2MceNbw6YrlCZe09l61RajCiz\n'
'y5eZc96/1j436wynmqJn46hzc1gC3APjrkuYrvUNKORp8y//ye+6TX1mVbYW+M5n\n'
'CZsIjjm9URUXf4wsacNlCHln1nwBxUe6D4e2Hxh2Oc0cocrAipxuNAa8Afn5AgMB\n'
'AAEwDQYJKoZIhvcNAQEFBQADggEBADUHf1UXsiKCOYam9u3c0GRjg4V0TKkIeZWc\n'
'uN59JWnpa/6RBJbykiZh8AMwdTonu02g95+13g44kjlUnK3WG5vGeUTrGv+6cnAf\n'
'4B4XwnWTHADQxbdRLja/YXqTkZrXkd7W3Ipxdi0bDCOSi/BXSmiblyWdbNU4cHF/\n'
'Ex4dTWeGFiTWY2upX8sa+1PuZjk/Ry+RPMLzuamvzP20mVXmKtEIfQTzz4b8+Pom\n'
'T1gqPkNEbe2j1DciRNUOH1iuY+cL/b7JqZvvdQK34w3t9Cz7GtMWKo+g+ZRdh3+q\n'
'2sn5m3EkrUb1hSKQbMWTbnaG4C/F3i4KVkH+8AZmR9OvOmZ+7Lo=\n'
'-----END CERTIFICATE-----' ))
data_str_long = dict(cert=(
'MIIDUjCCAjoCCQD0/aLLkLY/QDANBgkqhkiG9w0BAQUFADBqMRAwDgYDVQQKFAdm'
'Z19jb3JlMRYwFAYDVQQHEw1ZZWthdGVyaW5idXJnMR0wGwYDVQQIExRTdmVyZGxv'
'dnNrYXlhIG9ibGFzdDELMAkGA1UEBhMCUlUxEjAQBgNVBAMTCWxvY2FsaG9zdDAg'
'Fw0xMzA0MjQwODUxMTRaGA8yMDUzMDQxNDA4NTExNFowajEQMA4GA1UEChQHZmdf'
'Y29yZTEWMBQGA1UEBxMNWWVrYXRlcmluYnVyZzEdMBsGA1UECBMUU3ZlcmRsb3Zz'
'a2F5YSBvYmxhc3QxCzAJBgNVBAYTAlJVMRIwEAYDVQQDEwlsb2NhbGhvc3QwggEi'
'MA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCnZr3jbhfb5bUhORhmXOXOml8N'
'fAli/ak6Yv+LRBtmOjke2gFybPZFuXYr0lYGQ4KgarN904vEg7WUbSlwwJuszJxQ'
'Lz3xSDqQDqF74m1XeBYywZQIywKIbA/rfop3qiMeDWo3WavYp2kaxW28Xd/ZcsTd'
'bN/eRo+Ft1bor1VPiQbkQKaOOi6K8M9a/2TK1ei2MceNbw6YrlCZe09l61RajCiz'
'y5eZc96/1j436wynmqJn46hzc1gC3APjrkuYrvUNKORp8y//ye+6TX1mVbYW+M5n'
'CZsIjjm9URUXf4wsacNlCHln1nwBxUe6D4e2Hxh2Oc0cocrAipxuNAa8Afn5AgMB'
'AAEwDQYJKoZIhvcNAQEFBQADggEBADUHf1UXsiKCOYam9u3c0GRjg4V0TKkIeZWc'
'uN59JWnpa/6RBJbykiZh8AMwdTonu02g95+13g44kjlUnK3WG5vGeUTrGv+6cnAf'
'4B4XwnWTHADQxbdRLja/YXqTkZrXkd7W3Ipxdi0bDCOSi/BXSmiblyWdbNU4cHF/'
'Ex4dTWeGFiTWY2upX8sa+1PuZjk/Ry+RPMLzuamvzP20mVXmKtEIfQTzz4b8+Pom'
'T1gqPkNEbe2j1DciRNUOH1iuY+cL/b7JqZvvdQK34w3t9Cz7GtMWKo+g+ZRdh3+q'
'2sn5m3EkrUb1hSKQbMWTbnaG4C/F3i4KVkH+8AZmR9OvOmZ+7Lo=' ))
class DumpTests(unittest.TestCase):
def yaml_var(self, ys, raw=False):
ys = textwrap.dedent(ys).replace('\t', ' ')
return ys if raw else yaml.safe_load(ys)
def flatten(self, data, path=tuple()):
dst = list()
if isinstance(data, (tuple, list)):
for v in data: dst.extend(self.flatten(v, path + ('!!list',)))
elif isinstance(data, dict):
for k,v in data.items(): dst.extend(self.flatten(v, path + (k,)))
else: dst.append((path, data))
return tuple(sorted(dst, key=lambda v: json.dumps(v, sort_keys=True)))
def pos_list(self, ys, sep='\n'):
pos, pos_list = 0, list()
while True:
pos = ys.find(sep, pos+1)
if pos < 0: break
pos_list.append(pos)
return pos_list
def empty_line_list(self, ys):
return list(n for n, line in enumerate(ys.splitlines()) if not line)
def test_dst(self):
buff = io.BytesIO()
self.assertIs(pyaml.dump(data, buff), None)
self.assertIsInstance(pyaml.dump(data, str), str)
self.assertIsInstance(pyaml.dump(data, bytes), bytes)
def test_simple(self):
a = self.flatten(data)
b = pyaml.dump(data)
self.assertEqual(a, self.flatten(yaml.safe_load(b)))
def test_vspacing(self):
data = yaml.safe_load(large_yaml)
a = self.flatten(data)
b = pyaml.dump(data, vspacing=dict(split_lines=10, split_count=2))
self.assertEqual(a, self.flatten(yaml.safe_load(b)))
self.assertEqual( self.pos_list(b, '\n'),
[12, 13, 25, 33, 52, 73, 88, 107, 157, 184, 264, 299, 344, 345, 355, 375, 399,
424, 425, 458, 459, 467, 505, 561, 600, 601, 607, 660, 681, 705, 738, 767, 795,
796, 805, 806, 820, 831, 866, 936, 937, 949, 950, 963, 998, 1021, 1041, 1072,
1073, 1092, 1113, 1163, 1185, 1224, 1247, 1266, 1290, 1291, 1302, 1315, 1331,
1349, 1364, 1365, 1373, 1387, 1403, 1421, 1422, 1440, 1441, 1454, 1455, 1471,
1472, 1483, 1511, 1528, 1542, 1553, 1566, 1584, 1585, 1593, 1608, 1618, 1626,
1656, 1665, 1686, 1696] )
b = pyaml.dump(data, vspacing=False)
self.assertNotIn('\n\n', b)
def test_ids(self):
b = pyaml.dump(data, force_embed=False)
self.assertNotIn('&id00', b)
self.assertIn('query_dump_clone: *query_dump_clone', b)
self.assertIn('id в уникоде: &ids_-_id2_', b)
if not unidecode: self.assertIn('id в уникоде: &ids_-_id2__id00', b)
def test_ids_unidecode(self):
if not unidecode:
self.skipTest('No unidecode module to test ids from non-ascii keys')
b = pyaml.dump(data, force_embed=False)
self.assertNotIn('&id00', b)
self.assertNotIn('_id00', b)
self.assertIn('id в уникоде: &ids_-_id2_v_unikode', b)
def test_force_embed(self):
for check, fe in (self.assertNotIn, True), (self.assertIn, False):
dump = pyaml.dump(data, force_embed=fe)
for c in '*&': check(c, dump)
def test_encoding(self):
b = pyaml.dump(data, force_embed=True)
b_lines = list(map(str.strip, b.splitlines()))
chk = ['query_dump:', 'key1: тест1', 'key2: тест2', 'key3: тест3', 'последний:']
pos = b_lines.index('query_dump:')
self.assertEqual(b_lines[pos:pos + len(chk)], chk)
def test_str_long(self):
b = pyaml.dump(data_str_long)
self.assertNotIn('"', b)
self.assertNotIn("'", b)
self.assertEqual(len(b.splitlines()), 1)
def test_str_multiline(self):
b = pyaml.dump(data_str_multiline)
b_lines = b.splitlines()
self.assertGreater(len(b_lines), len(data_str_multiline['cert'].splitlines()))
for line in b_lines: self.assertLess(len(line), 100)
def test_dumps(self):
b = pyaml.dumps(data_str_multiline)
self.assertIsInstance(b, str)
def test_print(self):
self.assertIs(pyaml.print, pyaml.pprint)
self.assertIs(pyaml.print, pyaml.p)
buff = io.BytesIO()
b = pyaml.dump(data_str_multiline, dst=bytes)
pyaml.print(data_str_multiline, file=buff)
self.assertEqual(b, buff.getvalue())
def test_print_args(self):
buff = io.BytesIO()
args = 1, 2, 3
b = pyaml.dump(args, dst=bytes)
pyaml.print(*args, file=buff)
self.assertEqual(b, buff.getvalue())
def test_str_styles(self):
a = pyaml.dump(data_str_multiline)
b = pyaml.dump(data_str_multiline, string_val_style='|')
self.assertEqual(a, b)
b = pyaml.dump(data_str_multiline, string_val_style='plain')
self.assertNotEqual(a, b)
c = pyaml.dump(data_str_multiline, string_val_style='literal')
self.assertNotEqual(c, a)
self.assertNotEqual(c, b)
self.assertTrue(pyaml.dump('waka waka', string_val_style='|').startswith('|-\n'))
a = pyaml.dump(data_int := dict(a=123))
self.assertEqual(a, 'a: 123\n')
self.assertEqual(pyaml.dump(data_int, string_val_style='|'), a)
self.assertEqual(pyaml.dump(data_int, string_val_style='literal'), a)
a = pyaml.dump(data_str := dict(a='123'))
b = pyaml.dump(data_str, string_val_style='|')
self.assertEqual(a, "a: '123'\n")
self.assertEqual(self.flatten(data_str), self.flatten(yaml.safe_load(a)))
self.assertNotEqual(a, b)
self.assertEqual(self.flatten(data_str), self.flatten(yaml.safe_load(b)))
def test_colons_in_strings(self):
val1 = {'foo': ['bar:', 'baz', 'bar:bazzo', 'a: b'], 'foo:': 'yak:'}
val1_str = pyaml.dump(val1)
val2 = yaml.safe_load(val1_str)
val2_str = pyaml.dump(val2)
val3 = yaml.safe_load(val2_str)
self.assertEqual(val1, val2)
self.assertEqual(val1_str, val2_str)
self.assertEqual(val2, val3)
def test_unqouted_spaces(self):
val1 = {'key': 'word1 word2 word3', 'key key': 'asd', 'k3': 'word: stuff'}
val1_str = pyaml.dump(val1)
val2 = yaml.safe_load(val1_str)
self.assertEqual(val1, val2)
self.assertIn('key: word1 word2 word3', val1_str)
def test_empty_strings(self):
val1 = {'key': ['', 'stuff', '', 'more'], '': 'value', 'k3': ''}
val1_str = pyaml.dump(val1)
val2 = yaml.safe_load(val1_str)
val2_str = pyaml.dump(val2)
val3 = yaml.safe_load(val2_str)
self.assertEqual(val1, val2)
self.assertEqual(val1_str, val2_str)
self.assertEqual(val2, val3)
def test_single_dash_strings(self):
strip_seq_dash = lambda line: line.lstrip().lstrip('-').lstrip()
val1 = {'key': ['-', '-stuff', '- -', '- more-', 'more-', '--']}
val1_str = pyaml.dump(val1)
val2 = yaml.safe_load(val1_str)
val2_str = pyaml.dump(val2)
val3 = yaml.safe_load(val2_str)
self.assertEqual(val1, val2)
self.assertEqual(val1_str, val2_str)
self.assertEqual(val2, val3)
val1_str_lines = val1_str.splitlines()
self.assertEqual(strip_seq_dash(val1_str_lines[2]), '-stuff')
self.assertEqual(strip_seq_dash(val1_str_lines[5]), 'more-')
self.assertEqual(strip_seq_dash(val1_str_lines[6]), '--')
val1 = {'key': '-'}
val1_str = pyaml.dump(val1)
val2 = yaml.safe_load(val1_str)
val2_str = pyaml.dump(val2)
val3 = yaml.safe_load(val2_str)
def test_namedtuple(self):
TestTuple = cs.namedtuple('TestTuple', 'y x z')
val = TestTuple(1, 2, 3)
val_str = pyaml.dump(val, sort_keys=False)
self.assertEqual(val_str, 'y: 1\nx: 2\nz: 3\n') # namedtuple order was preserved
def test_ordereddict(self):
d = cs.OrderedDict((i, '') for i in reversed(range(10)))
lines = pyaml.dump(d, sort_keys=False).splitlines()
self.assertEqual(lines, list(reversed(sorted(lines))))
def test_enum(self):
c = test_const.heartbeat
d1 = {'a': c, 'b': c.value, c: 'testx'}
self.assertEqual(d1['a'], d1['b'])
s = pyaml.dump(d1)
d2 = yaml.safe_load(s)
self.assertEqual(d1['a'], d2['a'])
self.assertEqual(d1['a'], c)
self.assertEqual(d1[c], 'testx')
self.assertIn('a: 123 # test_const.heartbeat', s)
def test_pyyaml_params(self):
d = {'foo': 'lorem ipsum ' * 30} # 300+ chars
for w in 40, 80, 200:
lines = pyaml.dump(d, width=w, indent=10).splitlines()
for n, line in enumerate(lines, 1):
self.assertLess(len(line), w*1.2)
if n != len(lines): self.assertGreater(len(line), w*0.8)
def test_multiple_docs(self):
docs = [yaml.safe_load(large_yaml), dict(a=1, b=2, c=3)]
docs_str = pyaml.dump_all(docs, vspacing=True)
self.assertTrue(docs_str.startswith('---'))
self.assertIn('---\n\na: 1\n\nb: 2\n\nc: 3\n', docs_str)
docs_str2 = pyaml.dump(docs, vspacing=True, multiple_docs=True)
self.assertEqual(docs_str, docs_str2)
docs_str2 = pyaml.dump(docs, vspacing=True)
self.assertNotEqual(docs_str, docs_str2)
docs_str2 = pyaml.dump_all(docs, explicit_start=False)
self.assertFalse(docs_str2.startswith('---'))
self.assertNotEqual(docs_str, docs_str2)
docs_str = pyaml.dump(docs, multiple_docs=True, explicit_start=False)
self.assertEqual(docs_str, docs_str2)
def test_ruamel_yaml(self):
try: from ruamel.yaml import YAML
except ImportError: self.skipTest('No ruamel.yaml module to test it')
data = YAML(typ='safe').load(large_yaml)
yaml_str = pyaml.dump(data)
def test_dump_stream_kws(self):
data = [1, 2, 3]
buff1, buff2 = io.StringIO(), io.StringIO()
pyaml.dump(data, dst=buff1)
pyaml.dump(data, stream=buff2)
self.assertEqual(buff1.getvalue(), buff2.getvalue())
buff1.seek(0); buff1.truncate()
pyaml.dump(data, dst=buff1, stream=buff1)
self.assertEqual(buff1.getvalue(), buff2.getvalue())
ys = pyaml.dump(data, dst=str, stream=str)
self.assertEqual(ys, buff2.getvalue())
buff1.seek(0); buff1.truncate(); buff2.seek(0); buff2.truncate()
with self.assertRaises(TypeError):
pyaml.dump(data, dst=buff1, stream=buff2)
with self.assertRaises(TypeError):
pyaml.dump(data, dst=str, stream=buff2)
self.assertEqual(buff1.getvalue(), '')
self.assertEqual(buff2.getvalue(), '')
def test_vspacing_limits(self):
itm = self.yaml_var('''
builtIn: 1
datasource:
type: grafana
uid: -- Grafana --
enable: yes
hide: yes
iconColor: rgba(0, 211, 255, 1)
name: Annotations & Alerts
type: dashboard''')
ys = pyaml.dump(dict(mylist=[itm]*10))
self.assertEqual(
self.empty_line_list(ys),
[1, 11, 21, 31, 41, 51, 61, 71, 81, 91] )
ys = self.yaml_var('''
panels:
- datasource:
type: datasource
uid: grafana
fieldConfig:''', raw=True)
for n in range(60): ys += '\n' + ' '*3 + f'field{n}: value-{n}'
ys = pyaml.dump(yaml.safe_load(ys), vspacing=dict(oneline_split=True))
self.assertEqual(self.empty_line_list(ys), list(range(4, 126, 2)))
ys = pyaml.dump(yaml.safe_load(ys))
self.assertEqual(self.empty_line_list(ys), [4])
def test_split_block_types(self):
data = self.yaml_var('''
test: # list items should be split
- key1: 1
key2: A
- key1: 1
key2: A
- key1: 1
key2: A
test2: # list items should be split
- key1:
key11: A
key12: B
- key2:
key21: A
- val
test3: # only top-level keys should be split
key1:
key11:
key12:
key2:
- key21: A
- key22: B
- key23: C
key3:
- key31:
- key32:
- key33:
''')
ys = pyaml.dump(data, vspacing=dict(split_lines=0))
self.assertEqual(self.empty_line_list(ys), [1, 4, 7, 10, 12, 16, 19, 21, 23, 27, 32])
def test_anchor_cutoff(self):
data = self.yaml_var('''
similique-natus-inventore-deserunt-amet-explicabo-cum-accusamus-temporibus:
quam-nulla-dolorem-dolore-velit-quis-deserunt-est-ullam-exercitationem:
culpa-quia-incidunt-accusantium-ad-dicta-nobis-rerum-veritatis: &test
test: 1
similique-commodi-aperiam-libero-error-eos-quidem-eius:
ipsam-labore-enim,-vero-voluptatem-eaque-dolores-blanditiis-recusandae:
quas-atque-maxime-itaque-ullam-sequi-suscipit-quis-vitae-veritatis: *test''')
ys = pyaml.dump(data, force_embed=False)
for c in '&', r'\*':
self.assertTrue(m := re.search(fr'(?<= ){c}\S+', ys))
self.assertLess(len(m[0]), 50)
self.assertIn('similique', m[0])
self.assertIn('veritatis', m[0])
data = dict(test1=dict(test2=(v := dict(a=1, b=2, c=3))), test3=dict(test4=v))
ys = pyaml.dump(data, force_embed=False)
for c in '&', r'\*':
self.assertTrue(m := re.search(fr'(?<= ){c}\S+', ys))
self.assertLess(len(m[0]), 30)
self.assertEqual(len(re.findall(r'test\d', m[0])), 2)
def test_group_oneline_values(self):
data = self.yaml_var('''
similique-natus: 1
similique-commodi:
aperiam-libero: 2
'vel praesentium quo':
exercitationem debitis: porro beatae id
rerum commodi ipsum: nesciunt veritatis
amet quaerat:
assumenda: odio tenetur saepe
"111": digit-string
deserunt-est-2: asdasd
deserunt-est-1: |
line1
line2
culpa-quia: 1234
deserunt-est-3: asdasd
10: test1
200: test
30: test2''')
# Sort and oneline-group
ys1 = pyaml.dump( data,
sort_dicts=pyaml.PYAMLSort.oneline_group,
vspacing=dict(split_lines=0, split_count=0) )
self.assertEqual(self.empty_line_list(ys1), [8, 12, 15, 19])
# No oneline-group by default
ys2 = pyaml.dump(data, vspacing=dict(split_lines=0, split_count=0))
self.assertNotEqual(ys1, ys2)
self.assertEqual( self.empty_line_list(ys2),
[1, 4, 6, 9, 11, 13, 15, 17, 21, 23, 25, 27, 29] )
# No-sort oneline-group overrides oneline-split for conseq oneliners
ys3 = pyaml.dump(data, sort_keys=False, vspacing=dict(
split_lines=0, split_count=0, oneline_group=True, oneline_split=True ))
self.assertNotEqual(ys1, ys3)
self.assertNotEqual(ys2, ys3)
self.assertEqual(self.empty_line_list(ys3), [1, 4, 8, 11, 14, 18])
def test_repr_undefined(self):
class TestType1:
def __repr__(self): return large_yaml.decode()
class TestType2:
def __repr__(self): return '# should end up quoted' + ' '*100
unknown3 = type('TestType3', (object,), dict(a='b'))()
with open('/dev/null') as unknown4:
data = dict( known='val', unknown1=TestType1(),
unknown2=TestType2(), unknown3=unknown3, unknown4=unknown4 )
with self.assertRaises(yaml.YAMLError): ys = pyaml.dump(data)
with self.assertRaises(yaml.YAMLError): ys = pyaml.dump(unknown3)
self.assertIn(' ...[1/', pyaml.dump(data, repr_unknown=1))
ys = pyaml.dump(data, repr_unknown=True)
self.assertEqual(len(ys.splitlines()), 5)
self.assertIn('.test_dump.TestType3', ys)
self.assertIn('TextIOWrapper', ys)
ys = pyaml.dump(data, width=30, repr_unknown=30)
self.assertEqual(len(ys.splitlines()), 5)
self.assertIn(' ...[30/', ys)
self.assertNotIn('unknown2: # should end up quoted', ys)
self.assertIn('.test_dump.TestType3', ys)
def test_long_value_split(self):
data = dict(
sshkey=dict(__XML__PARAM__line=dict(__XML__value=f'ssh-rsa {"X"*372} user@host')),
vlan=[{'__XML__PARAM__vlan-id-create-delete': dict(__XML__value='1, 30, 50,'
' 110, 121-126, 999-1001, 1010, 1012, 1014, 1016, 1018, 1020-1022, 1060-1063,'
' 1080-1083, 1111-1113, 2000, 2201-2202, 2206, 2208, 2210, 2254, 3000, 3002,'
' 3006, 3008, 3021, 3100-3102, 3106, 3108, 3121, 3322, 3326, 3328, 3448-3452,'
' 3455-3458, 3950, 3955, 3960, 3965-3966')}] )
ys = pyaml.dump(data)
self.assertEqual(len(ys.splitlines()), 9)
self.assertEqual(yaml.safe_load(ys), data)
if __name__ == '__main__':
unittest.main()
# print('-'*80)
# pyaml.dump(yaml.safe_load(large_yaml), sys.stdout)
# print('-'*80)
# pyaml.dump(data, sys.stdout)