robustify nonce management by retry mechanismHEADmaster

1 files changed, 39 insertions, 5 deletions

diff --git a/tests/test_acme_dns_tiny.py b/tests/test_acme_dns_tiny.py
index 0d61b97..3a29ce4 100644
--- a/tests/test_acme_dns_tiny.py
+++ b/tests/test_acme_dns_tiny.py
@@ -39,6 +39,28 @@ key_types = (
         ('ecdsa', 256, False),
 )
 
+# the first case of success expectation, try also to test the anti-replay
+# nonce expiration retry mechanism by waiting for it to expire (more than
+# 5min).
+first_success_case_nonce_timeout_done = False
+original_sreq_method = acme_dns_tiny.ACME.sreq
+
+
+nonce_expiration_sreq_wrapper_counter_max = 3
+nonce_expiration_sreq_wrapper_counter = nonce_expiration_sreq_wrapper_counter_max
+def nonce_expiration_sreq_wrapper(*args, **kwargs):
+    global nonce_expiration_sreq_wrapper_counter
+    nonce_expiration_sreq_wrapper_counter -= 1
+    if nonce_expiration_sreq_wrapper_counter <= 0:
+        nonce_expiration_sreq_wrapper_counter = nonce_expiration_sreq_wrapper_counter_max
+        wait = 30
+        logging.info('waiting %dmin for nonce expiration test', wait)
+        for i in range(wait, 0, -1):
+            logging.info('%d min left', i)
+            time.sleep(60)
+        acme_dns_tiny.ACME.sreq = original_sreq_method # reset original method
+    return original_sreq_method(*args, **kwargs)
+
 
 @contextlib.contextmanager
 def does_not_raise():
@@ -211,14 +233,17 @@ def assert_cert(capsys, args):
     #assert not captured.err
     #certlist = captured.out.split()
 
+    logging.debug('captured stdout %s', captured.out)
+    logging.debug('captured stderr %s', captured.err)
+
     if args['--separator'] is None:
         assert '\0' not in captured.out
     else:
         assert '\0' in captured.out
 
     for chain in captured.out.split('\0'):
-        assert chain.count('-----BEGIN CERTIFICATE-----') == 2
-        assert chain.count('-----END CERTIFICATE-----') == 2
+        assert chain.count('-----BEGIN CERTIFICATE-----') >= 2
+        assert chain.count('-----END CERTIFICATE-----') >= 2
 
         textchain = acme_dns_tiny.openssl(['x509', '-text', '-noout'],
                 stdin=chain)
@@ -228,14 +253,19 @@ def assert_cert(capsys, args):
             'certtool', '-e', '--verify-profile', 'low'], input=chain,
             encoding='utf8')
         assert ' Verified.' in certtool_out
-        assert certtool_out.count('Subject:') == 3
+        assert certtool_out.count('Subject:') >= 3
 
 
-def module_main_caller(*, capsys, args, expectation):
+def module_main_caller(*, capsys, args, expectation, do_expire_nonce):
     logging.info(f'module_main_caller({args}, {expectation})')
 
     logging.debug('before call to acme_dns_tiny.main()')
     with expectation:
+        acme_dns_tiny.ACME.sreq = original_sreq_method
+        if do_expire_nonce:
+            logging.info('doing expire nonce test')
+            first_success_case_nonce_timeout_done = True
+            acme_dns_tiny.ACME.sreq = nonce_expiration_sreq_wrapper
         acme_dns_tiny.main(args)
         # check_cert is under the expectation context manager since if
         # acme_dns_tiny.main() raises, following statement must not be run.
@@ -247,11 +277,15 @@ def test_main(main_args, capsys):
     t_start = time.time()
     args = main_args[0]
     raise_expected = main_args[1]
+    do_expire_nonce = False
     #print('subj', subj, 'args', args)
     expectation = does_not_raise()
     if raise_expected:
         expectation = pytest.raises(ValueError)
-    module_main_caller(capsys=capsys, args=args, expectation=expectation)
+    elif not first_success_case_nonce_timeout_done:
+        do_expire_nonce = True
+    module_main_caller(capsys=capsys, args=args, expectation=expectation,
+                       do_expire_nonce=do_expire_nonce)
     t_stop = time.time()
 
     # calculate for letsencrypt rate limit (50account per hour)