Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-27 02:24:18 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #8 from rouli/master

Browse Source 
Improving timestamp feature
This commit is contained in:
Aldo Cortesi 2013-01-16 19:15:39 -08:00
commit ad92d37147
2 changed files with 28 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
netlib/tcp.py
View File

@ -44,7 +44,7 @@ class _FileLike:
def __init__(self, o): def __init__(self, o):
self.o = o self.o = o
self._log = None self._log = None
self.timestamp = None self.first_byte_timestamp = None
def set_descriptor(self, o): def set_descriptor(self, o):
self.o = o self.o = o
@ -81,8 +81,8 @@ class _FileLike:
if self.is_logging(): if self.is_logging():
self._log.append(v) self._log.append(v)
def reset_timestamp(self): def reset_timestamps(self):
self.timestamp = None self.first_byte_timestamp = None
class Writer(_FileLike): class Writer(_FileLike):
def flush(self): def flush(self):
@ -134,7 +134,7 @@ class Reader(_FileLike):
raise NetLibDisconnect raise NetLibDisconnect
except SSL.SysCallError, v: except SSL.SysCallError, v:
raise NetLibDisconnect raise NetLibDisconnect
self.timestamp = self.timestamp or time.time() self.first_byte_timestamp = self.first_byte_timestamp or time.time()
if not data: if not data:
break break
result += data result += data

24
test/test_tcp.py
View File

@ -313,3 +313,27 @@ class TestFileLike:
s.write("x") s.write("x")
assert s.get_log() == "xx" assert s.get_log() == "xx"
def test_reset_timestamps(self):
s = cStringIO.StringIO("foobar\nfoobar")
s = tcp.Reader(s)
s.first_byte_timestamp = 500
s.reset_timestamps()
assert not s.first_byte_timestamp
def test_first_byte_timestamp_updated_on_read(self):
s = cStringIO.StringIO("foobar\nfoobar")
s = tcp.Reader(s)
s.read(1)
assert s.first_byte_timestamp
expected = s.first_byte_timestamp
s.read(5)
assert s.first_byte_timestamp == expected
def test_first_byte_timestamp_updated_on_readline(self):
s = cStringIO.StringIO("foobar\nfoobar\nfoobar")
s = tcp.Reader(s)
s.readline()
assert s.first_byte_timestamp
expected = s.first_byte_timestamp
s.readline()
assert s.first_byte_timestamp == expected