diff --git a/cryptol-remote-api/python/tests/cryptol/test_basics.py b/cryptol-remote-api/python/tests/cryptol/test_basics.py index 28613cc5c..d18f0918b 100644 --- a/cryptol-remote-api/python/tests/cryptol/test_basics.py +++ b/cryptol-remote-api/python/tests/cryptol/test_basics.py @@ -3,6 +3,7 @@ from pathlib import Path import unittest import io +import os import time import cryptol import cryptol.cryptoltypes @@ -68,15 +69,43 @@ def test_check_timeout(self): self.assertLess(t2 - t1, 5) def test_interrupt(self): - c = self.c - c.load_file(str(Path('tests','cryptol','test-files', 'examples','AES.cry'))) + # Check if this test is using a local server, if not we assume it's a remote HTTP server + if os.getenv('CRYPTOL_SERVER') is not None: + c = self.c + c.load_file(str(Path('tests','cryptol','test-files', 'examples','AES.cry'))) + + t1 = time.time() + c.check("\\(bv : [256]) -> ~ (~ (~ (~bv))) == bv", num_tests="all", timeout=30.0) + # ^ .result() intentionally omitted so we don't wait on it's result and we can interrupt + # it on the next line. We add a timeout just in case to the test fails + time.sleep(.5) + c.interrupt() + self.assertTrue(c.safe("aesEncrypt").result()) + t2 = time.time() + self.assertLess(t2 - t1, 15.0) # ensure th interrupt ended things and not the timeout + elif os.getenv('CRYPTOL_SERVER_URL') is not None: + c = self.c + other_c = cryptol.connect(verify=False) + # Since this is the HTTP server, due to client implementation details + # the requests don't return until they get a response, so we fork + # to interrupt the server + newpid = os.fork() + if newpid == 0: + time.sleep(5) + other_c.interrupt() + os._exit(0) + + c.load_file(str(Path('tests','cryptol','test-files', 'examples','AES.cry'))) + + t1 = time.time() + c.check("\\(bv : [256]) -> ~ (~ (~ (~bv))) == bv", num_tests="all", timeout=60.0) + self.assertTrue(c.safe("aesEncrypt").result()) + t2 = time.time() + self.assertLess(t2 - t1, 20.0) # ensure th interrupt ended things and not the timeout + else: + # Otherwise fail... since this shouldn't be possible + self.assertFalse("Impossible") - c.check("\\(bv : [256]) -> ~ (~ (~ (~bv))) == bv", num_tests="all") - # ^ .result() intentionally omitted so we don't wait on it's result and we can interrupt - # it on the next line. - time.sleep(.5) - c.interrupt() - self.assertTrue(c.safe("aesEncrypt").result()) def test_prove_timeout(self): c = self.c