9384631762

Committed 05 Jun 2024 01:03PM UTC coverage: 71.915% (-0.1%) from 72.058%

Build # 9384631762

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #14 from Alviner/lost-changes

lost changes

Run Details

641 of 986 branches covered (65.01%)

Branch coverage included in aggregate %.

16 of 18 new or added lines in 4 files covered. (88.89%)

1 existing line in 1 file now uncovered.

1999 of 2685 relevant lines covered (74.45%)

3.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.43

/dnslib/lex.py

import collections
import string
from io import StringIO


class Lexer(object):
    """
    Simple Lexer base class. Provides basic lexer framework and
    helper functionality (read/peek/pushback etc)

    Each state is implemented using a method (lexXXXX) which should
    match a single token and return a (token,lexYYYY) tuple, with lexYYYY
    representing the next state. If token is None this is not emitted
    and if lexYYYY is None or the lexer reaches the end of the
    input stream the lexer exits.

    The 'parse' method returns a generator that will return tokens
    (the class also acts as an iterator)

    The default start state is 'lexStart'

    Input can either be a string/bytes or file object.

    The approach is based loosely on Rob Pike's Go lexer presentation
    (using generators rather than channels).

    >>> p = Lexer("a bcd efgh")
    >>> p.read()
    'a'
    >>> p.read()
    ' '
    >>> p.peek(3)
    'bcd'
    >>> p.read(5)
    'bcd e'
    >>> p.pushback('e')
    >>> p.read(4)
    'efgh'
    """

    escape_chars = "\\"
    escape = {"n": "\n", "t": "\t", "r": "\r"}

    def __init__(self, f, debug=False):
        if hasattr(f, "read"):
            self.f = f
        elif type(f) == str:
            self.f = StringIO(f)
        elif type(f) == bytes:
            self.f = StringIO(f.decode())
        else:
            raise ValueError("Invalid input")
        self.debug = debug
        self.q = collections.deque()
        self.state = self.lex_start
        self.escaped = False
        self.eof = False

    def __iter__(self):
        return self.parse()

    def next_token(self):
        if self.debug:
            print("STATE", self.state)
        (tok, self.state) = self.state()
        return tok

    def parse(self):
        while self.state is not None and not self.eof:
            tok = self.next_token()
            if tok:
                yield tok

    def read(self, n=1):
        s = ""
        while self.q and n > 0:
            s += self.q.popleft()
            n -= 1
        s += self.f.read(n)
        if s == "":
            self.eof = True
        if self.debug:
            print("Read: >%s<" % repr(s))
        return s

    def peek(self, n=1):
        s = ""
        i = 0
        while len(self.q) > i and n > 0:
            s += self.q[i]
            i += 1
            n -= 1
        r = self.f.read(n)
        if n > 0 and r == "":
            self.eof = True
        self.q.extend(r)
        if self.debug:
            print("Peek : >%s<" % repr(s + r))
        return s + r

    def pushback(self, s):
        p = collections.deque(s)
        p.extend(self.q)
        self.q = p

    def readescaped(self):
        c = self.read(1)
        if c in self.escape_chars:
            self.escaped = True
            n = self.peek(3)
            if n.isdigit():
                n = self.read(3)
                if self.debug:
                    print("Escape: >%s<" % n)
                return chr(int(n, 8))
            elif n[0] in "x":
                x = self.read(3)
                if self.debug:
                    print("Escape: >%s<" % x)
                return chr(int(x[1:], 16))
            else:
                c = self.read(1)
                if self.debug:
                    print("Escape: >%s<" % c)
                return self.escape.get(c, c)
        else:
            self.escaped = False
            return c

    def lex_start(self):
        return (None, None)


class WordLexer(Lexer):
    """
    Example lexer which will split input stream into words (respecting
    quotes)

    To emit SPACE tokens: self.spacetok = ('SPACE',None)
    To emit NL tokens: self.nltok = ('NL',None)

    >>> l = WordLexer(r'abc "def\100\x3d\\. ghi" jkl')
    >>> list(l)
    [('ATOM', 'abc'), ('ATOM', 'def@=. ghi'), ('ATOM', 'jkl')]
    >>> l = WordLexer(r"1 '2 3 4' 5")
    >>> list(l)
    [('ATOM', '1'), ('ATOM', '2 3 4'), ('ATOM', '5')]
    >>> l = WordLexer("abc# a comment")
    >>> list(l)
    [('ATOM', 'abc'), ('COMMENT', 'a comment')]
    """

    wordchars = set(string.ascii_letters) | set(string.digits) | set(string.punctuation)
    quotechars = set("\"'")
    commentchars = set("#")
    spacechars = set(" \t\x0b\x0c")
    nlchars = set("\r\n")
    spacetok = None
    nltok = None

    def lex_start(self):
        return (None, self.lex_space)

    def lex_space(self):
        s = []
        if self.spacetok:
            tok = lambda n: (self.spacetok, n) if s else (None, n)
        else:
            tok = lambda n: (None, n)
        while not self.eof:
            c = self.peek()
            if c in self.spacechars:
                s.append(self.read())
            elif c in self.nlchars:
                return tok(self.lex_nl)
            elif c in self.commentchars:
                return tok(self.lex_comment)
            elif c in self.quotechars:
                return tok(self.lex_quote)
            elif c in self.wordchars:
                return tok(self.lex_word)
            elif c:
                raise ValueError(
                    "Invalid input [%d]: %s"
                    % (
                        self.f.tell(),
                        c,
                    ),
                )
        return (None, None)

    def lex_nl(self):
        while True:
            c = self.read()
            if c not in self.nlchars:
                self.pushback(c)
                return (self.nltok, self.lex_space)

    def lex_comment(self):
        s = []
        tok = lambda n: (("COMMENT", "".join(s)), n) if s else (None, n)
        start = False
        _ = self.read()
        while not self.eof:
            c = self.read()
            if c == "\n":
                self.pushback(c)
                return tok(self.lex_nl)
            elif start or c not in string.whitespace:
                start = True
                s.append(c)
        return tok(None)

    def lex_word(self):
        s = []
        tok = lambda n: (("ATOM", "".join(s)), n) if s else (None, n)
        while not self.eof:
            c = self.peek()
            if c == '"':
                return tok(self.lex_quote)
            elif c in self.commentchars:
                return tok(self.lex_comment)
            elif c.isspace():
                return tok(self.lex_space)
            elif c in self.wordchars:
                s.append(self.read())
            elif c:
                raise ValueError(
                    "Invalid input [%d]: %s"
                    % (
                        self.f.tell(),
                        c,
                    ),
                )
        return tok(None)

    def lex_quote(self):
        s = []
        tok = lambda n: (("ATOM", "".join(s)), n)
        q = self.read(1)
        while not self.eof:
            c = self.readescaped()
            if c == q and not self.escaped:
                break
            else:
                s.append(c)
        return tok(self.lex_space)


class RandomLexer(Lexer):
    """
    Test lexing from infinite stream.

    Extract strings of letters/numbers from /dev/urandom

    >>> import itertools,sys
    >>> if sys.version[0] == '2':
    ...     f = open("/dev/urandom")
    ... else:
    ...     f = open("/dev/urandom",encoding="ascii",errors="replace")
    >>> r = RandomLexer(f)
    >>> i = iter(r)
    >>> len(list(itertools.islice(i,10)))
    10
    """

    minalpha = 4
    mindigits = 3

    def lex_start(self):
        return (None, self.lex_random)

    def lex_random(self):
        n = 0
        c = self.peek(1)
        while not self.eof:
            if c.isalpha():
                return (None, self.lex_alpha)
            elif c.isdigit():
                return (None, self.lex_digits)
            else:
                n += 1
                _ = self.read(1)
                c = self.peek(1)
        return (None, None)

    def lex_digits(self):
        s = []
        c = self.read(1)
        while c.isdigit():
            s.append(c)
            c = self.read(1)
        self.pushback(c)
        if len(s) >= self.mindigits:
            return (("NUMBER", "".join(s)), self.lex_random)
        else:
            return (None, self.lex_random)

    def lex_alpha(self):
        s = []
        c = self.read(1)
        while c.isalpha():
            s.append(c)
            c = self.read(1)
        self.pushback(c)
        if len(s) >= self.minalpha:
            return (("STRING", "".join(s)), self.lex_random)
        else:
            return (None, self.lex_random)


if __name__ == "__main__":

    import argparse
    import doctest
    import sys

    p = argparse.ArgumentParser(description="Lex Tester")
    p.add_argument(
        "--lex",
        "-l",
        action="store_true",
        default=False,
        help="Lex input (stdin)",
    )
    p.add_argument(
        "--nl",
        action="store_true",
        default=False,
        help="Output NL tokens",
    )
    p.add_argument(
        "--space",
        action="store_true",
        default=False,
        help="Output Whitespace tokens",
    )
    p.add_argument("--wordchars", help="Wordchars")
    p.add_argument("--quotechars", help="Quotechars")
    p.add_argument("--commentchars", help="Commentchars")
    p.add_argument("--spacechars", help="Spacechars")
    p.add_argument("--nlchars", help="NLchars")

    args = p.parse_args()

    if args.lex:
        l = WordLexer(sys.stdin)
        if args.wordchars:
            l.wordchars = set(args.wordchars)
        if args.quotechars:
            l.quotechars = set(args.quotechars)
        if args.commentchars:
            l.commentchars = set(args.commentchars)
        if args.spacechars:
            l.spacechars = set(args.spacechars)
        if args.nlchars:
            l.nlchars = set(args.nlchars)
        if args.space:
            l.spacetok = ("SPACE",)
        if args.nl:
            l.nltok = ("NL",)

        for tok in l:
            print(tok)
    else:
        try:
            # Test if we have /dev/urandom
            open("/dev/urandom")
            sys.exit(0 if doctest.testmod().failed == 0 else 1)
        except IOError:
            # Don't run stream test
            doctest.run_docstring_examples(Lexer, globals())
            doctest.run_docstring_examples(WordLexer, globals())

1	import collections	3✔
2	import string	5✔
3	from io import StringIO	5✔
4
5
6	class Lexer(object):	5✔
7	"""
8	Simple Lexer base class. Provides basic lexer framework and
9	helper functionality (read/peek/pushback etc)
10
11	Each state is implemented using a method (lexXXXX) which should
12	match a single token and return a (token,lexYYYY) tuple, with lexYYYY
13	representing the next state. If token is None this is not emitted
14	and if lexYYYY is None or the lexer reaches the end of the
15	input stream the lexer exits.
16
17	The 'parse' method returns a generator that will return tokens
18	(the class also acts as an iterator)
19
20	The default start state is 'lexStart'
21
22	Input can either be a string/bytes or file object.
23
24	The approach is based loosely on Rob Pike's Go lexer presentation
25	(using generators rather than channels).
26
27	>>> p = Lexer("a bcd efgh")
28	>>> p.read()
29	'a'
30	>>> p.read()
31	' '
32	>>> p.peek(3)
33	'bcd'
34	>>> p.read(5)
35	'bcd e'
36	>>> p.pushback('e')
37	>>> p.read(4)
38	'efgh'
39	"""
40
41	escape_chars = "\\"	5✔
42	escape = {"n": "\n", "t": "\t", "r": "\r"}	5✔
43
44	def __init__(self, f, debug=False):	5✔
45	if hasattr(f, "read"):	5✔
46	self.f = f	5✔
47	elif type(f) == str:	5!
48	self.f = StringIO(f)	5✔
49	elif type(f) == bytes:	×
50	self.f = StringIO(f.decode())	×
51	else:
52	raise ValueError("Invalid input")	×
53	self.debug = debug	5✔
54	self.q = collections.deque()	5✔
55	self.state = self.lex_start	5✔
56	self.escaped = False	5✔
57	self.eof = False	5✔
58
59	def __iter__(self):	5✔
60	return self.parse()	5✔
61
62	def next_token(self):	5✔
63	if self.debug:	5!
64	print("STATE", self.state)	×
65	(tok, self.state) = self.state()	5✔
66	return tok	5✔
67
68	def parse(self):	5✔
69	while self.state is not None and not self.eof:	5✔
70	tok = self.next_token()	5✔
71	if tok:	5✔
72	yield tok	5✔
73
74	def read(self, n=1):	5✔
75	s = ""	5✔
76	while self.q and n > 0:	5✔
77	s += self.q.popleft()	5✔
78	n -= 1	5✔
79	s += self.f.read(n)	5✔
80	if s == "":	5✔
81	self.eof = True	5✔
82	if self.debug:	5!
83	print("Read: >%s<" % repr(s))	×
84	return s	5✔
85
86	def peek(self, n=1):	5✔
87	s = ""	5✔
88	i = 0	5✔
89	while len(self.q) > i and n > 0:	5✔
90	s += self.q[i]	5✔
91	i += 1	5✔
92	n -= 1	5✔
93	r = self.f.read(n)	5✔
94	if n > 0 and r == "":	5✔
95	self.eof = True	5✔
96	self.q.extend(r)	5✔
97	if self.debug:	5!
98	print("Peek : >%s<" % repr(s + r))	×
99	return s + r	5✔
100
101	def pushback(self, s):	5✔
102	p = collections.deque(s)	5✔
103	p.extend(self.q)	5✔
104	self.q = p	5✔
105
106	def readescaped(self):	5✔
107	c = self.read(1)	5✔
108	if c in self.escape_chars:	5✔
109	self.escaped = True	5✔
110	n = self.peek(3)	5✔
111	if n.isdigit():	5!
112	n = self.read(3)	×
113	if self.debug:	×
114	print("Escape: >%s<" % n)	×
115	return chr(int(n, 8))	×
116	elif n[0] in "x":	5!
117	x = self.read(3)	×
118	if self.debug:	×
119	print("Escape: >%s<" % x)	×
120	return chr(int(x[1:], 16))	×
121	else:
122	c = self.read(1)	5✔
123	if self.debug:	5!
124	print("Escape: >%s<" % c)	×
125	return self.escape.get(c, c)	5✔
126	else:
127	self.escaped = False	5✔
128	return c	5✔
129
130	def lex_start(self):	5✔
131	return (None, None)	×
132
133
134	class WordLexer(Lexer):	5✔
135	"""
136	Example lexer which will split input stream into words (respecting
137	quotes)
138
139	To emit SPACE tokens: self.spacetok = ('SPACE',None)
140	To emit NL tokens: self.nltok = ('NL',None)
141
142	>>> l = WordLexer(r'abc "def\100\x3d\\. ghi" jkl')
143	>>> list(l)
144	[('ATOM', 'abc'), ('ATOM', 'def@=. ghi'), ('ATOM', 'jkl')]
145	>>> l = WordLexer(r"1 '2 3 4' 5")
146	>>> list(l)
147	[('ATOM', '1'), ('ATOM', '2 3 4'), ('ATOM', '5')]
148	>>> l = WordLexer("abc# a comment")
149	>>> list(l)
150	[('ATOM', 'abc'), ('COMMENT', 'a comment')]
151	"""
152
153	wordchars = set(string.ascii_letters) \| set(string.digits) \| set(string.punctuation)	5✔
154	quotechars = set("\"'")	5✔
155	commentchars = set("#")	5✔
156	spacechars = set(" \t\x0b\x0c")	5✔
157	nlchars = set("\r\n")	5✔
158	spacetok = None	5✔
159	nltok = None	5✔
160
161	def lex_start(self):	5✔
162	return (None, self.lex_space)	5✔
163
164	def lex_space(self):	5✔
165	s = []	5✔
166	if self.spacetok:	5✔
167	tok = lambda n: (self.spacetok, n) if s else (None, n)	5✔
168	else:
169	tok = lambda n: (None, n)	5✔
170	while not self.eof:	5!
171	c = self.peek()	5✔
172	if c in self.spacechars:	5✔
173	s.append(self.read())	5✔
174	elif c in self.nlchars:	5✔
175	return tok(self.lex_nl)	5✔
176	elif c in self.commentchars:	5✔
177	return tok(self.lex_comment)	5✔
178	elif c in self.quotechars:	5✔
179	return tok(self.lex_quote)	5✔
180	elif c in self.wordchars:	5!
181	return tok(self.lex_word)	5✔
182	elif c:	×
183	raise ValueError(	×
184	"Invalid input [%d]: %s"
185	% (
186	self.f.tell(),
187	c,
188	),
189	)
190	return (None, None)	×
191
192	def lex_nl(self):	5✔
193	while True:	3✔
194	c = self.read()	5✔
195	if c not in self.nlchars:	5✔
196	self.pushback(c)	5✔
197	return (self.nltok, self.lex_space)	5✔
198
199	def lex_comment(self):	5✔
200	s = []	5✔
201	tok = lambda n: (("COMMENT", "".join(s)), n) if s else (None, n)	5✔
202	start = False	5✔
203	_ = self.read()	5✔
204	while not self.eof:	5✔
205	c = self.read()	5✔
206	if c == "\n":	5✔
207	self.pushback(c)	5✔
208	return tok(self.lex_nl)	5✔
209	elif start or c not in string.whitespace:	5✔
210	start = True	5✔
211	s.append(c)	5✔
212	return tok(None)	5✔
213
214	def lex_word(self):	5✔
215	s = []	5✔
216	tok = lambda n: (("ATOM", "".join(s)), n) if s else (None, n)	5✔
217	while not self.eof:	5✔
218	c = self.peek()	5✔
219	if c == '"':	5!
220	return tok(self.lex_quote)	×
221	elif c in self.commentchars:	5✔
222	return tok(self.lex_comment)	5✔
223	elif c.isspace():	5✔
224	return tok(self.lex_space)	5✔
225	elif c in self.wordchars:	5✔
226	s.append(self.read())	5✔
227	elif c:	5!
228	raise ValueError(	×
229	"Invalid input [%d]: %s"
230	% (
231	self.f.tell(),
232	c,
233	),
234	)
235	return tok(None)	5✔
236
237	def lex_quote(self):	5✔
238	s = []	5✔
239	tok = lambda n: (("ATOM", "".join(s)), n)	5✔
240	q = self.read(1)	5✔
241	while not self.eof:	5!
242	c = self.readescaped()	5✔
243	if c == q and not self.escaped:	5✔
244	break	5✔
245	else:
246	s.append(c)	5✔
247	return tok(self.lex_space)	5✔
248
249
250	class RandomLexer(Lexer):	5✔
251	"""
252	Test lexing from infinite stream.
253
254	Extract strings of letters/numbers from /dev/urandom
255
256	>>> import itertools,sys
257	>>> if sys.version[0] == '2':
258	... f = open("/dev/urandom")
259	... else:
260	... f = open("/dev/urandom",encoding="ascii",errors="replace")
261	>>> r = RandomLexer(f)
262	>>> i = iter(r)
263	>>> len(list(itertools.islice(i,10)))
264	10
265	"""
266
267	minalpha = 4	5✔
268	mindigits = 3	5✔
269
270	def lex_start(self):	5✔
271	return (None, self.lex_random)	5✔
272
273	def lex_random(self):	5✔
274	n = 0	5✔
275	c = self.peek(1)	5✔
276	while not self.eof:	5!
277	if c.isalpha():	5✔
278	return (None, self.lex_alpha)	5✔
279	elif c.isdigit():	5✔
280	return (None, self.lex_digits)	5✔
281	else:
282	n += 1	5✔
283	_ = self.read(1)	5✔
284	c = self.peek(1)	5✔
285	return (None, None)	×
286
287	def lex_digits(self):	5✔
288	s = []	5✔
289	c = self.read(1)	5✔
290	while c.isdigit():	5✔
291	s.append(c)	5✔
292	c = self.read(1)	5✔
293	self.pushback(c)	5✔
294	if len(s) >= self.mindigits:	5!
UNCOV 295	return (("NUMBER", "".join(s)), self.lex_random)	×
296	else:
297	return (None, self.lex_random)	5✔
298
299	def lex_alpha(self):	5✔
300	s = []	5✔
301	c = self.read(1)	5✔
302	while c.isalpha():	5✔
303	s.append(c)	5✔
304	c = self.read(1)	5✔
305	self.pushback(c)	5✔
306	if len(s) >= self.minalpha:	5✔
307	return (("STRING", "".join(s)), self.lex_random)	5✔
308	else:
309	return (None, self.lex_random)	5✔
310
311
312	if __name__ == "__main__":	5!
313
314	import argparse	×
315	import doctest	×
316	import sys	×
317
318	p = argparse.ArgumentParser(description="Lex Tester")	×
319	p.add_argument(	×
320	"--lex",
321	"-l",
322	action="store_true",
323	default=False,
324	help="Lex input (stdin)",
325	)
326	p.add_argument(	×
327	"--nl",
328	action="store_true",
329	default=False,
330	help="Output NL tokens",
331	)
332	p.add_argument(	×
333	"--space",
334	action="store_true",
335	default=False,
336	help="Output Whitespace tokens",
337	)
338	p.add_argument("--wordchars", help="Wordchars")	×
339	p.add_argument("--quotechars", help="Quotechars")	×
340	p.add_argument("--commentchars", help="Commentchars")	×
341	p.add_argument("--spacechars", help="Spacechars")	×
342	p.add_argument("--nlchars", help="NLchars")	×
343
344	args = p.parse_args()	×
345
346	if args.lex:	×
347	l = WordLexer(sys.stdin)	×
348	if args.wordchars:	×
349	l.wordchars = set(args.wordchars)	×
350	if args.quotechars:	×
351	l.quotechars = set(args.quotechars)	×
352	if args.commentchars:	×
353	l.commentchars = set(args.commentchars)	×
354	if args.spacechars:	×
355	l.spacechars = set(args.spacechars)	×
356	if args.nlchars:	×
357	l.nlchars = set(args.nlchars)	×
358	if args.space:	×
359	l.spacetok = ("SPACE",)	×
360	if args.nl:	×
361	l.nltok = ("NL",)	×
362
363	for tok in l:	×
364	print(tok)	×
365	else:
366	try:	×
367	# Test if we have /dev/urandom
368	open("/dev/urandom")	×
369	sys.exit(0 if doctest.testmod().failed == 0 else 1)	×
370	except IOError:	×
371	# Don't run stream test
372	doctest.run_docstring_examples(Lexer, globals())	×
373	doctest.run_docstring_examples(WordLexer, globals())	×

Alviner / dnslib3 / 9384631762

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous