1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
import os
import socket
import subprocess
import sys
import time
class StripeMock(object):
PATH_SPEC = (
os.path.dirname(os.path.realpath(__file__)) + "/openapi/spec3.json"
)
PATH_FIXTURES = (
os.path.dirname(os.path.realpath(__file__)) + "/openapi/fixtures3.json"
)
_port = -1
_process = None
@classmethod
def start(cls):
if not os.path.isfile(cls.PATH_SPEC):
return False
if cls._process is not None:
print("stripe-mock already running on port %s" % cls._port)
return True
cls._port = cls.find_available_port()
print("Starting stripe-mock on port %s..." % cls._port)
cls._process = subprocess.Popen(
[
"stripe-mock",
"-http-port",
str(cls._port),
"-spec",
cls.PATH_SPEC,
"-fixtures",
cls.PATH_FIXTURES,
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
time.sleep(1)
if cls._process.poll() is None:
print("Started stripe-mock, PID = %d" % cls._process.pid)
else:
print("stripe-mock terminated early: %d" % cls._process.returncode)
sys.exit(1)
return True
@classmethod
def stop(cls):
if cls._process is None:
return
print("Stopping stripe-mock...")
cls._process.terminate()
cls._process.wait()
cls._process = None
cls._port = -1
print("Stopped stripe-mock")
@classmethod
def port(cls):
return cls._port
@staticmethod
def find_available_port():
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("localhost", 0))
s.listen(1)
port = s.getsockname()[1]
s.close()
return port
|
