import io
import tempfile
import shutil
import socket
import os
import os.path
import subprocess
import time
import yaml

from unittest import TestCase

import simple_beanstalk


class TestingBeanStalk(object):
    MAX_STARTUP_TIME = 0.5

    def __init__(self, wal_directory):
        if not os.path.exists(wal_directory):
            os.makedirs(wal_directory)
        # get a port to bind to
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind(('localhost', 0))
        host, port = s.getsockname()
        s.close()
        self.host = host
        self.port = port
        cmdline = ['beanstalkd', '-b', wal_directory, '-l', self.host, '-p', str(self.port)]
        self.p = subprocess.Popen(cmdline)
        start_time = time.time()
        self.client = simple_beanstalk.BeanstalkClient(self.host, self.port)
        while (time.time() - start_time) < self.MAX_STARTUP_TIME:
            try:
                self.status()
                return
            except (socket.error, OSError):
                pass
            time.sleep(self.MAX_STARTUP_TIME / 20.0)
        raise ValueError('beanstalk never started!')

    def stop(self):
        self.p.terminate()
        self.p.wait()

    def status(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect((self.host, self.port))
        s.sendall(b'stats\r\n')
        top, rest = s.recv(1024).split(b'\r\n', 1)
        status, count = top.split(b' ')
        count = int(count) + 2
        bio = io.BytesIO()
        bytes_read = len(rest)
        bio.write(rest)
        while bytes_read < count:
            message = s.recv(count - bytes_read)
            if not message:
                break
            bio.write(message)
        bio.seek(0, 0)
        return yaml.safe_load(bio)


class IntegrationBaseTestCase(TestCase):
    def setUp(self):
        self.wd = tempfile.mkdtemp()
        self.bs1 = TestingBeanStalk(os.path.join(self.wd, 'bs1'))
        self.bs2 = TestingBeanStalk(os.path.join(self.wd, 'bs2'))

    def tearDown(self):
        shutil.rmtree(self.wd)
        self.bs1.stop()
        self.bs2.stop()

@@ -172,16 +172,217 @@
 stalk%0A%0A%0A
+class BeanstalkProcessError(Exception):%0A    def __init__(self, status):%0A        self.status = status%0A%0A    def __repr__(self):%0A        return '%7B0%7D(%7B1!r%7D)'.format(self.__class__.__name__, self.status)%0A%0A%0A
 class Te
@@ -432,11 +432,11 @@
 E = 
-0.5
+1.0
 %0A%0A  
@@ -464,32 +464,107 @@
 wal_directory):%0A
+        self.wal_directory = wal_directory%0A%0A        for i in range(3):%0A    
         if not o
@@ -605,16 +605,20 @@
         
+    
 os.maked
@@ -636,16 +636,310 @@
 ectory)%0A
+            self.p = None%0A%0A            try:%0A                self.try_startup()%0A                return%0A            except BeanstalkProcessError:%0A                if self.p is not None:%0A                    self.p.kill()%0A                    shutil.rmtree(wal_directory)%0A%0A    def try_startup(self):%0A
         
@@ -1265,16 +1265,21 @@
 , '-b', 
+self.
 wal_dire
@@ -1324,16 +1324,117 @@
 .port)%5D%0A
+        # give beanstalkd enough time to exit if the port is already in use%0A        time.sleep(0.02)%0A
         
@@ -1644,16 +1644,163 @@
 P_TIME:%0A
+            if self.p.poll() is not None:%0A                exit_status = self.p.returncode%0A                raise BeanstalkProcessError(exit_status)%0A
         
@@ -2148,68 +2148,25 @@
 ket.
-socket(socket.AF_INET, socket.SOCK_STREAM)%0A        s.
+create_
 connect
+ion
 ((se
@@ -2184,16 +2184,30 @@
 lf.port)
+, timeout=0.25
 )%0A      

add retry logic for when the stupid "find unused port" thing does not work and beanstalkd cannot start