Skip to content
Snippets Groups Projects
Commit 29193820 authored by inso's avatar inso
Browse files

Fix tests

parent cc8d80b0
Branches
Tags
No related merge requests found
......@@ -9,6 +9,7 @@ from PyQt5.QtWidgets import QDialog, QDialogButtonBox, QMessageBox, QApplication
from PyQt5.QtCore import QLocale, Qt
from PyQt5.QtTest import QTest
from ucoinpy.api.bma import API
from ucoinpy.documents import Peer
from sakia.tests.mocks.bma import init_new_community
from sakia.core.registry.identities import IdentitiesRegistry
from sakia.gui.certification import CertificationDialog
......@@ -29,8 +30,10 @@ class TestCertificationDialog(unittest.TestCase, QuamashTest):
self.application = Application(self.qapplication, self.lp, self.identities_registry)
self.application.preferences['notifications'] = False
self.mock_new_community = init_new_community.get_mock(self.lp)
self.endpoint = BMAEndpoint("", "127.0.0.1", "", 50010)
self.node = Node("test_currency", [self.endpoint],
self.node = Node(self.mock_new_community.peer(),
"", "HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk",
None, Node.ONLINE,
time.time(), {}, "ucoin", "0.14.0", 0)
......@@ -50,21 +53,15 @@ class TestCertificationDialog(unittest.TestCase, QuamashTest):
self.password_asker.password = "testsakia"
self.password_asker.remember = True
def tearDown(self):
self.tearDownQuamash()
def test_certification_init_community(self):
mock = init_new_community.get_mock(self.lp)
time.sleep(2)
certification_dialog = CertificationDialog(self.application,
self.account,
self.password_asker)
async def open_dialog(certification_dialog):
srv, port, url = await mock.create_server()
srv, port, url = await self.mock_new_community.create_server()
self.addCleanup(srv.close)
self.endpoint.port = port
result = await certification_dialog.async_exec()
self.assertEqual(result, QDialog.Accepted)
......
......@@ -31,8 +31,8 @@ class TestIdentitiesTable(unittest.TestCase, QuamashTest):
self.application = Application(self.qapplication, self.lp, self.identities_registry)
self.application.preferences['notifications'] = False
self.endpoint = BMAEndpoint("", "127.0.0.1", "", 50002)
self.node = Node("test_currency", [self.endpoint],
self.mock_nice_blockchain = nice_blockchain.get_mock(self.lp)
self.node = Node(self.mock_nice_blockchain.peer(),
"", "HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk",
None, Node.ONLINE,
time.time(), {}, "ucoin", "0.14.0", 0)
......@@ -56,7 +56,6 @@ class TestIdentitiesTable(unittest.TestCase, QuamashTest):
self.tearDownQuamash()
def test_search_identity_found(self):
mock = nice_blockchain.get_mock(self.lp)
time.sleep(2)
identities_tab = IdentitiesTabWidget(self.application)
future = asyncio.Future()
......@@ -71,14 +70,13 @@ class TestIdentitiesTable(unittest.TestCase, QuamashTest):
future.set_result(True)
async def exec_test():
srv, port, url = await mock.create_server()
srv, port, url = await self.mock_nice_blockchain.create_server()
self.addCleanup(srv.close)
self.endpoint.port = port
identities_tab.change_account(self.account, self.password_asker)
identities_tab.change_community(self.community)
await asyncio.sleep(1)
urls = [mock.get_request(i).url for i in range(0, 7)]
urls = [self.mock_nice_blockchain.get_request(i).url for i in range(0, 7)]
self.assertTrue('/wot/certifiers-of/7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ' in urls,
msg="Not found in {0}".format(urls))
self.assertTrue('/wot/lookup/7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ' in urls,
......@@ -95,8 +93,8 @@ class TestIdentitiesTable(unittest.TestCase, QuamashTest):
await asyncio.sleep(2)
req = 8
self.assertEqual(mock.get_request(req).method, 'GET')
self.assertEqual(mock.get_request(req).url,
self.assertEqual(self.mock_nice_blockchain.get_request(req).method, 'GET')
self.assertEqual(self.mock_nice_blockchain.get_request(req).url,
'/blockchain/memberships/FADxcH5LmXGmGFgdixSes6nWnC4Vb4pRUBYT81zQRhjn')
req += 1
......
......@@ -63,7 +63,7 @@ class ProcessAddCommunity(unittest.TestCase, QuamashTest):
await asyncio.sleep(1)
self.assertEqual(mock.get_request(0).method, 'GET')
self.assertEqual(mock.get_request(0).url, '/network/peering')
self.assertEqual(process_community._step_init.node._endpoints[0].port, port)
self.assertEqual(process_community._step_init.node.endpoint.port, port)
self.assertEqual(mock.get_request(1).method, 'GET')
self.assertEqual(mock.get_request(1).url,
'/wot/certifiers-of/7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ')
......
......@@ -31,8 +31,8 @@ class TestTransferDialog(unittest.TestCase, QuamashTest):
self.application = Application(self.qapplication, self.lp, self.identities_registry)
self.application.preferences['notifications'] = False
self.endpoint = BMAEndpoint("", "127.0.0.1", "", 50002)
self.node = Node("test_currency", [self.endpoint],
self.mock_nice_blockchain = nice_blockchain.get_mock(self.lp)
self.node = Node(self.mock_nice_blockchain.peer(),
"", "HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk",
None, Node.ONLINE,
time.time(), {}, "ucoin", "0.14.0", 0)
......@@ -56,8 +56,6 @@ class TestTransferDialog(unittest.TestCase, QuamashTest):
self.tearDownQuamash()
def test_transfer_nice_community(self):
mock = nice_blockchain.get_mock(self.lp)
time.sleep(2)
transfer_dialog = TransferMoneyDialog(self.application,
self.account,
self.password_asker,
......@@ -66,9 +64,8 @@ class TestTransferDialog(unittest.TestCase, QuamashTest):
self.account.wallets[0].init_cache(self.application, self.community)
async def open_dialog(transfer_dialog):
srv, port, url = await mock.create_server()
srv, port, url = await self.mock_nice_blockchain.create_server()
self.addCleanup(srv.close)
self.endpoint.port = port
result = await transfer_dialog.async_exec()
self.assertEqual(result, QDialog.Accepted)
......
......@@ -27,8 +27,8 @@ class TestWotTab(unittest.TestCase, QuamashTest):
self.application = Application(self.qapplication, self.lp, self.identities_registry)
self.application.preferences['notifications'] = False
self.endpoint = BMAEndpoint("", "127.0.0.1", "", 50003)
self.node = Node("test_currency", [self.endpoint],
self.mock_nice_blockchain = nice_blockchain.get_mock(self.lp)
self.node = Node(self.mock_nice_blockchain.peer(),
"", "HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk",
None, Node.ONLINE,
time.time(), {}, "ucoin", "0.14.0", 0)
......@@ -52,8 +52,6 @@ class TestWotTab(unittest.TestCase, QuamashTest):
self.tearDownQuamash()
def test_empty_wot_tab(self):
mock = nice_blockchain.get_mock(self.lp)
time.sleep(2)
wot_tab = WotTabWidget(self.application)
future = asyncio.Future()
......@@ -62,9 +60,8 @@ class TestWotTab(unittest.TestCase, QuamashTest):
return future
async def async_open_widget():
srv, port, url = await mock.create_server()
srv, port, url = await self.mock_nice_blockchain.create_server()
self.addCleanup(srv.close)
self.endpoint.port = port
await open_widget()
def close_dialog():
......
from aiohttp import web, log
import json
import socket
from ucoinpy.documents import Peer
def bma_peering_generator(port):
......@@ -18,6 +19,18 @@ def bma_peering_generator(port):
}
def peer_document_generator(port):
return Peer.from_signed_raw("""Version: 1
Type: Peer
Currency: meta_brouzouf
PublicKey: HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk
Block: 30152-00003E7F9234E7542FCF669B69B0F84FF79CCCD3
Endpoints:
BASIC_MERKLED_API 127.0.0.1 {port}
cXuqZuDfyHvxYAEUkPH1TQ1M+8YNDpj8kiHGYi3LIaMqEdVqwVc4yQYGivjxFMYyngRfxXkyvqBKZA6rKOulCA==
""".format(port=port))
class Request():
def __init__(self, method, url, content):
self.url = url
......@@ -35,6 +48,8 @@ class MockServer():
keep_alive_on=False,
access_log=log.access_logger)
self.port = self.find_unused_port()
def get_request(self, i):
return self.requests[i]
......@@ -56,12 +71,14 @@ class MockServer():
s.close()
return port
def peer(self):
return peer_document_generator(self.port)
async def create_server(self, ssl_ctx=None):
port = self.find_unused_port()
srv = await self.lp.create_server(self.handler, '127.0.0.1', port)
srv = await self.lp.create_server(self.handler, '127.0.0.1', self.port)
protocol = "https" if ssl_ctx else "http"
url = "{}://127.0.0.1:{}".format(protocol, port)
url = "{}://127.0.0.1:{}".format(protocol, self.port)
self.add_route('GET', '/network/peering', bma_peering_generator(port))
self.add_route('GET', '/network/peering', bma_peering_generator(self.port))
return srv, port, url
\ No newline at end of file
return srv, self.port, url
\ No newline at end of file
import sys
import unittest
import asyncio
import quamash
import logging
import time
from PyQt5.QtCore import QLocale
from sakia.core.registry.identities import Identity, IdentitiesRegistry, LocalState, BlockchainState
......@@ -11,10 +7,8 @@ from sakia.tests.mocks.bma import nice_blockchain, corrupted
from sakia.tests import QuamashTest
from sakia.core import Application, Community
from sakia.core.net import Network, Node
from ucoinpy.documents.peer import BMAEndpoint
from ucoinpy.documents.peer import Peer
from sakia.core.net.api.bma.access import BmaAccess
from sakia.tools.exceptions import MembershipNotFoundError
from ucoinpy.api.bma import API
class TestBmaAccess(unittest.TestCase, QuamashTest):
......@@ -26,8 +20,16 @@ class TestBmaAccess(unittest.TestCase, QuamashTest):
self.application = Application(self.qapplication, self.lp, self.identities_registry)
self.application.preferences['notifications'] = False
self.endpoint = BMAEndpoint("", "127.0.0.1", "", 50004)
self.node = Node("test_currency", [self.endpoint],
self.peer = Peer.from_signed_raw("""Version: 1
Type: Peer
Currency: meta_brouzouf
PublicKey: 8Fi1VSTbjkXguwThF4v2ZxC5whK7pwG2vcGTkPUPjPGU
Block: 48698-000005E0F228038E4DDD4F6CA4ACB01EC88FBAF8
Endpoints:
BASIC_MERKLED_API ucoin.inso.ovh 80
82o1sNCh1bLpUXU6nacbK48HBcA9Eu2sPkL1/3c2GtDPxBUZd2U2sb7DxwJ54n6ce9G0Oy7nd1hCxN3fS0oADw==
""")
self.node = Node(self.peer,
"", "HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk",
None, Node.ONLINE,
time.time(), {}, "ucoin", "0.12.0", 0)
......
import sys
import unittest
import logging
from distutils.version import StrictVersion
from PyQt5.QtCore import QLocale
from sakia.core.net.api.bma.access import BmaAccess
from sakia.core.net.network import Network
......@@ -22,6 +22,6 @@ class TestCommunity(unittest.TestCase, QuamashTest):
community = Community("test_currency", network, bma_access)
json_data = community.jsonify()
community_from_json = Community.load(json_data)
community_from_json = Community.load(json_data, StrictVersion('0.12.0'))
self.assertEqual(community.name, community_from_json.name)
self.assertEqual(len(community.network._nodes), len(community_from_json.network._nodes))
......@@ -27,8 +27,8 @@ class TestIdentity(unittest.TestCase, QuamashTest):
self.application = Application(self.qapplication, self.lp, self.identities_registry)
self.application.preferences['notifications'] = False
self.endpoint = BMAEndpoint("", "127.0.0.1", "", 50009)
self.node = Node("test_currency", [self.endpoint],
self.mock_nice_blockchain = nice_blockchain.get_mock(self.lp)
self.node = Node(self.mock_nice_blockchain.peer(),
"", "HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk",
None, Node.ONLINE,
time.time(), {}, "ucoin", "0.12.0", 0)
......@@ -40,14 +40,13 @@ class TestIdentity(unittest.TestCase, QuamashTest):
self.tearDownQuamash()
def test_identity_certifiers_of(self):
mock = nice_blockchain.get_mock(self.lp)
time.sleep(1)
identity = Identity("john", "7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ", 1441130831,
LocalState.COMPLETED, BlockchainState.VALIDATED)
async def exec_test():
srv, port, url = await mock.create_server()
self.endpoint.port = port
srv, port, url = await self.mock_nice_blockchain.create_server()
self.addCleanup(srv.close)
certifiers = await identity.certifiers_of(self.identities_registry, self.community)
self.assertEqual(len(certifiers), 1)
......@@ -58,14 +57,13 @@ class TestIdentity(unittest.TestCase, QuamashTest):
self.lp.run_until_complete(exec_test())
def test_identity_membership(self):
mock = nice_blockchain.get_mock(self.lp)
time.sleep(2)
time.sleep(1)
identity = Identity("john", "7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ", 1441130831,
LocalState.COMPLETED, BlockchainState.VALIDATED)
async def exec_test():
srv, port, url = await mock.create_server()
self.endpoint.port = port
srv, port, url = await self.mock_nice_blockchain.create_server()
self.addCleanup(srv.close)
ms = await identity.membership(self.community)
self.assertEqual(ms["blockNumber"], 0)
self.assertEqual(ms["blockHash"], "DA39A3EE5E6B4B0D3255BFEF95601890AFD80709")
......@@ -76,15 +74,23 @@ class TestIdentity(unittest.TestCase, QuamashTest):
def test_identity_corrupted_membership(self):
mock = corrupted.get_mock(self.lp)
time.sleep(2)
time.sleep(1)
node = Node(mock.peer(),
"", "HnFcSms8jzwngtVomTTnzudZx7SHUQY8sVE1y8yBmULk",
None, Node.ONLINE,
time.time(), {}, "ucoin", "0.12.0", 0)
network = Network.create(node)
bma_access = BmaAccess.create(network)
community = Community("test_currency", network, bma_access)
identity = Identity("john", "7Aqw6Efa9EzE7gtsc8SveLLrM7gm6NEGoywSv4FJx6pZ", 1441130831,
LocalState.COMPLETED, BlockchainState.VALIDATED)
async def exec_test():
srv, port, url = await mock.create_server()
self.endpoint.port = port
self.addCleanup(srv.close)
with self.assertRaises(MembershipNotFoundError):
await identity.membership(self.community)
await identity.membership(community)
self.lp.run_until_complete(exec_test())
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment