Files
@ 5c80ca07f00c
Branch filter:
Location: Morevna/src/tests/test_overall.py
5c80ca07f00c
2.8 KiB
text/x-python
reformatted whitespace with more respect for PEP-8
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | import os
import shutil
import hashlib
import multiprocessing
from logging import FileHandler
from unittest import TestCase
import config
from hashtree import HashTree
from client import Client, Connection as ClientConnection, DeniedConnection
from server import Miniserver
from . import RedirectedOutput
config.logger.removeHandler(config.handler)
handler = FileHandler("/tmp/morevna.log")
handler.setFormatter(config.formatter)
config.logger.addHandler(handler)
config.batchSize.hash = 8
config.batchSize.data = 8
dataDir = os.path.join(config.directory, "src/tests/data")
filename = os.path.join(dataDir, "test.img")
def compareFiles(f1, f2):
with open(f1, mode="rb") as f:
h2 = hashlib.sha256(f.read()).hexdigest()
with open(f2, mode="rb") as f:
h = hashlib.sha256(f.read()).hexdigest()
return (h, h2)
class TestMorevna(RedirectedOutput, TestCase):
_stdout = None
def setUp(self):
src = os.path.join(dataDir, "test1.img")
shutil.copyfile(src, filename)
@classmethod
def tearDownClass(cls):
super().tearDownClass()
os.remove(filename)
def test_build(self):
treeFile = os.path.join(dataDir, "test.bin")
refFile = os.path.join(dataDir, "test1.bin")
tree = HashTree.fromFile(os.path.join(dataDir, "test1.img"))
tree.save(treeFile)
self.assertEqual(*compareFiles(refFile, treeFile))
os.remove(treeFile)
def test_push(self):
config.port += 1
ms = Miniserver(filename)
p = multiprocessing.Process(target=ms.serve)
p.start()
for clientFile in ("test2.img", "test3.img", "test4.img"):
clientFile = os.path.join(dataDir, clientFile)
c = Client(clientFile)
with ClientConnection("127.0.0.1", config.port) as con:
c.setConnection(con)
c.init("push")
blocksToTransfer = c.negotiate()
c.sendData(blocksToTransfer)
self.assertEqual(*compareFiles(clientFile, filename))
p.terminate()
p.join()
def test_pull(self):
config.port += 1
serverFile = os.path.join(dataDir, "test3.img")
ms = Miniserver(serverFile)
p = multiprocessing.Process(target=ms.serve)
p.start()
c = Client(filename)
with ClientConnection("127.0.0.1", config.port) as con:
c.setConnection(con)
c.init("pull")
blocksToTransfer = c.negotiate()
c.pullData(blocksToTransfer)
self.assertEqual(*compareFiles(serverFile, filename))
p.terminate()
p.join()
def test_deny(self):
config.port += 1
ms = Miniserver(filename)
p = multiprocessing.Process(target=ms.serve)
p.start()
c1 = Client(os.path.join(dataDir, "test2.img"))
with ClientConnection("127.0.0.1", config.port) as con1:
c1.setConnection(con1)
c1.init("push")
c2 = Client(os.path.join(dataDir, "test3.img"))
with ClientConnection("127.0.0.1", config.port) as con2:
c2.setConnection(con2)
with self.assertRaises(DeniedConnection):
c2.init("push")
c1.sendData([]) # to unlock the server
p.terminate()
p.join()
|