import socket
import sys
import threading

ITERATIONS=100

services = [
    ('echo', 7),
    ('ftp', 21),
    ('ssh', 22),
    ('telnet', 23),
    ('smtp', 25)]

protos = [
    ('icmp', 1),
    ('ipv4', 4),
    ('igp', 9),
    ('udp', 17),
    ('ipv6', 41)]

def getservbyname(barrier, service, port):
    barrier.wait()
    assert socket.getservbyname(service) == port

def getservbyport(barrier, service, port):
    barrier.wait()
    assert socket.getservbyport(port) == service

def getprotobyname(barrier, proto, port):
    barrier.wait()
    assert socket.getprotobyname(proto) == port

def run_threads(threads):
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def test_getservbyname():
    barrier = threading.Barrier(len(services))
    threads = [threading.Thread(target=getservbyname, args=(barrier, service, port))
               for service, port in services]
    run_threads(threads)

def test_getservbyport():
    barrier = threading.Barrier(len(services))
    threads = [threading.Thread(target=getservbyport, args=(barrier, service, port))
               for service, port in services]
    run_threads(threads)

def test_getprotobyname():
    barrier = threading.Barrier(len(protos))
    threads = [threading.Thread(target=getprotobyname, args=(barrier, proto, port))
               for proto, port in protos]
    run_threads(threads)

bailing = threading.Event()
old_except_hook = None

def bail(args):
    bailing.set()
    old_except_hook(args)

if __name__ == '__main__':
    old_except_hook = threading.excepthook
    threading.excepthook = bail
    for _ in range(ITERATIONS):
        if bailing.is_set():
            break
        test_getservbyname()
        test_getservbyport()
        test_getprotobyname()

import pwd
import threading
import unittest

N=5
ITERATIONS=1000

pwds = set(pwd.getpwall())

def getpwall(barrier):
    barrier.wait()
    assert pwds == set(pwd.getpwall())

def run_threads(threads):
    for t in threads:
        t.start()
    for t in threads:
        t.join()

if __name__ == '__main__':
    for _ in range(ITERATIONS):
        barrier = threading.Barrier(N)
        threads = [threading.Thread(target=getpwall, args=(barrier,)) for _ in range(N)]
        run_threads(threads)

import dbm
import random
import tempfile
import threading
from pathlib import Path

N=5
ITERATIONS=1000

def run_threads(threads):
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def traverse(barrier, db):
    barrier.wait()
    for key in db.keys():
        try:
            assert db[key] == key
        except KeyError:
            pass

def inserter(barrier, db):
    barrier.wait()
    for _ in range(N):
        ch = chr(ord('a') + random.randrange(26))
        db[ch] = ch

def deleter(barrier, db):
    barrier.wait()
    for _ in range(N):
        key = random.choice(db.keys())
        try:
            del db[key]
        except KeyError:
            pass

def defaulter(barrier, db):
    barrier.wait()
    for _ in range(N):
        ch = chr(ord('a') + random.randrange(26))
        result = db.setdefault(ch, ch)
        # TODO: Is this a gdbm bug?
        if isinstance(result, bytes):
            result = result.decode()
        assert result == ch, f"{result} != {ch}"

def checker(barrier, db):
    barrier.wait()
    for _ in range(N * 10):
        assert len(db) >= 0
        key = random.choice(db.keys())
        key in db

def init_db(path):
    with dbm.open(path, 'c') as db:
        for s in ['foo', 'bar', 'baz']:
            db[s] = s
    return path

def test_db(path):
    for _ in range(ITERATIONS):
        barrier = threading.Barrier(N + 4)
        with dbm.open(path, 'w') as db:
            args = (barrier, db)
            threads = [threading.Thread(target=traverse, args=args) for _ in range(N)]
            threads.append(threading.Thread(target=inserter, args=args))
            threads.append(threading.Thread(target=deleter, args=args))
            threads.append(threading.Thread(target=defaulter, args=args))
            threads.append(threading.Thread(target=checker, args=args))
            run_threads(threads)

if __name__ == '__main__':
    # dbm.sqlite3 only allows accessing DBs from the same thread that opened
    # them (and enforces this)
    for backend in ['dbm.gnu', 'dbm.ndbm']:
        dbm.whichdb = lambda x: backend
        with tempfile.TemporaryDirectory() as dir:
            path = Path(dir) / "test-db"
            init_db(path)
            test_db(path)