<commit_before>import json
import pickle
import subprocess
import time
import random

import os
import pytest

from src import node
from src.utils.utils import value_and_tally

DIR = 'logs/'
NODE_CMD_PREFIX = ['python2', '-u', '-m', 'src.node']  # -u forces stdin/stdout/stderr to be unbuffered
# NODE_CMD_PREFIX = ['python2', '-u', '-m', 'src.node', '--debug']


def delete_contents_of_dir(dname):
    for f in os.listdir(dname):
        fpath = os.path.join(dname, f)
        if os.path.isfile(fpath):
            os.unlink(fpath)


def search_for_string(fname, target):
    with open(fname, 'r') as f:
        for line in f:
            if target in line:
                print "Test: found target in line", target, line
                return line
    print "Test: did not find", target, "in file", fname
    return None


def search_for_string_in_dir(folder, target, f=lambda x: x):
    res = []
    for fname in os.listdir(folder):
        msg = search_for_string(folder + fname, target)
        if msg is not None:
            msg = msg.split(target)[-1].strip()
            print 'Test: found', f(msg)
            res.append(f(msg))
    return res


def run_subprocesses(prefix, cmds):
    ps = []
    for cmd in cmds:
        print "Test: running subprocess", prefix + cmd
        p = subprocess.Popen(prefix + cmd)
        ps.append(p)
    return ps


@pytest.fixture
def discover():
    p = subprocess.Popen(['python2', '-m', 'src.discovery'])
    time.sleep(1)  # wait for it to spin up
    yield None
    print "Test: tear down discovery"
    p.terminate()


@pytest.fixture
def folder():
    if not os.path.exists(DIR):
        os.makedirs(DIR)

    delete_contents_of_dir(DIR)


def check_acs_files(n, t):
    target = 'ACS: DONE'
    print os.listdir(DIR)
    acs_dones = search_for_string_in_dir(DIR, target, json.loads)

    # do various checks
    assert len(acs_dones) >= n - t, "ACS incorrect length! len = {}, n = {}, t = {}".format(len(acs_dones), n, t)

    # first find the agreed set and check there's majority
    # NOTE: dictionary are unhashable, so we cheat and use pickle to convert it to a string, and then reload it
    s, tally_s = value_and_tally([pickle.dumps(x['set']) for x in acs_dones])
    assert tally_s >= n - t
    s = pickle.loads(s)

    # filter the messages that is not in the agreed set
    key_of_ones = [k for k, v in s.iteritems() if v == 1]
    print key_of_ones
    assert len(key_of_ones) >= n - t

    # NOTE: we use the same trick with pickle
    _msgs, tally_msgs = value_and_tally([pickle.dumps(x['msgs']) for x in acs_dones])
    msgs = {k: pickle.loads(_msgs)[k] for k in key_of_ones}

    # check that we have enough messages
    assert len(msgs) >= n - t


def check_bracha_files(n, t):
    target = 'Bracha: DELIVER'
    bracha_delivers = search_for_string_in_dir(DIR, target)

    assert len(bracha_delivers) >= n - t, "Bracha incorrect length! len = {}, n = {}, t = {}".format(len(bracha_delivers), n, t)

    m, tally = value_and_tally(bracha_delivers)
    assert tally >= n - t, "Bracha incorrect tally! tally = {}, n = {}, t = {}, m = {}".format(tally, n, t, m)


def check_mo14_files(n, t, expected_v):
    # TODO check expected value v
    target = 'Mo14: DECIDED'
    mo14_decides = search_for_string_in_dir(DIR, target)

    assert len(mo14_decides) >= n - t, "Mo14 incorrect length! len = {}, n = {}, t = {}".format(len(mo14_decides), n, t)

    vs = [int(x) for x in mo14_decides]
    v, tally = value_and_tally(vs)

    assert tally >= n - t, "Mo14 incorrect tally! tally = {}, n = {}, t = {}, v = {}".format(tally, n, t, v)
    assert int(v) == expected_v


@pytest.mark.parametrize("n,t,f", [
    (4, 1, 'omission'),
    (7, 2, 'omission'),
    (19, 6, 'omission'),
    (4, 1, 'byzantine'),
    (7, 2, 'byzantine'),
    (19, 6, 'byzantine'),
])
def test_acs(n, t, f, discover, folder):
    configs = []
    for i in range(n - t):
        port = 12345 + i
        configs.append(node.Config(port, n, t, test='acs', output=DIR + str(port) + '.out'))
    for i in range(t):
        port = 11111 + i
        configs.append(node.Config(port, n, t, test='acs', failure=f, output=DIR + str(port) + '.out'))

    ps = run_subprocesses(NODE_CMD_PREFIX, [cfg.make_args() for cfg in configs])

    time.sleep(30)

    for p in ps:
        p.terminate()

    # TODO not sure where to flush, so use sleep for now...
    assert 0 == subprocess.call('sync', shell=True)
    time.sleep(1)
    print "Test: ACS nodes terminated"
    check_acs_files(n, t)
    print "Test: ACS test passed"


@pytest.mark.parametrize("n,t,f", [
    (4, 1, 'omission'),
    (7, 2, 'omission'),
    (19, 6, 'omission'),
])
def test_bracha(n, t, f, discover, folder):
    configs = [node.Config(12345, n, t, test='bracha', output=DIR + '12345.out')]
    for i in range(n - t - 1):
        port = 12345 + 1 + i
        configs.append(node.Config(port, n, t, output=DIR + str(port) + '.out'))
    for i in range(t):
        port = 11111 + i
        configs.append(node.Config(port, n, t, failure=f, output=DIR + str(port) + '.out'))

    ps = run_subprocesses(NODE_CMD_PREFIX, [cfg.make_args() for cfg in configs])

    time.sleep(20)
    for p in ps:
        p.terminate()

    # TODO not sure where to flush, so use sleep for now...
    assert 0 == subprocess.call('sync', shell=True)
    time.sleep(1)
    print "Test: Bracha nodes terminated"
    check_bracha_files(n, t)
    print "Test: Bracha test passed"


@pytest.mark.parametrize("n,t,f", [
    (4, 1, 'byzantine'),
    (7, 2, 'byzantine'),
    (19, 6, 'byzantine'),
    (4, 1, 'omission'),
    (7, 2, 'omission'),
    (19, 6, 'omission'),
])
def test_mo14(n, t, f, discover, folder):
    v = random.randint(0, 1)
    configs = []
    for i in range(n - t):
        port = 12345 + i
        configs.append(node.Config(port, n, t, test='mo14', value=v, output=DIR + str(port) + '.out'))
    for i in range(t):
        port = 11111 + i
        randv = random.randint(0, 1)
        configs.append(node.Config(port, n, t, test='mo14', value=randv, failure=f, output=DIR + str(port) + '.out'))

    ps = run_subprocesses(NODE_CMD_PREFIX, [cfg.make_args() for cfg in configs])

    time.sleep(20)
    for p in ps:
        p.terminate()

    # TODO not sure where to flush, so use sleep for now...
    assert 0 == subprocess.call('sync', shell=True)
    time.sleep(1)
    print "Test: Mo14 nodes terminates"
    check_mo14_files(n, t, v)
    print "Test: Mo14 test passed"

if __name__ == '__main__':
    check_acs_files(4, 1)<commit_msg>Create folder before running discovery<commit_after>import json
import pickle
import subprocess
import time
import random

import os
import pytest

from src import node
from src.utils.utils import value_and_tally

DIR = 'logs/'
NODE_CMD_PREFIX = ['python2', '-u', '-m', 'src.node']  # -u forces stdin/stdout/stderr to be unbuffered
# NODE_CMD_PREFIX = ['python2', '-u', '-m', 'src.node', '--debug']


def delete_contents_of_dir(dname):
    for f in os.listdir(dname):
        fpath = os.path.join(dname, f)
        if os.path.isfile(fpath):
            os.unlink(fpath)


def search_for_string(fname, target):
    with open(fname, 'r') as f:
        for line in f:
            if target in line:
                print "Test: found target in line", target, line
                return line
    print "Test: did not find", target, "in file", fname
    return None


def search_for_string_in_dir(folder, target, f=lambda x: x):
    res = []
    for fname in os.listdir(folder):
        msg = search_for_string(folder + fname, target)
        if msg is not None:
            msg = msg.split(target)[-1].strip()
            print 'Test: found', f(msg)
            res.append(f(msg))
    return res


def run_subprocesses(prefix, cmds):
    ps = []
    for cmd in cmds:
        print "Test: running subprocess", prefix + cmd
        p = subprocess.Popen(prefix + cmd)
        ps.append(p)
    return ps


@pytest.fixture
def discover():
    p = subprocess.Popen(['python2', '-m', 'src.discovery'])
    time.sleep(1)  # wait for it to spin up
    yield None
    print "Test: tear down discovery"
    p.terminate()


@pytest.fixture
def folder():
    if not os.path.exists(DIR):
        os.makedirs(DIR)

    delete_contents_of_dir(DIR)


def check_acs_files(n, t):
    target = 'ACS: DONE'
    print os.listdir(DIR)
    acs_dones = search_for_string_in_dir(DIR, target, json.loads)

    # do various checks
    assert len(acs_dones) >= n - t, "ACS incorrect length! len = {}, n = {}, t = {}".format(len(acs_dones), n, t)

    # first find the agreed set and check there's majority
    # NOTE: dictionary are unhashable, so we cheat and use pickle to convert it to a string, and then reload it
    s, tally_s = value_and_tally([pickle.dumps(x['set']) for x in acs_dones])
    assert tally_s >= n - t
    s = pickle.loads(s)

    # filter the messages that is not in the agreed set
    key_of_ones = [k for k, v in s.iteritems() if v == 1]
    print key_of_ones
    assert len(key_of_ones) >= n - t

    # NOTE: we use the same trick with pickle
    _msgs, tally_msgs = value_and_tally([pickle.dumps(x['msgs']) for x in acs_dones])
    msgs = {k: pickle.loads(_msgs)[k] for k in key_of_ones}

    # check that we have enough messages
    assert len(msgs) >= n - t


def check_bracha_files(n, t):
    target = 'Bracha: DELIVER'
    bracha_delivers = search_for_string_in_dir(DIR, target)

    assert len(bracha_delivers) >= n - t, "Bracha incorrect length! len = {}, n = {}, t = {}".format(len(bracha_delivers), n, t)

    m, tally = value_and_tally(bracha_delivers)
    assert tally >= n - t, "Bracha incorrect tally! tally = {}, n = {}, t = {}, m = {}".format(tally, n, t, m)


def check_mo14_files(n, t, expected_v):
    # TODO check expected value v
    target = 'Mo14: DECIDED'
    mo14_decides = search_for_string_in_dir(DIR, target)

    assert len(mo14_decides) >= n - t, "Mo14 incorrect length! len = {}, n = {}, t = {}".format(len(mo14_decides), n, t)

    vs = [int(x) for x in mo14_decides]
    v, tally = value_and_tally(vs)

    assert tally >= n - t, "Mo14 incorrect tally! tally = {}, n = {}, t = {}, v = {}".format(tally, n, t, v)
    assert int(v) == expected_v


@pytest.mark.parametrize("n,t,f", [
    (4, 1, 'omission'),
    (7, 2, 'omission'),
    (19, 6, 'omission'),
    (4, 1, 'byzantine'),
    (7, 2, 'byzantine'),
    (19, 6, 'byzantine'),
])
def test_acs(n, t, f, folder, discover):
    configs = []
    for i in range(n - t):
        port = 12345 + i
        configs.append(node.Config(port, n, t, test='acs', output=DIR + str(port) + '.out'))
    for i in range(t):
        port = 11111 + i
        configs.append(node.Config(port, n, t, test='acs', failure=f, output=DIR + str(port) + '.out'))

    ps = run_subprocesses(NODE_CMD_PREFIX, [cfg.make_args() for cfg in configs])

    time.sleep(30)

    for p in ps:
        p.terminate()

    # TODO not sure where to flush, so use sleep for now...
    assert 0 == subprocess.call('sync', shell=True)
    time.sleep(5)
    print "Test: ACS nodes terminated"
    check_acs_files(n, t)
    print "Test: ACS test passed"


@pytest.mark.parametrize("n,t,f", [
    (4, 1, 'omission'),
    (7, 2, 'omission'),
    (19, 6, 'omission'),
])
def test_bracha(n, t, f, folder, discover):
    configs = [node.Config(12345, n, t, test='bracha', output=DIR + '12345.out')]
    for i in range(n - t - 1):
        port = 12345 + 1 + i
        configs.append(node.Config(port, n, t, output=DIR + str(port) + '.out'))
    for i in range(t):
        port = 11111 + i
        configs.append(node.Config(port, n, t, failure=f, output=DIR + str(port) + '.out'))

    ps = run_subprocesses(NODE_CMD_PREFIX, [cfg.make_args() for cfg in configs])

    time.sleep(20)
    for p in ps:
        p.terminate()

    # TODO not sure where to flush, so use sleep for now...
    assert 0 == subprocess.call('sync', shell=True)
    time.sleep(1)
    print "Test: Bracha nodes terminated"
    check_bracha_files(n, t)
    print "Test: Bracha test passed"


@pytest.mark.parametrize("n,t,f", [
    (4, 1, 'byzantine'),
    (7, 2, 'byzantine'),
    (19, 6, 'byzantine'),
    (4, 1, 'omission'),
    (7, 2, 'omission'),
    (19, 6, 'omission'),
])
def test_mo14(n, t, f, folder, discover):
    v = random.randint(0, 1)
    configs = []
    for i in range(n - t):
        port = 12345 + i
        configs.append(node.Config(port, n, t, test='mo14', value=v, output=DIR + str(port) + '.out'))
    for i in range(t):
        port = 11111 + i
        randv = random.randint(0, 1)
        configs.append(node.Config(port, n, t, test='mo14', value=randv, failure=f, output=DIR + str(port) + '.out'))

    ps = run_subprocesses(NODE_CMD_PREFIX, [cfg.make_args() for cfg in configs])

    time.sleep(20)
    for p in ps:
        p.terminate()

    # TODO not sure where to flush, so use sleep for now...
    assert 0 == subprocess.call('sync', shell=True)
    time.sleep(1)
    print "Test: Mo14 nodes terminates"
    check_mo14_files(n, t, v)
    print "Test: Mo14 test passed"

if __name__ == '__main__':
    check_acs_files(4, 1)
{'fraction_non_alphanumeric': 0.14828403271420929, 'fraction_numerical': 0.019720247649621648, 'mean_word_length': 2.8516337945245804, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2452102', 'n_tokens_mistral': 5007, 'n_tokens_neox': 4721, 'n_words': 1391}