Browse Source

More multi process based tests

PYADPT-55
Edouard Tisserant 9 years ago
parent
commit
9f859ce26f
  1. 2
      src/pEpmodule.cc
  2. 47
      test/mp_sync_test.py
  3. 20
      test/multipEp.py

2
src/pEpmodule.cc

@ -60,7 +60,7 @@ BOOST_PYTHON_MODULE(pEp)
std::locale::global(gen("")); std::locale::global(gen(""));
scope().attr("about") = about(); scope().attr("about") = about();
auto identity_class = class_<Identity>("Identity", auto identity_class = class_<Identity>("Identity",
"Identity(address, username, user_id='', fpr='', comm_type=0, lang='en')\n" "Identity(address, username, user_id='', fpr='', comm_type=0, lang='en')\n"
"\n" "\n"

47
test/mp_sync_test.py

@ -14,7 +14,7 @@ from multipEp import *
#("instance name", [instance_action_func, [args], {kwargs}], result_func), #("instance name", [instance_action_func, [args], {kwargs}], result_func),
#(manager_action_func, [args], {kwargs}, result_func), #(manager_action_func, [args], {kwargs}, result_func),
def scenario0(): def group_on_keygen():
for action in [ for action in [
("GroupA1", [create_account, ["some.one@some.where", "Some One"]]), ("GroupA1", [create_account, ["some.one@some.where", "Some One"]]),
("SoloA", [create_account, ["some.other@else.where", "Some Other"]]), ("SoloA", [create_account, ["some.other@else.where", "Some Other"]]),
@ -31,18 +31,55 @@ def scenario0():
"some.one@some.where", "some.one@some.where",
"read this", "this is a secret message"]]) "read this", "this is a secret message"]])
for action in [ for action in [
("GroupA1", [decrypt_message, [enc_msg]], expect(6)), ("GroupA1", [decrypt_message, [enc_msg]], expect(PEP_rating_reliable)),
(flush_all_mails,), (flush_all_mails,),
("GroupA2", [create_account, ["some.one@some.where", "Some One"]]), ("GroupA2", [create_account, ["some.one@some.where", "Some One"]]),
(cycle_until_no_change, ["GroupA1", "GroupA2"], expect(4)), (cycle_until_no_change, ["GroupA1", "GroupA2"], expect(4)),
("GroupA2", [decrypt_message, [enc_msg]], expect(6)), ("GroupA2", [decrypt_message, [enc_msg]], expect(PEP_rating_reliable))
] : yield action
return enc_msg
def group_on_cannotdecypt():
for action in [
("GroupA1", [create_account, ["some.one@some.where", "Some One"]]),
("SoloA", [create_account, ["some.other@else.where", "Some Other"]]),
# key exchange
("SoloA", [send_message, ["some.other@else.where",
"some.one@some.where",
"Hey Bro", "Heeeey Brooooo"]]),
("GroupA1", [send_message, ["some.one@some.where",
"some.other@else.where",
"Yo Dude", "Yooooo Duuuude"]])
] : yield action
enc_msg = yield ("SoloA", [encrypted_message, ["some.other@else.where",
"some.one@some.where",
"read this", "this is a secret message"]])
for action in [
("GroupA1", [decrypt_message, [enc_msg]], expect(PEP_rating_reliable)),
(flush_all_mails,),
("GroupA2", [create_account, ["some.one@some.where", "Some One"]]),
(flush_all_mails,),
("GroupA2", [decrypt_message, [enc_msg]], expect(PEP_rating_have_no_key)),
(cycle_until_no_change, ["GroupA1", "GroupA2"], expect(4)),
("GroupA2", [decrypt_message, [enc_msg]], expect(PEP_rating_reliable)),
] : yield action
def group_of_3_members():
enc_msg = yield from group_on_keygen()
for action in [
("GroupA3", [create_account, ["some.one@some.where", "Some One"]]), ("GroupA3", [create_account, ["some.one@some.where", "Some One"]]),
(cycle_until_no_change, ["GroupA1", "GroupA2", "GroupA3"], expect(3)), (cycle_until_no_change, ["GroupA1", "GroupA2", "GroupA3"], expect(3)),
# force consume messages # force consume messages
# ("GroupA3", [None, None, None, -60*15]), # ("GroupA3", [None, None, None, -60*15]),
("GroupA3", [decrypt_message, [enc_msg]], expect(6)) ("GroupA3", [decrypt_message, [enc_msg]], expect(PEP_rating_reliable))
] : yield action ] : yield action
return enc_msg
if __name__ == "__main__": if __name__ == "__main__":
run_scenario(scenario0) run_scenario(group_on_keygen)
run_scenario(group_on_cannotdecypt)
run_scenario(group_of_3_members)

20
test/multipEp.py

@ -8,6 +8,23 @@ import types
from copy import deepcopy from copy import deepcopy
from collections import OrderedDict from collections import OrderedDict
# FIXME : move to main pEp module
# PEP_rating :
PEP_rating_undefined = 0
PEP_rating_cannot_decrypt = 1
PEP_rating_have_no_key = 2
PEP_rating_unencrypted = 3
PEP_rating_unencrypted_for_some = 4
PEP_rating_unreliable = 5
PEP_rating_reliable = 6
PEP_rating_trusted = 7
PEP_rating_trusted_and_anonymized = 8
PEP_rating_fully_anonymous = 9
PEP_rating_mistrust = -1,
PEP_rating_b0rken = -2,
PEP_rating_under_attack = -3
# manager globals # manager globals
instances = None instances = None
@ -203,7 +220,6 @@ def pEp_instance_run(iname, conn, _msgs_folders, _handshakes_seen, _handshakes_v
if order is None: if order is None:
break break
print(order)
res = execute_order(order, handler) res = execute_order(order, handler)
conn.send(res) conn.send(res)
@ -278,7 +294,6 @@ def run_scenario(scenario):
output = action[-1] output = action[-1]
action = action[:-1] action = action[:-1]
print(action)
if type(action[0]) == str: if type(action[0]) == str:
res = run_instance_action(action) res = run_instance_action(action)
else: else:
@ -290,7 +305,6 @@ def run_scenario(scenario):
action = sc.send(res) action = sc.send(res)
except StopIteration: pass except StopIteration: pass
if "wait_for_debug" in sys.argv: if "wait_for_debug" in sys.argv:
input("#"*80 + "\n" + input("#"*80 + "\n" +
"Press ENTER to cleanup\n" + "Press ENTER to cleanup\n" +

Loading…
Cancel
Save