Commit 435cdb3d authored by ale's avatar ale
Browse files

Add a simple WebDAV probe

Also close the response and session after each usage, to attempt to
mitigate the memory leaks.
parent 6c990e21
Pipeline #7789 passed with stages
in 1 minute and 58 seconds
......@@ -5,7 +5,7 @@ from ai_diagnostics.adapters import ForcedIPHTTPSAdapter, LoggingHTTPAdapter
@contextlib.contextmanager
def make_browser(ctx):
def make_session(ctx):
session = requests.Session()
# Install DNS overrides, or just a simple logging adapter.
......@@ -17,24 +17,34 @@ def make_browser(ctx):
else:
session.mount('https://', LoggingHTTPAdapter(ctx=ctx))
browser = mechanicalsoup.StatefulBrowser(
soup_config={'features': 'lxml'},
raise_on_404=True,
session=session,
)
browser.set_verbose(2)
# Print a dump of browser state on errors.
try:
yield browser
except Exception as e:
ctx.log('ERROR: %s', e)
ctx.log('Browser URL: %s\n\nPage contents:\n\n%s',
browser.get_url(),
browser.get_current_page())
raise
yield session
finally:
browser.close()
session.close()
@contextlib.contextmanager
def make_browser(ctx):
with make_session(ctx) as s:
browser = mechanicalsoup.StatefulBrowser(
soup_config={'features': 'lxml'},
raise_on_404=True,
session=s,
)
browser.set_verbose(2)
# Print a dump of browser state on errors.
try:
yield browser
except Exception as e:
ctx.log('ERROR: %s', e)
ctx.log('Browser URL: %s\n\nPage contents:\n\n%s',
browser.get_url(),
browser.get_current_page())
raise
finally:
browser.close()
class LoginError(Exception):
......@@ -71,6 +81,7 @@ def open_url(ctx, browser, url):
browser['username'] = ctx['credentials']['username']
browser['password'] = ctx['credentials']['password']
ctx.log('login page detected, logging in as %s...', ctx['credentials']['username'])
resp.close()
resp = browser.submit_selected()
page = browser.get_current_page()
......
......@@ -7,6 +7,8 @@ from ai_diagnostics.probes.pannello import probe_pannello
from ai_diagnostics.probes.webmail import probe_webmail
from ai_diagnostics.probes.imap import probe_imap
from ai_diagnostics.probes.smtp import *
from ai_diagnostics.probes.xmpp import probe_xmpp
from ai_diagnostics.probes.webdav import probe_webdav
class Credentials(dict):
......
......@@ -9,12 +9,16 @@ def probe_pannello(ctx):
ctx.log('making initial request to %s', app_url)
resp = browser.open_url(ctx, b, app_url)
if resp.status_code != 200:
raise Exception('Received HTTP status code %d, expected 200' % resp.status_code)
# See if we are successfully logged in by looking for the logout link.
ctx.log('validating login by looking for logout link')
try:
b.get_current_page().find('a', title='Logout')
except LinkNotFoundError:
raise Exception('We did not successfully log in (unable to find logout link on the application page)')
if resp.status_code != 200:
raise Exception('Received HTTP status code %d, expected 200' % resp.status_code)
# See if we are successfully logged in by looking for the logout link.
ctx.log('validating login by looking for logout link')
try:
b.get_current_page().find('a', title='Logout')
except LinkNotFoundError:
raise Exception('We did not successfully log in (unable to find logout link on the application page)')
finally:
resp.close()
from ai_diagnostics import browser
def probe_webdav(ctx):
with browser.make_session(ctx) as s:
# Send an OPTIONS request to the DAV root URL.
dav_url = ctx['url']
ctx.log('making OPTIONS request to %s', dav_url)
resp = s.options(dav_url,
auth=(
ctx['credentials']['username'],
ctx['credentials']['password'],
))
try:
if resp.status_code != 200:
raise Exception('Received HTTP status code %d, expected 200' % resp.status_code)
ctx.log('request successful')
finally:
resp.close()
......@@ -9,17 +9,26 @@ def probe_webmail(ctx):
ctx.log('making initial request to %s', app_url)
resp = browser.open_url(ctx, b, app_url)
if resp.status_code != 200:
raise Exception('Received HTTP status code %d, expected 200' % resp.status_code)
try:
if resp.status_code != 200:
raise Exception('Received HTTP status code %d, expected 200' % resp.status_code)
finally:
resp.close()
# Go to the webmail page.
ctx.log('switching over to the webmail')
resp = b.follow_link('webmail.autistici.org')
if resp.status_code != 200:
raise Exception('Could not retrieve webmail page')
try:
if resp.status_code != 200:
raise Exception('Could not retrieve webmail page')
finally:
resp.close()
# If it worked, then we can logout.
ctx.log('validating successful rendering by looking for logout link')
resp = b.follow_link('/?_task=logout')
if resp.status_code != 200:
raise Exception('Could not logout')
try:
if resp.status_code != 200:
raise Exception('Could not logout')
finally:
resp.close()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment