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
|
import abc
import copy
import random
from functools import total_ordering
import pyparsing as pp
from . import base
@total_ordering
class _Action(base.Token):
"""
An action that operates on the raw data stream of the message. All
actions have one thing in common: an offset that specifies where the
action should take place.
"""
def __init__(self, offset):
self.offset = offset
def resolve(self, settings, msg):
"""
Resolves offset specifications to a numeric offset. Returns a copy
of the action object.
"""
c = copy.copy(self)
l = msg.length(settings)
if c.offset == "r":
c.offset = random.randrange(l)
elif c.offset == "a":
c.offset = l + 1
return c
def __lt__(self, other):
return self.offset < other.offset
def __eq__(self, other):
return self.offset == other.offset
def __repr__(self):
return self.spec()
@abc.abstractmethod
def spec(self): # pragma: no cover
pass
@abc.abstractmethod
def intermediate(self, settings): # pragma: no cover
pass
class PauseAt(_Action):
unique_name = None
def __init__(self, offset, seconds):
_Action.__init__(self, offset)
self.seconds = seconds
@classmethod
def expr(cls):
e = pp.Literal("p").suppress()
e += base.TokOffset
e += pp.Literal(",").suppress()
e += pp.MatchFirst(
[
base.v_integer,
pp.Literal("f")
]
)
return e.setParseAction(lambda x: cls(*x))
def spec(self):
return "p%s,%s" % (self.offset, self.seconds)
def intermediate(self, settings):
return (self.offset, "pause", self.seconds)
def freeze(self, settings_):
return self
class DisconnectAt(_Action):
def __init__(self, offset):
_Action.__init__(self, offset)
@classmethod
def expr(cls):
e = pp.Literal("d").suppress()
e += base.TokOffset
return e.setParseAction(lambda x: cls(*x))
def spec(self):
return "d%s" % self.offset
def intermediate(self, settings):
return (self.offset, "disconnect")
def freeze(self, settings_):
return self
class InjectAt(_Action):
unique_name = None
def __init__(self, offset, value):
_Action.__init__(self, offset)
self.value = value
@classmethod
def expr(cls):
e = pp.Literal("i").suppress()
e += base.TokOffset
e += pp.Literal(",").suppress()
e += base.TokValue
return e.setParseAction(lambda x: cls(*x))
def spec(self):
return "i%s,%s" % (self.offset, self.value.spec())
def intermediate(self, settings):
return (
self.offset,
"inject",
self.value.get_generator(settings)
)
def freeze(self, settings):
return InjectAt(self.offset, self.value.freeze(settings))
|