|
|
@@ -29,9 +29,15 @@ import functools |
|
|
|
import itertools |
|
|
|
import os |
|
|
|
import pathlib |
|
|
|
import shutil |
|
|
|
import subprocess |
|
|
|
import sys |
|
|
|
import tempfile |
|
|
|
import unittest |
|
|
|
|
|
|
|
from unittest.mock import patch |
|
|
|
from subprocess import DEVNULL, PIPE |
|
|
|
|
|
|
|
from Strobe.Strobe import Strobe, KeccakF |
|
|
|
from Strobe.Strobe import AuthenticationFailed |
|
|
|
|
|
|
@@ -375,7 +381,7 @@ async def main(): |
|
|
|
|
|
|
|
# seed the RNG |
|
|
|
prngseed = os.urandom(64) |
|
|
|
syote_comms.strobe_seed_prng((c_uint8 * |
|
|
|
syote_comms._lib.strobe_seed_prng((c_uint8 * |
|
|
|
len(prngseed))(*prngseed), len(prngseed)) |
|
|
|
|
|
|
|
# Create the state for testing |
|
|
@@ -1535,3 +1541,77 @@ class TestLoRaNodeMulticast(unittest.IsolatedAsyncioTestCase): |
|
|
|
|
|
|
|
with self.assertRaises(asyncio.CancelledError): |
|
|
|
await task |
|
|
|
|
|
|
|
class TestLORAmain(unittest.TestCase): |
|
|
|
def setUp(self): |
|
|
|
# setup temporary directory |
|
|
|
d = pathlib.Path(tempfile.mkdtemp()).resolve() |
|
|
|
self.basetempdir = d |
|
|
|
self.tempdir = d / 'subdir' |
|
|
|
self.tempdir.mkdir() |
|
|
|
self.origcwd = os.getcwd() |
|
|
|
|
|
|
|
os.chdir(pathlib.Path(__file__).parent) |
|
|
|
|
|
|
|
def tearDown(self): |
|
|
|
btd = self.basetempdir |
|
|
|
self.tempdir = None |
|
|
|
self.basetempdir = None |
|
|
|
shutil.rmtree(btd) |
|
|
|
|
|
|
|
os.chdir(self.origcwd) |
|
|
|
self.origcwd = None |
|
|
|
|
|
|
|
def test_sharedkey(self): |
|
|
|
keys = [] |
|
|
|
|
|
|
|
# shared key test |
|
|
|
keyfile = self.basetempdir / 'shared.key' |
|
|
|
with keyfile.open('w') as fp: |
|
|
|
fp.write(os.urandom(16).hex()) |
|
|
|
|
|
|
|
keys.append(('shared', [ [ '-s', keyfile ] ] * 2)) |
|
|
|
|
|
|
|
# pub key test |
|
|
|
init_key = X25519.gen() |
|
|
|
resp_key = X25519.gen() |
|
|
|
|
|
|
|
for name, (initargs, respargs) in keys: |
|
|
|
with self.subTest(name=name): |
|
|
|
self.run_program_keyargs(initargs, respargs) |
|
|
|
|
|
|
|
def run_program_keyargs(self, initargs, respargs): |
|
|
|
try: |
|
|
|
# XXX - macosx specific library path var |
|
|
|
with patch.dict(os.environ, |
|
|
|
dict(DYLD_LIBRARY_PATH=self.origcwd)): |
|
|
|
resp = subprocess.Popen([ 'python', __file__, |
|
|
|
'-r', 'dummy:TestAttr.dummy_func' ] + |
|
|
|
respargs, stdin=DEVNULL, stderr=PIPE, |
|
|
|
stdout=PIPE) |
|
|
|
|
|
|
|
init = subprocess.Popen([ 'python', __file__ ] + |
|
|
|
initargs + [ 'ping' ], stdin=DEVNULL, stderr=PIPE, |
|
|
|
stdout=DEVNULL) |
|
|
|
|
|
|
|
# make sure the command completes |
|
|
|
initret = init.wait(2) |
|
|
|
self.assertEqual(initret, 0) |
|
|
|
|
|
|
|
# terminate responder |
|
|
|
respret = resp.terminate() |
|
|
|
stdout, errout = resp.communicate() |
|
|
|
|
|
|
|
# make sure we got the message |
|
|
|
self.assertEqual(stdout, b"got msg: b'\\x04'\n") |
|
|
|
|
|
|
|
self.assertEqual(errout, b'') |
|
|
|
|
|
|
|
finally: |
|
|
|
init.terminate() |
|
|
|
_, errout = init.communicate() |
|
|
|
self.assertEqual(errout, b'') |
|
|
|
|
|
|
|
resp.terminate() |
|
|
|
_, errout = resp.communicate() |
|
|
|
self.assertEqual(errout, b'') |