forked from OpenTSDB/tcollector
-
Notifications
You must be signed in to change notification settings - Fork 0
/
mocks.py
78 lines (62 loc) · 2.3 KB
/
mocks.py
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
# This file is part of tcollector.
# Copyright (C) 2014 The tcollector Authors.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at your
# option) any later version. This program is distributed in the hope that it
# will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
# General Public License for more details. You should have received a copy
# of the GNU Lesser General Public License along with this program. If not,
# see <http://www.gnu.org/licenses/>.
import sys
import traceback
# for debugging
real_stderr = sys.stderr
class SocketDone(Exception):
pass
class Socket():
def __init__(self):
self.AF_INET = 0
self.SOCK_DGRAM = 0
self.error = Exception
self.state = { 'udp_in': [] }
self._socketSingleton = self.SocketSingleton(self.state)
def socket(self, ignored1, ignored2):
return self._socketSingleton
class SocketSingleton():
def __init__(self, state):
self.state = state
def bind(self, host_and_port):
return None
def close(self):
return None
def recvfrom(self, inBytes):
if (len(self.state['udp_in']) > 0):
line = self.state['udp_in'].pop(0)
return (line, None)
else:
raise SocketDone('stop reading from socket')
class Sys():
def __init__(self):
self.stderr = self.Stderr()
self.stdout = self.Stdout()
def exit(self, exitCode):
err = "\n".join(self.stderr.lines)
trace = traceback.format_exc()
msg = 'exit called with code %s\n stderr: %s\n trace: %s'
raise Exception(msg % (exitCode, err, trace))
class Stderr():
def __init__(self):
self.lines = []
def write(self, outString):
self.lines.append(outString)
class Stdout():
def __init__(self):
self.lines = []
def write(self, outString):
self.lines.append(outString)
class Utils():
def __init__(self):
self.drop_privileges = lambda: None