140 lines
6.7 KiB
Python
140 lines
6.7 KiB
Python
from .testlib import TestCase
|
|
|
|
import argparse
|
|
import logging
|
|
import os.path
|
|
import re
|
|
import unittest
|
|
from resync.client_utils import init_logging, count_true_args, parse_links, parse_link, parse_capabilities, parse_capability_lists, add_shared_misc_options, process_shared_misc_options
|
|
from resync.client import ClientFatalError
|
|
from resync.url_or_file_open import CONFIG
|
|
|
|
|
|
class TestClientUtils(TestCase):
|
|
|
|
def test01_init_logging(self):
|
|
# to_file=False, logfile=None, default_logfile='/tmp/resync.log',
|
|
# human=True, verbose=False, eval_mode=False,
|
|
# default_logger='client', extra_loggers=None):
|
|
tmplog = os.path.join(self.tmpdir, 'tmp.log')
|
|
init_logging(to_file=True, default_logfile=tmplog,
|
|
extra_loggers=['x1', 'x2'])
|
|
# check x1 and x2 set, not x3 (can tell by level)
|
|
self.assertTrue(logging.getLogger('x1').level, logging.DEBUG)
|
|
self.assertTrue(logging.getLogger('x2').level, logging.DEBUG)
|
|
self.assertEqual(logging.getLogger('x3').level, 0)
|
|
# write something, check goes to file
|
|
log = logging.getLogger('resync')
|
|
log.warning('PIGS MIGHT FLY')
|
|
logtxt = open(tmplog, 'r').read()
|
|
self.assertTrue(re.search(r'WARNING \| PIGS MIGHT FLY', logtxt))
|
|
|
|
def test02_count_true_args(self):
|
|
self.assertEqual(count_true_args(), 0)
|
|
self.assertEqual(count_true_args(True), 1)
|
|
self.assertEqual(count_true_args(False), 0)
|
|
self.assertEqual(count_true_args(0, 1, 2, 3), 3)
|
|
|
|
def test03_parse_links(self):
|
|
self.assertEqual(parse_links([]), [])
|
|
self.assertEqual(parse_links(['u,h']), [{'href': 'h', 'rel': 'u'}])
|
|
self.assertEqual(parse_links(['u,h', 'v,i']), [
|
|
{'href': 'h', 'rel': 'u'}, {'href': 'i', 'rel': 'v'}])
|
|
self.assertRaises(ClientFatalError, parse_links, 'xx')
|
|
self.assertRaises(ClientFatalError, parse_links, ['u'])
|
|
self.assertRaises(ClientFatalError, parse_links, ['u,h', 'u'])
|
|
|
|
def test04_parse_link(self):
|
|
# Input string of the form: rel,href,att1=val1,att2=val2
|
|
self.assertEqual(parse_link('u,h'), {'href': 'h', 'rel': 'u'})
|
|
self.assertEqual(parse_link('u,h,a=b'), {
|
|
'a': 'b', 'href': 'h', 'rel': 'u'})
|
|
self.assertEqual(parse_link('u,h,a=b,c=d'), {
|
|
'a': 'b', 'c': 'd', 'href': 'h', 'rel': 'u'})
|
|
self.assertEqual(parse_link('u,h,a=b,a=d'), {
|
|
'a': 'd', 'href': 'h', 'rel': 'u'}) # desired??
|
|
self.assertRaises(ClientFatalError, parse_link, '')
|
|
self.assertRaises(ClientFatalError, parse_link, 'u')
|
|
self.assertRaises(ClientFatalError, parse_link, 'u,')
|
|
self.assertRaises(ClientFatalError, parse_link, 'u,h,,')
|
|
self.assertRaises(ClientFatalError, parse_link, 'u,h,a')
|
|
self.assertRaises(ClientFatalError, parse_link, 'u,h,a=')
|
|
self.assertRaises(ClientFatalError, parse_link, 'u,h,a=b,=c')
|
|
|
|
def test05_parse_capabilities(self):
|
|
# Input string of the form: cap_name=uri,cap_name=uri
|
|
# good
|
|
c = parse_capabilities('a=')
|
|
self.assertEqual(len(c), 1)
|
|
self.assertEqual(c['a'], '')
|
|
c = parse_capabilities('a=b,c=')
|
|
self.assertEqual(len(c), 2)
|
|
self.assertEqual(c['a'], 'b')
|
|
# bad
|
|
self.assertRaises(ClientFatalError, parse_capabilities, 'a')
|
|
self.assertRaises(ClientFatalError, parse_capabilities, 'a=b,')
|
|
|
|
def test06_parse_capability_lists(self):
|
|
# Input string of the form: uri,uri
|
|
self.assertEqual(parse_capability_lists('a,b'), ['a', 'b'])
|
|
|
|
def test07_add_shared_misc_options(self):
|
|
"""Test add_shared_misc_options method."""
|
|
parser = argparse.ArgumentParser()
|
|
add_shared_misc_options(parser, default_logfile='/tmp/abc.log')
|
|
args = parser.parse_args(['--hash', 'md5', '--hash', 'sha-1',
|
|
'--checksum',
|
|
'--from', '2020-01-01T01:01:01Z',
|
|
'--exclude', 'ex1', '--exclude', 'ex2',
|
|
'--multifile',
|
|
'--logger', '--logfile', 'log.out',
|
|
'--spec-version', '1.0',
|
|
'-v'])
|
|
self.assertEqual(args.hash, ['md5', 'sha-1'])
|
|
self.assertTrue(args.checksum)
|
|
self.assertEqual(args.from_datetime, '2020-01-01T01:01:01Z')
|
|
self.assertEqual(args.exclude, ['ex1', 'ex2'])
|
|
self.assertTrue(args.multifile)
|
|
self.assertTrue(args.logger)
|
|
self.assertEqual(args.logfile, 'log.out')
|
|
self.assertEqual(args.spec_version, '1.0')
|
|
self.assertTrue(args.verbose)
|
|
# Remote options
|
|
parser = argparse.ArgumentParser()
|
|
add_shared_misc_options(parser, default_logfile='/tmp/abc.log', include_remote=True)
|
|
args = parser.parse_args(['--noauth',
|
|
'--access-token', 'VerySecretToken',
|
|
'--delay', '1.23',
|
|
'--user-agent', 'rc/2.1.1'])
|
|
self.assertTrue(args.noauth)
|
|
self.assertEqual(args.access_token, 'VerySecretToken')
|
|
self.assertEqual(args.delay, 1.23)
|
|
self.assertEqual(args.user_agent, 'rc/2.1.1')
|
|
# Remote options note selected
|
|
parser = argparse.ArgumentParser()
|
|
add_shared_misc_options(parser, default_logfile='/tmp/abc.log', include_remote=False)
|
|
self.assertRaises(SystemExit, parser.parse_args, ['--access-token', 'VerySecretToken'])
|
|
|
|
def test08_process_shared_misc_options(self):
|
|
"""Test process_shared_misc_options method."""
|
|
global CONFIG
|
|
config_copy = CONFIG.copy()
|
|
args = argparse.Namespace(hash=['sha-1'], checksum='md5')
|
|
process_shared_misc_options(args)
|
|
self.assertEqual(args.hash, ['sha-1', 'md5'])
|
|
# Remote options
|
|
args = argparse.Namespace(access_token='ExtraSecretToken',
|
|
delay=2.5,
|
|
user_agent='me',
|
|
checksum=None)
|
|
process_shared_misc_options(args, include_remote=True)
|
|
self.assertEqual(CONFIG['bearer_token'], 'ExtraSecretToken')
|
|
self.assertEqual(CONFIG['delay'], 2.5)
|
|
self.assertEqual(CONFIG['user_agent'], 'me')
|
|
# Negative delay is bad...
|
|
args = argparse.Namespace(access_token=None, delay=-1.0, user_agent=None, checksum=None)
|
|
self.assertRaises(argparse.ArgumentTypeError, process_shared_misc_options, args, include_remote=True)
|
|
# Config is a global so reset back to old version
|
|
for (k, v) in config_copy.items():
|
|
CONFIG[k] = v
|
