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
|
import unittest
class WebhookTestCaseBase(unittest.TestCase):
thread = None
def setUp(self):
import sys
# Use our custom importer to replace certain modules with stub modules to
# enable testing of other parts of GAD
sys.meta_path.append(StubImporter())
def start_gad(self, test_config):
import sys
import time
import os
# Add repo root to sys path
repo_root = os.path.realpath(os.path.join(os.path.dirname(os.path.realpath(__file__)), '..'))
sys.path.insert(1, repo_root)
from gitautodeploy.cli.config import get_config_defaults, init_config
config = get_config_defaults()
config.update(test_config)
init_config(config)
# Create a GAD instance
self.thread = GADRunnerThread(config)
# Run GAD in a thread
self.thread.start()
def gad_port(self):
return self.thread.port
def await_gad(self):
"""Waits for GAD and all it's threads to complete."""
import threading
# Wait for GAD runner thread to finish
self.thread.join()
# Wait for all request handler threads to finish
main_thread = threading.currentThread()
for current_thread in threading.enumerate():
if current_thread == main_thread:
continue
# print "Waiting for thread %s to finish" % current_thread
current_thread.join()
def tearDown(self):
pass
class StubImporter(object):
overload_modules = ['gitautodeploy.wrappers.git', 'gitautodeploy.wrappers.process']
def find_module(self, full_name, package_path):
# Intervene when any wrapper module is imported
if full_name in self.overload_modules:
# Return a loader
return self
return None
def load_module(self, full_name):
"""Load matching module from stubs package (test stub) instead of main gitautodeploy package."""
import imp
import sys
import os
repo_root = os.path.realpath(os.path.join(os.path.dirname(os.path.realpath(__file__)), '..'))
module_name = full_name.split('.')[-1]
module_info = imp.find_module(module_name, [repo_root+'/test/stubs'])
module = imp.load_module(module_name, *module_info)
sys.modules[module_name] = module
# print "returning module %s" % module_name
# module.history.append('WAH')
# Return a module
return module
import threading
class GADRunnerThread(threading.Thread):
def __init__(self, config):
super(GADRunnerThread, self).__init__()
import sys
import time
from gitautodeploy import GitAutoDeploy
self._app = GitAutoDeploy()
self._app.setup(config)
# Setup HTTP server, but do not start serving
self._app.serve_http(serve_forever=False)
# Store PID and port in thread instance
self.pid = self._app._pid
self.port = self._app._http_port
def run(self):
self._app._http_server.handle_request()
def exit(self):
self._app.stop()
self._app.close()
|
