youandme/tests/test_client.py

import unittest
import socket
from threading import Thread
import sys
import time
import stem
import stem.process
from stem.control import Controller
from youandme import client

class Connection:
    connected = True

def get_open_port():
    # taken from (but modified) https://stackoverflow.com/a/2838309 by https://stackoverflow.com/users/133374/albert ccy-by-sa-3 https://creativecommons.org/licenses/by-sa/3.0/
    # changes from source: import moved to top of file, bind specifically to localhost
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("127.0.0.1",0))
    s.listen(1)
    port = s.getsockname()[1]
    s.close()
    return port

control_port = str(get_open_port())
socks_port = str(get_open_port())
assert control_port != socks_port

stem.process.launch_tor_with_config(
config = {
    'ControlPort':  control_port,
    'SocksPort': socks_port,
    'Log': [
    'NOTICE stdout'
    ],
}, take_ownership=True)

class Address:
    address = ""

def fake_server():
    with Controller.from_port(port=int(control_port)) as controller:
        controller.authenticate()
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            ip = '127.0.0.1'
            s.bind((ip, 0))
            s.listen(1)
            port = s.getsockname()[1]
            serv = controller.create_ephemeral_hidden_service(
                    {1337: '127.0.0.1:' + str(port)},
                    key_content='ED25519-V3',
                    await_publication=True,
                )
            Address.address = serv.service_id
            conn, addr = s.accept()
            while True:
                conn.send(chr(116).encode('utf-8'))
                data = conn.recv(1)


class TestClient(unittest.TestCase):

    def test_client(self):
        Thread(target=fake_server, daemon=True).start()
        send_data = bytearray()
        recv_data = bytearray()
        while Address.address == "":
            time.sleep(1)
        print(Address.address)
        Thread(target=client.client, args=[0.01, Address.address, int(socks_port), send_data, recv_data, Connection], daemon=True).start()
        start = time.time()
        try:
            while True:
                try:
                    if chr(recv_data.pop(0)) in "t"*100:
                        break
                except IndexError:
                    self.assertLess((time.time() - start), 60)
                    time.sleep(0.01)
        except KeyboardInterrupt:
            raise

unittest.main()