kopia lustrzana https://github.com/dgtlmoon/changedetection.io
Extend Request Parameters to add Body & Method (#325)
Simon Caron
GitHub
rodzic
e3bcd8c9bf
commit
dfcae4ee64
|
|
@ -445,6 +445,8 @@ def changedetection_app(config=None, datastore_o=None):
|
|||
'tag': form.tag.data.strip(),
|
||||
'title': form.title.data.strip(),
|
||||
'headers': form.headers.data,
|
||||
'body': form.body.data,
|
||||
'method': form.method.data,
|
||||
'fetch_backend': form.fetch_backend.data,
|
||||
'trigger_text': form.trigger_text.data,
|
||||
'notification_title': form.notification_title.data,
|
||||
|
|
|
|||
|
|
@ -131,10 +131,12 @@ class html_webdriver(Fetcher):
|
|||
class html_requests(Fetcher):
|
||||
fetcher_description = "Basic fast Plaintext/HTTP Client"
|
||||
|
||||
def run(self, url, timeout, request_headers):
|
||||
def run(self, url, timeout, request_headers, request_body, request_method):
|
||||
import requests
|
||||
|
||||
r = requests.get(url,
|
||||
r = requests.request(method=request_method,
|
||||
data=request_body,
|
||||
url=url,
|
||||
headers=request_headers,
|
||||
timeout=timeout,
|
||||
verify=False)
|
||||
|
|
|
|||
|
|
@ -80,6 +80,8 @@ class perform_site_check():
|
|||
else:
|
||||
timeout = self.datastore.data['settings']['requests']['timeout']
|
||||
url = self.datastore.get_val(uuid, 'url')
|
||||
request_body = self.datastore.get_val(uuid, 'body')
|
||||
request_method = self.datastore.get_val(uuid, 'method')
|
||||
|
||||
# Pluggable content fetcher
|
||||
prefer_backend = watch['fetch_backend']
|
||||
|
|
@ -91,7 +93,7 @@ class perform_site_check():
|
|||
|
||||
|
||||
fetcher = klass()
|
||||
fetcher.run(url, timeout, request_headers)
|
||||
fetcher.run(url, timeout, request_headers, request_body, request_method)
|
||||
# Fetching complete, now filters
|
||||
# @todo move to class / maybe inside of fetcher abstract base?
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,16 @@ import re
|
|||
|
||||
from changedetectionio.notification import default_notification_format, valid_notification_formats, default_notification_body, default_notification_title
|
||||
|
||||
valid_method = {
|
||||
'GET',
|
||||
'POST',
|
||||
'PUT',
|
||||
'PATCH',
|
||||
'DELETE',
|
||||
}
|
||||
|
||||
default_method = 'GET'
|
||||
|
||||
class StringListField(StringField):
|
||||
widget = widgets.TextArea()
|
||||
|
||||
|
|
@ -224,8 +234,22 @@ class watchForm(commonSettingsForm):
|
|||
|
||||
ignore_text = StringListField('Ignore Text', [ValidateListRegex()])
|
||||
headers = StringDictKeyValue('Request Headers')
|
||||
body = TextAreaField('Request Body', [validators.Optional()])
|
||||
method = SelectField('Request Method', choices=valid_method, default=default_method)
|
||||
trigger_text = StringListField('Trigger/wait for text', [validators.Optional(), ValidateListRegex()])
|
||||
|
||||
def validate(self, **kwargs):
|
||||
if not super().validate():
|
||||
return False
|
||||
|
||||
result = True
|
||||
|
||||
# Fail form validation when a body is set for a GET
|
||||
if self.method.data == 'GET' and self.body.data:
|
||||
self.body.errors.append('Body must be empty when Request Method is set to GET')
|
||||
result = False
|
||||
|
||||
return result
|
||||
|
||||
class globalSettingsForm(commonSettingsForm):
|
||||
|
||||
|
|
|
|||
|
|
@ -70,6 +70,8 @@ class ChangeDetectionStore:
|
|||
'previous_md5': "",
|
||||
'uuid': str(uuid_builder.uuid4()),
|
||||
'headers': {}, # Extra headers to send
|
||||
'body': None,
|
||||
'method': 'GET',
|
||||
'history': {}, # Dict of timestamp and output stripped filename
|
||||
'ignore_text': [], # List of text to ignore when calculating the comparison checksum
|
||||
# Custom notification content
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@
|
|||
<div class="tabs">
|
||||
<ul>
|
||||
<li class="tab" id="default-tab"><a href="#general">General</a></li>
|
||||
<li class="tab"><a href="#request">Request</a></li>
|
||||
<li class="tab"><a href="#notifications">Notifications</a></li>
|
||||
<li class="tab"><a href="#filters">Filters</a></li>
|
||||
<li class="tab"><a href="#triggers">Triggers</a></li>
|
||||
|
|
@ -41,14 +42,6 @@
|
|||
href="{{ url_for('settings_page', uuid=uuid) }}">default global settings</a>.</span>
|
||||
{% endif %}
|
||||
</div>
|
||||
<fieldset class="pure-group">
|
||||
{{ render_field(form.headers, rows=5, placeholder="Example
|
||||
Cookie: foobar
|
||||
User-Agent: wonderbra 1.0") }}
|
||||
<span class="pure-form-message-inline">
|
||||
Note: ONLY used by Basic fast Plaintext/HTTP Client
|
||||
</span>
|
||||
</fieldset>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.fetch_backend) }}
|
||||
<span class="pure-form-message-inline">
|
||||
|
|
@ -62,6 +55,26 @@ User-Agent: wonderbra 1.0") }}
|
|||
</fieldset>
|
||||
</div>
|
||||
|
||||
<div class="tab-pane-inner" id="request">
|
||||
<strong>Note: <i>These settings are ONLY used by Basic fast Plaintext/HTTP Client.</i></strong>
|
||||
<fieldset class="pure-group">
|
||||
{{ render_field(form.headers, rows=5, placeholder="Example
|
||||
Cookie: foobar
|
||||
User-Agent: wonderbra 1.0") }}
|
||||
</fieldset>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.body, rows=5, placeholder="Example
|
||||
{
|
||||
\"name\":\"John\",
|
||||
\"age\":30,
|
||||
\"car\":null
|
||||
}") }}
|
||||
</div>
|
||||
<div class="pure-control-group">
|
||||
{{ render_field(form.method) }}
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="tab-pane-inner" id="notifications">
|
||||
<strong>Note: <i>These settings override the global settings.</i></strong>
|
||||
<fieldset>
|
||||
|
|
|
|||
|
|
@ -1,80 +0,0 @@
|
|||
import json
|
||||
import time
|
||||
from flask import url_for
|
||||
from . util import set_original_response, set_modified_response, live_server_setup
|
||||
|
||||
# Hard to just add more live server URLs when one test is already running (I think)
|
||||
# So we add our test here (was in a different file)
|
||||
def test_headers_in_request(client, live_server):
|
||||
live_server_setup(live_server)
|
||||
|
||||
# Add our URL to the import page
|
||||
test_url = url_for('test_headers', _external=True)
|
||||
|
||||
# Add the test URL twice, we will check
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
cookie_header = '_ga=GA1.2.1022228332; cookie-preferences=analytics:accepted;'
|
||||
|
||||
|
||||
# Add some headers to a request
|
||||
res = client.post(
|
||||
url_for("edit_page", uuid="first"),
|
||||
data={
|
||||
"url": test_url,
|
||||
"tag": "",
|
||||
"fetch_backend": "html_requests",
|
||||
"headers": "xxx:ooo\ncool:yeah\r\ncookie:"+cookie_header},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Updated watch." in res.data
|
||||
|
||||
|
||||
# Give the thread time to pick up the first version
|
||||
time.sleep(5)
|
||||
|
||||
# The service should echo back the request headers
|
||||
res = client.get(
|
||||
url_for("preview_page", uuid="first"),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# Flask will convert the header key to uppercase
|
||||
assert b"Xxx:ooo" in res.data
|
||||
assert b"Cool:yeah" in res.data
|
||||
|
||||
# The test call service will return the headers as the body
|
||||
from html import escape
|
||||
assert escape(cookie_header).encode('utf-8') in res.data
|
||||
|
||||
time.sleep(5)
|
||||
|
||||
# Re #137 - Examine the JSON index file, it should have only one set of headers entered
|
||||
watches_with_headers = 0
|
||||
with open('test-datastore/url-watches.json') as f:
|
||||
app_struct = json.load(f)
|
||||
for uuid in app_struct['watching']:
|
||||
if (len(app_struct['watching'][uuid]['headers'])):
|
||||
watches_with_headers += 1
|
||||
|
||||
# Should be only one with headers set
|
||||
assert watches_with_headers==1
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,211 @@
|
|||
import json
|
||||
import time
|
||||
from flask import url_for
|
||||
from . util import set_original_response, set_modified_response, live_server_setup
|
||||
|
||||
def test_setup(live_server):
|
||||
live_server_setup(live_server)
|
||||
|
||||
# Hard to just add more live server URLs when one test is already running (I think)
|
||||
# So we add our test here (was in a different file)
|
||||
def test_headers_in_request(client, live_server):
|
||||
# Add our URL to the import page
|
||||
test_url = url_for('test_headers', _external=True)
|
||||
|
||||
# Add the test URL twice, we will check
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
cookie_header = '_ga=GA1.2.1022228332; cookie-preferences=analytics:accepted;'
|
||||
|
||||
|
||||
# Add some headers to a request
|
||||
res = client.post(
|
||||
url_for("edit_page", uuid="first"),
|
||||
data={
|
||||
"url": test_url,
|
||||
"tag": "",
|
||||
"fetch_backend": "html_requests",
|
||||
"headers": "xxx:ooo\ncool:yeah\r\ncookie:"+cookie_header},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Updated watch." in res.data
|
||||
|
||||
|
||||
# Give the thread time to pick up the first version
|
||||
time.sleep(5)
|
||||
|
||||
# The service should echo back the request headers
|
||||
res = client.get(
|
||||
url_for("preview_page", uuid="first"),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# Flask will convert the header key to uppercase
|
||||
assert b"Xxx:ooo" in res.data
|
||||
assert b"Cool:yeah" in res.data
|
||||
|
||||
# The test call service will return the headers as the body
|
||||
from html import escape
|
||||
assert escape(cookie_header).encode('utf-8') in res.data
|
||||
|
||||
time.sleep(5)
|
||||
|
||||
# Re #137 - Examine the JSON index file, it should have only one set of headers entered
|
||||
watches_with_headers = 0
|
||||
with open('test-datastore/url-watches.json') as f:
|
||||
app_struct = json.load(f)
|
||||
for uuid in app_struct['watching']:
|
||||
if (len(app_struct['watching'][uuid]['headers'])):
|
||||
watches_with_headers += 1
|
||||
|
||||
# Should be only one with headers set
|
||||
assert watches_with_headers==1
|
||||
|
||||
def test_body_in_request(client, live_server):
|
||||
# Add our URL to the import page
|
||||
test_url = url_for('test_body', _external=True)
|
||||
|
||||
# Add the test URL twice, we will check
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
body_value = 'Test Body Value'
|
||||
|
||||
# Attempt to add a body with a GET method
|
||||
res = client.post(
|
||||
url_for("edit_page", uuid="first"),
|
||||
data={
|
||||
"url": test_url,
|
||||
"tag": "",
|
||||
"method": "GET",
|
||||
"fetch_backend": "html_requests",
|
||||
"body": "invalid"},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Body must be empty when Request Method is set to GET" in res.data
|
||||
|
||||
# Add a properly formatted body with a proper method
|
||||
res = client.post(
|
||||
url_for("edit_page", uuid="first"),
|
||||
data={
|
||||
"url": test_url,
|
||||
"tag": "",
|
||||
"method": "POST",
|
||||
"fetch_backend": "html_requests",
|
||||
"body": body_value},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Updated watch." in res.data
|
||||
|
||||
# Give the thread time to pick up the first version
|
||||
time.sleep(5)
|
||||
|
||||
# The service should echo back the body
|
||||
res = client.get(
|
||||
url_for("preview_page", uuid="first"),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# Check if body returned contains the specified data
|
||||
assert str.encode(body_value) in res.data
|
||||
|
||||
watches_with_body = 0
|
||||
with open('test-datastore/url-watches.json') as f:
|
||||
app_struct = json.load(f)
|
||||
for uuid in app_struct['watching']:
|
||||
if app_struct['watching'][uuid]['body']==body_value:
|
||||
watches_with_body += 1
|
||||
|
||||
# Should be only one with body set
|
||||
assert watches_with_body==1
|
||||
|
||||
def test_method_in_request(client, live_server):
|
||||
# Add our URL to the import page
|
||||
test_url = url_for('test_method', _external=True)
|
||||
|
||||
# Add the test URL twice, we will check
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
res = client.post(
|
||||
url_for("import_page"),
|
||||
data={"urls": test_url},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"1 Imported" in res.data
|
||||
|
||||
# Attempt to add a method which is not valid
|
||||
res = client.post(
|
||||
url_for("edit_page", uuid="first"),
|
||||
data={
|
||||
"url": test_url,
|
||||
"tag": "",
|
||||
"fetch_backend": "html_requests",
|
||||
"method": "invalid"},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Not a valid choice" in res.data
|
||||
|
||||
# Add a properly formatted body
|
||||
res = client.post(
|
||||
url_for("edit_page", uuid="first"),
|
||||
data={
|
||||
"url": test_url,
|
||||
"tag": "",
|
||||
"fetch_backend": "html_requests",
|
||||
"method": "PATCH"},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Updated watch." in res.data
|
||||
|
||||
# Give the thread time to pick up the first version
|
||||
time.sleep(5)
|
||||
|
||||
# The service should echo back the request verb
|
||||
res = client.get(
|
||||
url_for("preview_page", uuid="first"),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
# The test call service will return the verb as the body
|
||||
assert b"PATCH" in res.data
|
||||
|
||||
time.sleep(5)
|
||||
|
||||
watches_with_method = 0
|
||||
with open('test-datastore/url-watches.json') as f:
|
||||
app_struct = json.load(f)
|
||||
for uuid in app_struct['watching']:
|
||||
if app_struct['watching'][uuid]['method'] == 'PATCH':
|
||||
watches_with_method += 1
|
||||
|
||||
# Should be only one with method set to PATCH
|
||||
assert watches_with_method == 1
|
||||
|
||||
|
|
@ -56,6 +56,21 @@ def live_server_setup(live_server):
|
|||
|
||||
return "\n".join(output)
|
||||
|
||||
# Just return the body in the request
|
||||
@live_server.app.route('/test-body', methods=['POST', 'GET'])
|
||||
def test_body():
|
||||
|
||||
from flask import request
|
||||
|
||||
return request.data
|
||||
|
||||
# Just return the verb in the request
|
||||
@live_server.app.route('/test-method', methods=['POST', 'GET', 'PATCH'])
|
||||
def test_method():
|
||||
|
||||
from flask import request
|
||||
|
||||
return request.method
|
||||
|
||||
# Where we POST to as a notification
|
||||
@live_server.app.route('/test_notification_endpoint', methods=['POST', 'GET'])
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue