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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
|
import logging
import os
import signal
import subprocess
import sys
import time
import unittest
from urllib.parse import urlunsplit
import psycopg2
import requests
from django.conf import settings
DJANGO_SETTINGS_MODULE = "integration_test.settings"
os.environ.setdefault("DJANGO_SETTINGS_MODULE", DJANGO_SETTINGS_MODULE)
logger = logging.getLogger(__name__)
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
class Process:
@staticmethod
def _command(args):
return list(args)
@classmethod
def run(cls, *args):
subprocess.check_call(cls._command(args))
def __init__(self, *args):
self.args = list(args)
def start(self):
self.process = subprocess.Popen(self._command(self.args), preexec_fn=os.setsid)
logger.info(f"START PROCESS args:{self.args} pid:{self.process.pid}")
time.sleep(1)
def stop(self):
# to be sure we kill all the children:
os.killpg(self.process.pid, signal.SIGTERM)
def __enter__(self):
self.start()
return self
def __exit__(self, *args):
self.stop()
class DjangoCommand(Process):
@staticmethod
def _command(args):
return ["./manage.py"] + list(args) + ["--settings", DJANGO_SETTINGS_MODULE]
def terminate_all_postgres_connections(profile="default"):
db_settings = settings.DATABASES[profile]
conn_params = {
'database': 'template1',
'user': db_settings["USER"],
'password': db_settings["PASSWORD"],
'host': db_settings["HOST"],
'port': db_settings["PORT"],
}
with psycopg2.connect(**conn_params) as conn:
cursor = conn.cursor()
cursor.execute(
"""
SELECT pg_terminate_backend(pg_stat_activity.pid)
FROM pg_stat_activity
WHERE pg_stat_activity.datname = %s
""",
(db_settings["NAME"],),
)
class IntegrationTest(unittest.TestCase):
ADDRPORT = "127.0.0.1:8000"
HOME_URL = urlunsplit(("http", ADDRPORT, "/", "", ""))
def setUp(self):
DjangoCommand.run("flush", "--noinput")
# self.site = DjangoCommand("runserver", self.ADDRPORT)
self.site = Process(
"gunicorn",
"-b",
self.ADDRPORT,
"--timeout",
"600", # useful for worker debugging
"integration_test.wsgi:application",
)
self.site.start()
def tearDown(self):
self.site.stop()
def assertFailure(self):
r = requests.get(self.HOME_URL)
self.assertEqual(r.status_code, 500)
def assertEntries(self, expected):
r = requests.get(self.HOME_URL)
self.assertEqual(r.status_code, 200)
self.assertEqual(r.text, "Entries: {}".format(",".join(expected)))
def enqueue(self, name):
r = requests.post(self.HOME_URL, {"name": name})
self.assertEqual(r.status_code, 200)
self.assertEqual(r.text, "Enqueued")
def test_db_is_empty(self):
self.assertEntries([])
def test_burst(self):
self.enqueue("first")
DjangoCommand.run("rqworker", "--burst")
self.assertEntries(["first"])
def test_site_fails_and_the_reconnects(self):
self.enqueue("first")
DjangoCommand.run("rqworker", "--burst")
terminate_all_postgres_connections()
# the DB connection is gone, so the worker must first detect the problem:
self.assertFailure()
# now the gunicorn worker is ok again:
self.assertEntries(["first"])
def test_worker_lost_connection(self):
with DjangoCommand("rqworker"):
self.enqueue("first")
time.sleep(2) # wait for the worker to do the job
self.assertEntries(["first"]) # job is done
terminate_all_postgres_connections()
self.enqueue("second")
time.sleep(2) # wait for the worker to do the job
self.assertFailure() # let the gunicorn worker reconnect
self.assertEntries(["first", "second"]) # work is done
if __name__ == '__main__':
unittest.main()
|
