summaryrefslogtreecommitdiff
path: root/test/monkeyfarmer.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/monkeyfarmer.py')
-rw-r--r--test/monkeyfarmer.py390
1 files changed, 390 insertions, 0 deletions
diff --git a/test/monkeyfarmer.py b/test/monkeyfarmer.py
new file mode 100644
index 000000000..239a63e0e
--- /dev/null
+++ b/test/monkeyfarmer.py
@@ -0,0 +1,390 @@
+# Copyright 2017, 2018 Daniel Silverstone <dsilvers@digital-scurf.org>
+#
+# This file is part of NetSurf, http://www.netsurf-browser.org/
+#
+# NetSurf is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; version 2 of the License.
+#
+# NetSurf 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Monkey Farmer
+
+The monkey farmer is a wrapper around `nsmonkey` which can be used to simplify
+access to the monkey behaviours and ultimately to write useful tests in an
+expressive but not overcomplicated DSLish way. Tests are, ultimately, still
+Python code.
+
+"""
+
+import asyncore
+import os
+import socket
+import subprocess
+import time
+
+class MonkeyFarmer(asyncore.dispatcher):
+ def __init__(self, monkey_cmd, online, quiet=False):
+ (mine, monkeys) = socket.socketpair()
+
+ asyncore.dispatcher.__init__(self, sock=mine)
+
+ self.monkey = subprocess.Popen(
+ monkey_cmd,
+ stdin=monkeys,
+ stdout=monkeys,
+ close_fds=[mine])
+
+ monkeys.close()
+
+ self.buffer = b""
+ self.incoming = b""
+ self.lines = []
+ self.scheduled = []
+ self.deadmonkey = False
+ self.online = online
+ self.quiet = quiet
+ self.discussion = []
+
+ def handle_connect(self):
+ pass
+
+ def handle_read(self):
+ got = self.recv(8192)
+ if not got:
+ self.deadmonkey = True
+ return
+ self.incoming += got
+ if b"\n" in self.incoming:
+ lines = self.incoming.split(b"\n")
+ self.incoming = lines.pop()
+ self.lines = lines
+
+ def writable(self):
+ return (len(self.buffer) > 0)
+
+ def handle_write(self):
+ sent = self.send(self.buffer)
+ self.buffer = self.buffer[sent:]
+
+ def tell_monkey(self, *args):
+ cmd = (" ".join(args))
+ if not self.quiet:
+ print(">>> {}".format(cmd))
+ self.discussion.append((">",cmd))
+ cmd = cmd + "\n"
+ self.buffer += cmd.encode('utf-8')
+
+ def monkey_says(self, line):
+ line = line.decode('utf-8')
+ if not self.quiet:
+ print("<<< {}".format(line))
+ self.discussion.append(("<", line))
+ self.online(line)
+
+ def schedule_event(self, event, secs=None, when=None):
+ assert(secs is not None or when is not None)
+ if when is None:
+ when = time.time() + secs
+ self.scheduled.append((when, event))
+ self.scheduled.sort(lambda a,b: cmp(a[0],b[0]))
+
+ def unschedule_event(self, event):
+ self.scheduled = [x for x in self.scheduled if x[1] != event]
+
+ def loop(self, once=False):
+ if len(self.lines) > 0:
+ self.monkey_says(self.lines.pop(0))
+ if once:
+ return
+ while not self.deadmonkey:
+ now = time.time()
+ while len(self.scheduled) > 0 and now >= self.scheduled[0][0]:
+ func = self.scheduled[0][1]
+ self.scheduled.pop(0)
+ func(self)
+ now = time.time()
+ if len(self.scheduled) > 0:
+ next = self.scheduled[0][0]
+ asyncore.loop(timeout=next-now, count=1)
+ else:
+ asyncore.loop(count=1)
+ if len(self.lines) > 0:
+ self.monkey_says(self.lines.pop(0))
+ if once:
+ break
+
+class Browser:
+ def __init__(self, monkey_cmd=["./nsmonkey"], quiet=False):
+ self.farmer = MonkeyFarmer(monkey_cmd=monkey_cmd, online=self.on_monkey_line, quiet=quiet)
+ self.windows = {}
+ self.current_draw_target = None
+ self.started = False
+ self.stopped = False
+ self.launchurl = None
+ now = time.time()
+ while not self.started:
+ self.farmer.loop(once=True)
+ if (time.time() - now) > 1:
+ break
+
+ def pass_options(self, *opts):
+ if len(opts) > 0:
+ self.farmer.tell_monkey("OPTIONS " + (" ".join(opts)))
+
+ def on_monkey_line(self, line):
+ parts = line.split(" ")
+ handler = getattr(self, "handle_" + parts[0], None)
+ if handler is not None:
+ handler(*parts[1:])
+
+ def quit(self):
+ self.farmer.tell_monkey("QUIT")
+
+ def quit_and_wait(self):
+ self.quit()
+ self.farmer.loop()
+ return self.stopped
+
+ def handle_GENERIC(self, what, *args):
+ if what == 'STARTED':
+ self.started = True
+ elif what == 'FINISHED':
+ self.stopped = True
+ elif what == 'LAUNCH':
+ self.launchurl = args[1]
+ else:
+ # TODO: Nothing for now?
+ pass
+
+ def handle_WINDOW(self, action, _win, winid, *args):
+ if action == "NEW":
+ new_win = BrowserWindow(self, winid, *args)
+ self.windows[winid] = new_win
+ else:
+ win = self.windows.get(winid, None)
+ if win is None:
+ print(" Unknown window id {}".format(winid))
+ else:
+ win.handle(action, *args)
+
+ def handle_PLOT(self, *args):
+ if self.current_draw_target is not None:
+ self.current_draw_target.handle_plot(*args)
+
+ def new_window(self, url=None):
+ if url is None:
+ self.farmer.tell_monkey("WINDOW NEW")
+ else:
+ self.farmer.tell_monkey("WINDOW NEW %s" % url)
+ wins_known = set(self.windows.keys())
+ while len(set(self.windows.keys()).difference(wins_known)) == 0:
+ self.farmer.loop(once=True)
+ poss_wins = set(self.windows.keys()).difference(wins_known)
+ return self.windows[poss_wins.pop()]
+
+
+class BrowserWindow:
+ def __init__(self, browser, winid, _for, coreid, _existing, otherid, _newtab, newtab, _clone, clone):
+ self.alive = True
+ self.browser = browser
+ self.winid = winid
+ self.coreid = coreid
+ self.existing = browser.windows.get(otherid, None)
+ self.newtab = newtab == "TRUE"
+ self.clone = clone == "TRUE"
+ self.width = 0
+ self.height = 0
+ self.title = ""
+ self.throbbing = False
+ self.scrollx = 0
+ self.scrolly = 0
+ self.content_width = 0
+ self.content_height = 0
+ self.status = ""
+ self.pointer = ""
+ self.scale = 1.0
+ self.url = ""
+ self.plotted = []
+ self.plotting = False
+
+ def kill(self):
+ self.browser.farmer.tell_monkey("WINDOW DESTROY %s" % self.winid)
+
+ def wait_until_dead(self, timeout=1):
+ now = time.time()
+ while self.alive:
+ self.browser.farmer.loop(once=True)
+ if (time.time() - now) > timeout:
+ break
+
+ def go(self, url, referer = None):
+ if referer is None:
+ self.browser.farmer.tell_monkey("WINDOW GO %s %s" % (
+ self.winid, url))
+ else:
+ self.browser.farmer.tell_monkey("WINDOW GO %s %s %s" % (
+ self.winid, url, referer))
+ self.wait_start_loading()
+
+ def reload(self):
+ self.browser.farmer.tell_monkey("WINDOW RELOAD %s" % self.winid)
+
+ def handle(self, action, *args):
+ handler = getattr(self, "handle_window_" + action, None)
+ if handler is not None:
+ handler(*args)
+
+ def handle_window_SIZE(self, _width, width, _height, height):
+ self.width = int(width)
+ self.height = int(height)
+
+ def handle_window_DESTROY(self):
+ self.alive = False
+
+ def handle_window_TITLE(self, _str, *title):
+ self.title = " ".join(title)
+
+ def handle_window_REDRAW(self):
+ pass
+
+ def handle_window_GET_DIMENSIONS(self, _width, width, _height, height):
+ self.width = width
+ self.height = height
+
+ def handle_window_NEW_CONTENT(self):
+ pass
+
+ def handle_window_NEW_ICON(self):
+ pass
+
+ def handle_window_START_THROBBER(self):
+ self.throbbing = True
+
+ def handle_window_STOP_THROBBER(self):
+ self.throbbing = False
+
+ def handle_window_SET_SCROLL(self, _x, x, _y, y):
+ self.scrollx = int(x)
+ self.scrolly = int(y)
+
+ def handle_window_UPDATE_BOX(self, _x, x, _y, y, _width, width, _height, height):
+ x = int(x)
+ y = int(y)
+ width = int(width)
+ height = int(height)
+ pass
+
+ def handle_window_UPDATE_EXTENT(self, _width, width, _height, height):
+ self.content_width = int(width)
+ self.content_height = int(height)
+
+ def handle_window_SET_STATUS(self, _str, *status):
+ self.status = (" ".join(status))
+
+ def handle_window_SET_POINTER(self, _ptr, ptr):
+ self.pointer = ptr
+
+ def handle_window_SET_SCALE(self, _scale, scale):
+ self.scale = float(scale)
+
+ def handle_window_SET_URL(self, _url, url):
+ self.url = url
+
+ def handle_window_GET_SCROLL(self, _x, x, _y, y):
+ self.scrollx = int(x)
+ self.scrolly = int(y)
+
+ def handle_window_SCROLL_START(self):
+ self.scrollx = 0
+ self.scrolly = 0
+
+ def handle_window_REDRAW(self, act):
+ if act == "START":
+ self.browser.current_draw_target = self
+ self.plotted = []
+ self.plotting = True
+ else:
+ self.browser.current_draw_target = None
+ self.plotting = False
+
+ def load_page(self, url=None, referer=None):
+ if url is not None:
+ self.go(url, referer)
+ self.wait_loaded()
+
+ def wait_start_loading(self):
+ while not self.throbbing:
+ self.browser.farmer.loop(once=True)
+
+ def wait_loaded(self):
+ self.wait_start_loading()
+ while self.throbbing:
+ self.browser.farmer.loop(once=True)
+
+ def handle_plot(self, *args):
+ self.plotted.append(args)
+
+ def redraw(self, coords=None):
+ if coords is None:
+ self.browser.farmer.tell_monkey("WINDOW REDRAW %s" % self.winid)
+ else:
+ self.browser.farmer.tell_monkey("WINDOW REDRAW %s %s" % (
+ self.winid, (" ".join(coords))))
+ while not self.plotting:
+ self.browser.farmer.loop(once=True)
+ while self.plotting:
+ self.browser.farmer.loop(once=True)
+ return self.plotted
+
+
+if __name__ == '__main__':
+ # Simple test is as follows...
+
+ browser = Browser(quiet=True)
+ win = browser.new_window()
+
+ fname = "test/js/inline-doc-write-simple.html"
+ full_fname = os.path.join(os.getcwd(), fname)
+
+ browser.pass_options("--enable_javascript=0")
+ win.load_page("file://" + full_fname)
+
+ print("Loaded, URL is {}".format(win.url))
+
+ cmds = win.redraw()
+ print("Received {} plot commands".format(len(cmds)))
+ for cmd in cmds:
+ if cmd[0] == "TEXT":
+ x = cmd[2]
+ y = cmd[4]
+ rest = " ".join(cmd[6:])
+ print("{} {} -> {}".format(x,y,rest))
+
+
+ browser.pass_options("--enable_javascript=1")
+ win.load_page("file://" + full_fname)
+
+ print("Loaded, URL is {}".format(win.url))
+
+ cmds = win.redraw()
+ print("Received {} plot commands".format(len(cmds)))
+ for cmd in cmds:
+ if cmd[0] == "TEXT":
+ x = cmd[2]
+ y = cmd[4]
+ rest = " ".join(cmd[6:])
+ print("{} {} -> {}".format(x,y,rest))
+
+ browser.quit_and_wait()
+
+ #print("Discussion was:")
+ #for line in browser.farmer.discussion:
+ # print("{} {}".format(line[0], line[1]))