Index: recipe_modules/url/example.py |
diff --git a/recipe_modules/url/example.py b/recipe_modules/url/example.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..613b5d5ad79387e0af5cbf56d9c3704eff92c530 |
--- /dev/null |
+++ b/recipe_modules/url/example.py |
@@ -0,0 +1,98 @@ |
+# Copyright 2017 The LUCI Authors. All rights reserved. |
+# Use of this source code is governed under the Apache License, Version 2.0 |
+# that can be found in the LICENSE file. |
+ |
+DEPS = [ |
+ 'context', |
+ 'path', |
+ 'step', |
+ 'url', |
+] |
+ |
+# NOTE: These examples *probably work*. They're not run as part of regular |
+# production testing, but they do access live resources that are available |
+# and functional at the time of writing this. |
+# |
+# If this test actually fails, please adjust the URLs to achieve the same |
+# effect. |
+TEST_HTTPS_URL = 'https://chromium.org' |
+TEST_HTTP_URL = 'http://chromium.org' |
+TEST_ERROR_URL = 'http://httpstat.us/500' |
+TEST_JSON_URL = 'https://chromium.googlesource.com/infra/infra?format=JSON' |
+TEST_BAD_CERTS = [ |
+ 'https://wrong.host.badssl.com/', |
+ 'https://expired.badssl.com/', |
+] |
+ |
+def RunSteps(api): |
+ assert api.url.quote('~foo') == '%7Efoo' |
+ assert api.url.urlencode({'foo': 'bar'}) == 'foo=bar' |
+ |
+ # get_file |
+ dest = api.path['start_dir'].join('download.bin') |
+ v = api.url.get_file(TEST_HTTPS_URL, dest, |
+ headers={'Authorization': 'thing'}) |
+ assert str(v.output) == str(dest) |
+ |
+ # get_text (HTTP) |
+ v = api.url.get_text(TEST_HTTP_URL) |
+ assert v.method == 'GET' |
+ assert 'The Chromium Projects' in v.output |
+ assert v.size == len(v.output) |
+ assert v.status_code == 200 |
+ |
+ # get_json (HTTPS) |
+ v = api.url.get_json(TEST_JSON_URL, log=True, |
+ strip_prefix=api.url.GERRIT_JSON_PREFIX) |
+ assert isinstance(v.output, dict) |
+ assert v.status_code == 200 |
+ |
+ # Check error conditions. |
+ def raises(fn, exc): |
+ raised = None |
+ try: |
+ fn() |
+ except exc as e: |
+ raised = e |
+ assert raised, 'Did not raise [%s]' % (exc,) |
+ return raised |
+ |
+ def test_error(): |
+ api.url.get_text(TEST_ERROR_URL, step_name='error', transient_retry=4) |
+ exc = raises(test_error, api.url.HTTPError) |
+ assert exc.response.error_body == '500 Internal Server Error' |
+ |
+ def test_infra_error(): |
+ with api.context(infra_steps=True): |
+ api.url.get_text(TEST_ERROR_URL, step_name='infra error', |
+ transient_retry=False) |
+ exc = raises(test_infra_error, api.url.InfraHTTPError) |
+ assert exc.response.error_body == '500 Internal Server Error' |
+ |
+ def test_auth_over_http(): |
+ api.url.get_text('http://foo/bar/text/error', |
+ headers={'Authorization': 'SECRET'}) |
+ raises(test_auth_over_http, ValueError) |
+ |
+ for bad_cert in TEST_BAD_CERTS: |
+ def test_bad_cert(): |
+ api.url.get_text(bad_cert) |
+ raises(test_bad_cert, api.step.StepFailure) |
+ |
+ |
+def GenTests(api): |
+ def get(url): |
+ return 'GET %s' % (url,) |
+ |
+ test = ( |
+ api.test('basic') + |
+ api.url.text(get(TEST_HTTPS_URL), 'ohai') + |
+ api.url.text(get(TEST_HTTP_URL), '<html>The Chromium Projects</html>') + |
+ api.url.json(get(TEST_JSON_URL), {'is_json': True}) + |
+ |
+ api.url.error('error', 500, body='500 Internal Server Error') + |
+ api.url.error('infra error', 500, body='500 Internal Server Error') |
+ ) |
+ for bad_cert in TEST_BAD_CERTS: |
+ test += api.step_data('GET ' + bad_cert, retcode=1) |
+ yield test |