| #!/usr/bin/env python |
| |
| # These tests access the network. python test.py runs a local test server and |
| # doesn't try to fetch anything over the internet, since the few tests here |
| # that do that are disabled by default since they have test tag "internet". |
| |
| # thanks Moof (aka Giles Antonio Radford) for some of these |
| |
| import errno |
| import os |
| import socket |
| import subprocess |
| import sys |
| import unittest |
| import urllib |
| import urllib2 |
| |
| import mechanize |
| from mechanize import CookieJar, HTTPCookieProcessor, \ |
| HTTPHandler, HTTPRefreshProcessor, \ |
| HTTPEquivProcessor, HTTPRedirectHandler, \ |
| HTTPRedirectDebugProcessor, HTTPResponseDebugProcessor |
| from mechanize._rfc3986 import urljoin |
| from mechanize._util import hide_experimental_warnings, \ |
| reset_experimental_warnings, write_file |
| import mechanize._opener |
| import mechanize._rfc3986 |
| import mechanize._sockettimeout |
| import mechanize._testcase |
| |
| |
| #from cookielib import CookieJar |
| #from urllib2 import build_opener, install_opener, urlopen |
| #from urllib2 import HTTPCookieProcessor, HTTPHandler |
| |
| #from mechanize import CreateBSDDBCookieJar |
| |
| ## import logging |
| ## logger = logging.getLogger("mechanize") |
| ## logger.addHandler(logging.StreamHandler(sys.stdout)) |
| ## #logger.setLevel(logging.DEBUG) |
| ## logger.setLevel(logging.INFO) |
| |
| |
| class TestCase(mechanize._testcase.TestCase): |
| |
| # testprogram sets self.no_proxies on each TestCase to request explicitly |
| # setting proxies so that http*_proxy environment variables are ignored |
| |
| def _configure_user_agent(self, ua): |
| if self.no_proxies: |
| ua.set_proxies({}) |
| |
| def make_browser(self): |
| browser = mechanize.Browser() |
| self._configure_user_agent(browser) |
| return browser |
| |
| def make_user_agent(self): |
| ua = mechanize.UserAgent() |
| self._configure_user_agent(ua) |
| return ua |
| |
| def build_opener(self, handlers=(), build_opener=None): |
| handlers += (mechanize.ProxyHandler(proxies={}),) |
| if build_opener is None: |
| build_opener = mechanize.build_opener |
| return build_opener(*handlers) |
| |
| def setUp(self): |
| mechanize._testcase.TestCase.setUp(self) |
| self.test_uri = urljoin(self.uri, "test_fixtures") |
| self.server = self.get_cached_fixture("server") |
| if self.no_proxies: |
| old_opener_m = mechanize._opener._opener |
| old_opener_u = urllib2._opener |
| mechanize.install_opener(mechanize.build_opener( |
| mechanize.ProxyHandler(proxies={}))) |
| urllib2.install_opener(urllib2.build_opener( |
| urllib2.ProxyHandler(proxies={}))) |
| def revert_install(): |
| mechanize.install_opener(old_opener_m) |
| urllib2.install_opener(old_opener_u) |
| self.add_teardown(revert_install) |
| |
| |
| def sanepathname2url(path): |
| urlpath = urllib.pathname2url(path) |
| if os.name == "nt" and urlpath.startswith("///"): |
| urlpath = urlpath[2:] |
| # XXX don't ask me about the mac... |
| return urlpath |
| |
| |
| def read_file(filename): |
| fh = open(filename) |
| try: |
| return fh.read() |
| finally: |
| fh.close() |
| |
| |
| class FtpTestCase(TestCase): |
| |
| def test_ftp(self): |
| server = self.get_cached_fixture("ftp_server") |
| browser = self.make_browser() |
| path = self.make_temp_dir(dir_=server.root_path) |
| file_path = os.path.join(path, "stuff") |
| data = "data\nmore data" |
| write_file(file_path, data) |
| relative_path = os.path.join(os.path.basename(path), "stuff") |
| r = browser.open("ftp://anon@localhost:%s/%s" % |
| (server.port, relative_path)) |
| self.assertEqual(r.read(), data) |
| |
| |
| class SocketTimeoutTest(TestCase): |
| |
| # the timeout tests in this module aren't full functional tests: in order |
| # to speed things up, don't actually call .settimeout on the socket. XXX |
| # allow running the tests against a slow server with a real timeout |
| |
| def _monkey_patch_socket(self): |
| class Delegator(object): |
| def __init__(self, delegate): |
| self._delegate = delegate |
| def __getattr__(self, name): |
| return getattr(self._delegate, name) |
| |
| assertEquals = self.assertEquals |
| |
| class TimeoutLog(object): |
| AnyValue = object() |
| def __init__(self): |
| self._nr_sockets = 0 |
| self._timeouts = [] |
| self.start() |
| def start(self): |
| self._monitoring = True |
| def stop(self): |
| self._monitoring = False |
| def socket_created(self): |
| if self._monitoring: |
| self._nr_sockets += 1 |
| def settimeout_called(self, timeout): |
| if self._monitoring: |
| self._timeouts.append(timeout) |
| def verify(self, value=AnyValue): |
| if sys.version_info[:2] < (2, 6): |
| # per-connection timeout not supported in Python 2.5 |
| self.verify_default() |
| else: |
| assertEquals(len(self._timeouts), self._nr_sockets) |
| if value is not self.AnyValue: |
| for timeout in self._timeouts: |
| assertEquals(timeout, value) |
| def verify_default(self): |
| assertEquals(len(self._timeouts), 0) |
| |
| log = TimeoutLog() |
| def settimeout(timeout): |
| log.settimeout_called(timeout) |
| orig_socket = socket.socket |
| def make_socket(*args, **kwds): |
| sock = Delegator(orig_socket(*args, **kwds)) |
| log.socket_created() |
| sock.settimeout = settimeout |
| return sock |
| self.monkey_patch(socket, "socket", make_socket) |
| return log |
| |
| |
| class SimpleTests(SocketTimeoutTest): |
| # thanks Moof (aka Giles Antonio Radford) |
| |
| def setUp(self): |
| super(SimpleTests, self).setUp() |
| self.browser = self.make_browser() |
| |
| def test_simple(self): |
| self.browser.open(self.test_uri) |
| self.assertEqual(self.browser.title(), 'Python bits') |
| # relative URL |
| self.browser.open('/mechanize/') |
| self.assertEqual(self.browser.title(), 'mechanize') |
| |
| def test_basic_auth(self): |
| uri = urljoin(self.uri, "basic_auth") |
| self.assertRaises(mechanize.URLError, self.browser.open, uri) |
| self.browser.add_password(uri, "john", "john") |
| self.browser.open(uri) |
| self.assertEqual(self.browser.title(), 'Basic Auth Protected Area') |
| |
| def test_digest_auth(self): |
| uri = urljoin(self.uri, "digest_auth") |
| self.assertRaises(mechanize.URLError, self.browser.open, uri) |
| self.browser.add_password(uri, "digestuser", "digestuser") |
| self.browser.open(uri) |
| self.assertEqual(self.browser.title(), 'Digest Auth Protected Area') |
| |
| def test_open_with_default_timeout(self): |
| timeout_log = self._monkey_patch_socket() |
| self.browser.open(self.test_uri) |
| self.assertEqual(self.browser.title(), 'Python bits') |
| timeout_log.verify_default() |
| |
| def test_open_with_timeout(self): |
| timeout_log = self._monkey_patch_socket() |
| timeout = 10. |
| self.browser.open(self.test_uri, timeout=timeout) |
| self.assertEqual(self.browser.title(), 'Python bits') |
| timeout_log.verify(timeout) |
| |
| def test_urlopen_with_default_timeout(self): |
| timeout_log = self._monkey_patch_socket() |
| response = mechanize.urlopen(self.test_uri) |
| self.assert_contains(response.read(), "Python bits") |
| timeout_log.verify_default() |
| |
| def test_urlopen_with_timeout(self): |
| timeout_log = self._monkey_patch_socket() |
| timeout = 10. |
| response = mechanize.urlopen(self.test_uri, timeout=timeout) |
| self.assert_contains(response.read(), "Python bits") |
| timeout_log.verify(timeout) |
| |
| def test_redirect_with_timeout(self): |
| timeout_log = self._monkey_patch_socket() |
| timeout = 10. |
| # 301 redirect due to missing final '/' |
| req = mechanize.Request(urljoin(self.test_uri, "test_fixtures"), |
| timeout=timeout) |
| r = self.browser.open(req) |
| self.assert_("GeneralFAQ.html" in r.read(2048)) |
| timeout_log.verify(timeout) |
| |
| def test_302_and_404(self): |
| # the combination of 302 and 404 (/redirected is configured to redirect |
| # to a non-existent URL /nonexistent) has caused problems in the past |
| # due to accidental double-wrapping of the error response |
| self.assertRaises( |
| mechanize.HTTPError, |
| self.browser.open, urljoin(self.uri, "/redirected"), |
| ) |
| |
| def test_reread(self): |
| # closing response shouldn't stop methods working (this happens also to |
| # be true for e.g. mechanize.OpenerDirector when mechanize's own |
| # handlers are in use, but is guaranteed to be true for |
| # mechanize.Browser) |
| r = self.browser.open(self.uri) |
| data = r.read() |
| r.close() |
| r.seek(0) |
| self.assertEqual(r.read(), data) |
| self.assertEqual(self.browser.response().read(), data) |
| |
| def test_error_recovery(self): |
| self.assertRaises(mechanize.URLError, self.browser.open, |
| 'file:///c|thisnoexistyiufheiurgbueirgbue') |
| self.browser.open(self.test_uri) |
| self.assertEqual(self.browser.title(), 'Python bits') |
| |
| def test_redirect(self): |
| # 301 redirect due to missing final '/' |
| codes = [] |
| class ObservingHandler(mechanize.BaseHandler): |
| def http_response(self, request, response): |
| codes.append(response.code) |
| return response |
| self.browser.add_handler(ObservingHandler()) |
| r = self.browser.open(urljoin(self.uri, "test_fixtures")) |
| self.assertEqual(r.code, 200) |
| self.assertTrue(301 in codes) |
| self.assert_("GeneralFAQ.html" in r.read(2048)) |
| |
| def test_refresh(self): |
| def refresh_request(seconds): |
| uri = urljoin(self.uri, "/cgi-bin/cookietest.cgi") |
| val = urllib.quote_plus('%d; url="%s"' % (seconds, self.uri)) |
| return uri + ("?refresh=%s" % val) |
| self.browser.set_handle_refresh(True, honor_time=False) |
| r = self.browser.open(refresh_request(5)) |
| self.assertEqual(r.geturl(), self.uri) |
| # Set a maximum refresh time of 30 seconds (these long refreshes tend |
| # to be there only because the website owner wants you to see the |
| # latest news, or whatever -- they're not essential to the operation of |
| # the site, and not really useful or appropriate when scraping). |
| refresh_uri = refresh_request(60) |
| self.browser.set_handle_refresh(True, max_time=30., honor_time=True) |
| r = self.browser.open(refresh_uri) |
| self.assertEqual(r.geturl(), refresh_uri) |
| # allow long refreshes (but don't actually wait 60 seconds) |
| self.browser.set_handle_refresh(True, max_time=None, honor_time=False) |
| r = self.browser.open(refresh_request(60)) |
| self.assertEqual(r.geturl(), self.uri) |
| |
| def test_file_url(self): |
| url = "file://%s" % sanepathname2url( |
| os.path.abspath(os.path.join("test", "test_functional.py"))) |
| r = self.browser.open(url) |
| self.assert_("this string appears in this file ;-)" in r.read()) |
| |
| def test_open_local_file(self): |
| # Since the file: URL scheme is not well standardised, Browser has a |
| # special method to open files by name, for convenience: |
| path = os.path.join("test", "test_functional.py") |
| response = self.browser.open_local_file(path) |
| self.assertIn("this string appears in this file ;-)", |
| response.get_data()) |
| |
| def test_open_novisit(self): |
| def test_state(br): |
| self.assert_(br.request is None) |
| self.assert_(br.response() is None) |
| self.assertRaises(mechanize.BrowserStateError, br.back) |
| test_state(self.browser) |
| uri = urljoin(self.uri, "test_fixtures") |
| # note this involves a redirect, which should itself be non-visiting |
| r = self.browser.open_novisit(uri) |
| test_state(self.browser) |
| self.assert_("GeneralFAQ.html" in r.read(2048)) |
| |
| # Request argument instead of URL |
| r = self.browser.open_novisit(mechanize.Request(uri)) |
| test_state(self.browser) |
| self.assert_("GeneralFAQ.html" in r.read(2048)) |
| |
| def test_non_seekable(self): |
| # check everything still works without response_seek_wrapper and |
| # the .seek() method on response objects |
| ua = self.make_user_agent() |
| ua.set_seekable_responses(False) |
| ua.set_handle_equiv(False) |
| response = ua.open(self.test_uri) |
| self.failIf(hasattr(response, "seek")) |
| data = response.read() |
| self.assert_("Python bits" in data) |
| |
| |
| class ResponseTests(TestCase): |
| |
| def test_seek(self): |
| br = self.make_browser() |
| r = br.open(self.uri) |
| html = r.read() |
| r.seek(0) |
| self.assertEqual(r.read(), html) |
| |
| def test_seekable_response_opener(self): |
| build_opener = mechanize.OpenerFactory( |
| mechanize.SeekableResponseOpener).build_opener |
| opener = self.build_opener(build_opener=build_opener) |
| r = opener.open(urljoin(self.uri, "test_fixtures/cctest2.txt")) |
| r.read() |
| r.seek(0) |
| self.assertEqual(r.read(), |
| r.get_data(), |
| "Hello ClientCookie functional test suite.\n") |
| |
| def test_seek_wrapper_class_name(self): |
| opener = self.make_user_agent() |
| opener.set_seekable_responses(True) |
| try: |
| opener.open(urljoin(self.uri, "nonexistent")) |
| except mechanize.HTTPError, exc: |
| self.assert_("HTTPError instance" in repr(exc)) |
| |
| def test_no_seek(self): |
| # should be possible to turn off UserAgent's .seek() functionality |
| def check_no_seek(opener): |
| r = opener.open(urljoin(self.uri, "test_fixtures/cctest2.txt")) |
| self.assert_(not hasattr(r, "seek")) |
| try: |
| opener.open(urljoin(self.uri, "nonexistent")) |
| except mechanize.HTTPError, exc: |
| self.assert_(not hasattr(exc, "seek")) |
| |
| # mechanize.UserAgent |
| opener = self.make_user_agent() |
| opener.set_handle_equiv(False) |
| opener.set_seekable_responses(False) |
| opener.set_debug_http(False) |
| check_no_seek(opener) |
| |
| # mechanize.OpenerDirector |
| opener = self.build_opener() |
| check_no_seek(opener) |
| |
| def test_consistent_seek(self): |
| # if we explicitly request that returned response objects have the |
| # .seek() method, then raised HTTPError exceptions should also have the |
| # .seek() method |
| def check(opener, excs_also): |
| r = opener.open(urljoin(self.uri, "test_fixtures/cctest2.txt")) |
| data = r.read() |
| r.seek(0) |
| self.assertEqual(data, r.read(), r.get_data()) |
| try: |
| opener.open(urljoin(self.uri, "nonexistent")) |
| except mechanize.HTTPError, exc: |
| data = exc.read() |
| if excs_also: |
| exc.seek(0) |
| self.assertEqual(data, exc.read(), exc.get_data()) |
| else: |
| self.assert_(False) |
| |
| opener = self.make_user_agent() |
| opener.set_debug_http(False) |
| |
| # Here, only the .set_handle_equiv() causes .seek() to be present, so |
| # exceptions don't necessarily support the .seek() method (and do not, |
| # at present). |
| opener.set_handle_equiv(True) |
| opener.set_seekable_responses(False) |
| check(opener, excs_also=False) |
| |
| # Here, (only) the explicit .set_seekable_responses() causes .seek() to |
| # be present (different mechanism from .set_handle_equiv()). Since |
| # there's an explicit request, ALL responses are seekable, even |
| # exception responses (HTTPError instances). |
| opener.set_handle_equiv(False) |
| opener.set_seekable_responses(True) |
| check(opener, excs_also=True) |
| |
| def test_set_response(self): |
| br = self.make_browser() |
| r = br.open(self.test_uri) |
| html = r.read() |
| self.assertEqual(br.title(), "Python bits") |
| |
| newhtml = """<html><body><a href="spam">click me</a></body></html>""" |
| |
| r.set_data(newhtml) |
| self.assertEqual(r.read(), newhtml) |
| self.assertEqual(br.response().read(), html) |
| br.response().set_data(newhtml) |
| self.assertEqual(br.response().read(), html) |
| self.assertEqual(list(br.links())[0].url, "http://sourceforge.net/") |
| |
| br.set_response(r) |
| self.assertEqual(br.response().read(), newhtml) |
| self.assertEqual(list(br.links())[0].url, "spam") |
| |
| def test_new_response(self): |
| br = self.make_browser() |
| data = ("<html><head><title>Test</title></head>" |
| "<body><p>Hello.</p></body></html>") |
| response = mechanize.make_response( |
| data, |
| [("Content-type", "text/html")], |
| "http://example.com/", |
| 200, |
| "OK") |
| br.set_response(response) |
| self.assertEqual(br.response().get_data(), data) |
| |
| def hidden_test_close_pickle_load(self): |
| print ("Test test_close_pickle_load is expected to fail unless Python " |
| "standard library patch http://python.org/sf/1144636 has been " |
| "applied") |
| import pickle |
| |
| b = self.make_browser() |
| r = b.open(urljoin(self.uri, "test_fixtures/cctest2.txt")) |
| r.read() |
| |
| r.close() |
| r.seek(0) |
| self.assertEqual(r.read(), |
| "Hello ClientCookie functional test suite.\n") |
| |
| HIGHEST_PROTOCOL = -1 |
| p = pickle.dumps(b, HIGHEST_PROTOCOL) |
| b = pickle.loads(p) |
| r = b.response() |
| r.seek(0) |
| self.assertEqual(r.read(), |
| "Hello ClientCookie functional test suite.\n") |
| |
| |
| class FunctionalTests(SocketTimeoutTest): |
| |
| def test_referer(self): |
| br = self.make_browser() |
| br.set_handle_refresh(True, honor_time=False) |
| referer = urljoin(self.uri, "test_fixtures/referertest.html") |
| info = urljoin(self.uri, "/cgi-bin/cookietest.cgi") |
| r = br.open(info) |
| self.assert_(referer not in r.get_data()) |
| |
| br.open(referer) |
| r = br.follow_link(text="Here") |
| self.assert_(referer in r.get_data()) |
| |
| def test_cookies(self): |
| # this test page depends on cookies, and an http-equiv refresh |
| #cj = CreateBSDDBCookieJar("/home/john/db.db") |
| cj = CookieJar() |
| handlers = [ |
| HTTPCookieProcessor(cj), |
| HTTPRefreshProcessor(max_time=None, honor_time=False), |
| HTTPEquivProcessor(), |
| |
| HTTPRedirectHandler(), # needed for Refresh handling in 2.4.0 |
| # HTTPHandler(True), |
| # HTTPRedirectDebugProcessor(), |
| # HTTPResponseDebugProcessor(), |
| ] |
| |
| opener = self.build_opener(handlers) |
| r = opener.open(urljoin(self.uri, "/cgi-bin/cookietest.cgi")) |
| data = r.read() |
| self.assert_(data.find("Your browser supports cookies!") >= 0) |
| self.assertEquals(len(cj), 2) |
| |
| # test response.seek() (added by HTTPEquivProcessor) |
| r.seek(0) |
| samedata = r.read() |
| r.close() |
| self.assertEquals(samedata, data) |
| |
| def test_robots(self): |
| plain_opener = self.build_opener( |
| [mechanize.HTTPRobotRulesProcessor]) |
| browser = self.make_browser() |
| for opener in plain_opener, browser: |
| opener.open(urljoin(self.uri, "robots")) |
| self.assertRaises( |
| mechanize.RobotExclusionError, |
| opener.open, urljoin(self.uri, "norobots")) |
| |
| def _check_retrieve(self, url, filename, headers): |
| from urllib import urlopen |
| self.assertEqual(headers.get('Content-Type'), 'text/html') |
| if self.no_proxies: |
| proxies = {} |
| else: |
| proxies = None |
| self.assertEqual(read_file(filename), |
| urlopen(url, proxies=proxies).read()) |
| |
| def test_retrieve_to_named_file(self): |
| url = urljoin(self.uri, "/mechanize/") |
| test_filename = os.path.join(self.make_temp_dir(), "python.html") |
| opener = self.build_opener() |
| verif = CallbackVerifier(self) |
| filename, headers = opener.retrieve(url, test_filename, verif.callback) |
| self.assertEqual(filename, test_filename) |
| self._check_retrieve(url, filename, headers) |
| self.assert_(os.path.isfile(filename)) |
| |
| def test_retrieve(self): |
| # not passing an explicit filename downloads to a temporary file |
| # using a Request object instead of a URL works |
| url = urljoin(self.uri, "/mechanize/") |
| opener = self.build_opener() |
| verif = CallbackVerifier(self) |
| request = mechanize.Request(url) |
| filename, headers = opener.retrieve(request, reporthook=verif.callback) |
| self.assertEquals(request.visit, False) |
| self._check_retrieve(url, filename, headers) |
| opener.close() |
| # closing the opener removed the temporary file |
| self.failIf(os.path.isfile(filename)) |
| |
| def test_urlretrieve(self): |
| timeout_log = self._monkey_patch_socket() |
| timeout = 10. |
| url = urljoin(self.uri, "/mechanize/") |
| verif = CallbackVerifier(self) |
| filename, headers = mechanize.urlretrieve(url, |
| reporthook=verif.callback, |
| timeout=timeout) |
| timeout_log.stop() |
| self._check_retrieve(url, filename, headers) |
| timeout_log.verify(timeout) |
| |
| def test_reload_read_incomplete(self): |
| browser = self.make_browser() |
| r1 = browser.open(urljoin(self.uri, |
| "test_fixtures/mechanize_reload_test.html")) |
| # if we don't do anything and go straight to another page, most of the |
| # last page's response won't be .read()... |
| r2 = browser.open(urljoin(self.uri, "mechanize")) |
| self.assert_(len(r1.get_data()) < 4097) # we only .read() a little bit |
| # ...so if we then go back, .follow_link() for a link near the end (a |
| # few kb in, past the point that always gets read in HTML files because |
| # of HEAD parsing) will only work if it causes a .reload()... |
| r3 = browser.back() |
| browser.follow_link(text="near the end") |
| # ... good, no LinkNotFoundError, so we did reload. |
| # we have .read() the whole file |
| self.assertEqual(len(r3._seek_wrapper__cache.getvalue()), 4202) |
| |
| ## def test_cacheftp(self): |
| ## from mechanize import CacheFTPHandler, build_opener |
| ## o = build_opener(CacheFTPHandler()) |
| ## r = o.open("ftp://ftp.python.org/pub/www.python.org/robots.txt") |
| ## data1 = r.read() |
| ## r.close() |
| ## r = o.open("ftp://ftp.python.org/pub/www.python.org/2.3.2/announce.txt") |
| ## data2 = r.read() |
| ## r.close() |
| ## self.assert_(data1 != data2) |
| |
| |
| class CommandFailedError(Exception): |
| |
| def __init__(self, message, rc): |
| Exception.__init__(self, message) |
| self.rc = rc |
| |
| |
| def get_cmd_stdout(args, **kwargs): |
| process = subprocess.Popen(args, stdout=subprocess.PIPE, **kwargs) |
| stdout, stderr = process.communicate() |
| rc = process.returncode |
| if rc != 0: |
| raise CommandFailedError( |
| "Command failed with return code %i: %s:\n%s" % |
| (rc, args, stderr), rc) |
| else: |
| return stdout |
| |
| |
| class ExamplesTests(TestCase): |
| |
| tags = "internet" |
| |
| def check_download_script(self, name): |
| python = sys.executable |
| parent_dir = os.path.dirname(os.path.dirname( |
| os.path.abspath(__file__))) |
| temp_dir = self.make_temp_dir() |
| get_cmd_stdout( |
| [python, os.path.join(parent_dir, "examples", name)], |
| cwd=temp_dir) |
| [tarball] = os.listdir(temp_dir) |
| self.assertTrue(tarball.endswith(".tar.gz")) |
| |
| def test_hack21(self): |
| self.check_download_script("hack21.py") |
| |
| def test_pypi(self): |
| self.check_download_script("pypi.py") |
| |
| |
| def add_to_path(env, name, value): |
| old = env.get(name) |
| if old is not None and old != "": |
| value = old + ":" + value |
| env[name] = value |
| |
| |
| class FormsExamplesTests(mechanize._testcase.GoldenTestCase): |
| |
| def check_forms_example(self, name, golden_path, fixup): |
| self.get_cached_fixture("server") |
| python = sys.executable |
| this_dir = os.path.dirname(os.path.abspath(__file__)) |
| parent_dir = os.path.dirname(this_dir) |
| forms_examples_dir = os.path.join(parent_dir, "examples", "forms") |
| output_dir = self.make_temp_dir() |
| env = os.environ.copy() |
| add_to_path(env, "PYTHONPATH", parent_dir) |
| output = get_cmd_stdout([python, name, self.uri], |
| env=env, |
| cwd=forms_examples_dir) |
| output = fixup(output) |
| write_file(os.path.join(output_dir, "output"), output) |
| self.assert_golden(output_dir, |
| os.path.join(this_dir, golden_path)) |
| |
| def test_simple(self): |
| def fixup(output): |
| return output.replace("POST %s" % self.uri.rstrip("/"), |
| "POST http://127.0.0.1:8000") |
| self.check_forms_example( |
| "simple.py", |
| os.path.join("functional_tests_golden", |
| "FormsExamplesTests.test_simple"), |
| fixup) |
| |
| def test_example(self): |
| def fixup(output): |
| lines = [l for l in output.splitlines(True) if |
| not l.startswith("Vary:") and |
| not l.startswith("Server:") and |
| not l.startswith("Transfer-Encoding:") and |
| not l.startswith("Content-Length:")] |
| output = "".join(lines) |
| return output.replace(self.uri.rstrip("/"), |
| "http://127.0.0.1:8000") |
| self.check_forms_example( |
| "example.py", |
| os.path.join("functional_tests_golden", |
| "FormsExamplesTests.test_example"), |
| fixup) |
| |
| |
| class CookieJarTests(TestCase): |
| |
| def _test_cookiejar(self, make_cookiejar, commit): |
| cookiejar = make_cookiejar() |
| br = self.make_browser() |
| #br.set_debug_http(True) |
| br.set_cookiejar(cookiejar) |
| br.set_handle_refresh(False) |
| url = urljoin(self.uri, "/cgi-bin/cookietest.cgi") |
| # no cookie was set on the first request |
| html = br.open(url).read() |
| self.assertEquals(html.find("Your browser supports cookies!"), -1) |
| self.assertEquals(len(cookiejar), 2) |
| # ... but now we have the cookie |
| html = br.open(url).read() |
| self.assertIn("Your browser supports cookies!", html) |
| self.assertIn("Received session cookie", html) |
| commit(cookiejar) |
| |
| # should still have the cookie when we load afresh |
| cookiejar = make_cookiejar() |
| br.set_cookiejar(cookiejar) |
| html = br.open(url).read() |
| self.assertIn("Your browser supports cookies!", html) |
| self.assertNotIn("Received session cookie", html) |
| |
| def test_mozilla_cookiejar(self): |
| filename = os.path.join(self.make_temp_dir(), "cookies.txt") |
| def make_cookiejar(): |
| cj = mechanize.MozillaCookieJar(filename=filename) |
| try: |
| cj.revert() |
| except IOError, exc: |
| if exc.errno != errno.ENOENT: |
| raise |
| return cj |
| def commit(cj): |
| cj.save() |
| self._test_cookiejar(make_cookiejar, commit) |
| |
| def test_firefox3_cookiejar(self): |
| try: |
| mechanize.Firefox3CookieJar |
| except AttributeError: |
| # firefox 3 cookiejar is only supported in Python 2.5 and later; |
| # also, sqlite3 must be available |
| raise unittest.SkipTest() |
| |
| filename = os.path.join(self.make_temp_dir(), "cookies.sqlite") |
| def make_cookiejar(): |
| hide_experimental_warnings() |
| try: |
| return mechanize.Firefox3CookieJar(filename=filename) |
| finally: |
| reset_experimental_warnings() |
| def commit(cj): |
| pass |
| self._test_cookiejar(make_cookiejar, commit) |
| |
| |
| class CallbackVerifier: |
| # for .test_urlretrieve() |
| def __init__(self, testcase): |
| self._count = 0 |
| self._testcase = testcase |
| def callback(self, block_nr, block_size, total_size): |
| self._testcase.assertEqual(block_nr, self._count) |
| self._count = self._count + 1 |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
