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
|
# Copyright 2009-2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Test harness for TAC (Twisted Application Configuration) files."""
__metaclass__ = type
__all__ = [
'TacTestSetup',
'TacException',
]
import errno
import os
import socket
import subprocess
import sys
import time
import warnings
from fixtures import Fixture
from txfixtures.tachandler import (
TacException,
TacTestFixture,
)
from lp.services.daemons import readyservice
from lp.services.osutils import (
get_pid_from_file,
kill_by_pidfile,
remove_if_exists,
two_stage_kill,
until_no_eintr,
)
twistd_script = os.path.abspath(os.path.join(
os.path.dirname(__file__),
os.pardir, os.pardir, os.pardir, os.pardir, 'bin', 'twistd'))
class TacTestSetup(TacTestFixture):
"""Setup an TAC file as daemon for use by functional tests.
You must override setUpRoot to set up a root directory for the daemon.
"""
def setUp(self, spew=False, umask=None):
# setUp() watches the logfile to determine when the daemon has fully
# started. If it sees an old logfile, then it will find the
# readyservice.LOG_MAGIC string and return immediately, provoking
# hard-to-diagnose race conditions. Delete the logfile to make sure
# this does not happen.
remove_if_exists(self.logfile)
TacTestFixture.setUp(self,
python_path=sys.executable,
twistd_script=twistd_script)
def _hasDaemonStarted(self):
"""Has the daemon started?
Startup is recognized by the appearance of readyservice.LOG_MAGIC in
the log file.
"""
if os.path.exists(self.logfile):
with open(self.logfile, 'r') as logfile:
return readyservice.LOG_MAGIC in logfile.read()
else:
return False
def truncateLog(self):
"""Truncate the log file.
Leaves everything up to and including the `LOG_MAGIC` marker in
place. If the `LOG_MAGIC` marker is not found the log is truncated to
0 bytes.
"""
if os.path.exists(self.logfile):
with open(self.logfile, "r+b") as logfile:
position = 0
for line in logfile:
position += len(line)
if readyservice.LOG_MAGIC in line:
logfile.truncate(position)
break
else:
logfile.truncate(0)
def setUpRoot(self):
"""Override this.
This should be able to cope with the root already existing, because it
will be left behind after each test in case it's needed to diagnose a
test failure (e.g. log files might contain helpful tracebacks).
"""
raise NotImplementedError
|