OLD | NEW |
(Empty) | |
| 1 # Copyright 2017 The LUCI Authors. All rights reserved. |
| 2 # Use of this source code is governed under the Apache License, Version 2.0 |
| 3 # that can be found in the LICENSE file. |
| 4 |
| 5 DEPS = [ |
| 6 'context', |
| 7 'path', |
| 8 'step', |
| 9 'url', |
| 10 ] |
| 11 |
| 12 # NOTE: These examples *probably work*. They're not run as part of regular |
| 13 # production testing, but they do access live resources that are available |
| 14 # and functional at the time of writing this. |
| 15 # |
| 16 # If this test actually fails, please adjust the URLs to achieve the same |
| 17 # effect. |
| 18 TEST_HTTPS_URL = 'https://chromium.org' |
| 19 TEST_HTTP_URL = 'http://chromium.org' |
| 20 TEST_ERROR_URL = 'http://httpstat.us/500' |
| 21 TEST_JSON_URL = 'https://chromium.googlesource.com/infra/infra?format=JSON' |
| 22 TEST_BAD_CERTS = [ |
| 23 'https://wrong.host.badssl.com/', |
| 24 'https://expired.badssl.com/', |
| 25 ] |
| 26 |
| 27 def RunSteps(api): |
| 28 assert api.url.quote('~foo') == '%7Efoo' |
| 29 assert api.url.urlencode({'foo': 'bar'}) == 'foo=bar' |
| 30 |
| 31 # get_file |
| 32 dest = api.path['start_dir'].join('download.bin') |
| 33 v = api.url.get_file(TEST_HTTPS_URL, dest, |
| 34 headers={'Authorization': 'thing'}) |
| 35 assert str(v.output) == str(dest) |
| 36 |
| 37 # get_text (HTTP) |
| 38 v = api.url.get_text(TEST_HTTP_URL) |
| 39 assert v.method == 'GET' |
| 40 assert 'The Chromium Projects' in v.output |
| 41 assert v.size == len(v.output) |
| 42 assert v.status_code == 200 |
| 43 |
| 44 # get_json (HTTPS) |
| 45 v = api.url.get_json(TEST_JSON_URL, log=True, |
| 46 strip_prefix=api.url.GERRIT_JSON_PREFIX) |
| 47 assert isinstance(v.output, dict) |
| 48 assert v.status_code == 200 |
| 49 |
| 50 # Check error conditions. |
| 51 def raises(fn, exc): |
| 52 raised = None |
| 53 try: |
| 54 fn() |
| 55 except exc as e: |
| 56 raised = e |
| 57 assert raised, 'Did not raise [%s]' % (exc,) |
| 58 return raised |
| 59 |
| 60 def test_error(): |
| 61 api.url.get_text(TEST_ERROR_URL, step_name='error', transient_retry=4) |
| 62 exc = raises(test_error, api.url.HTTPError) |
| 63 assert exc.response.error_body == '500 Internal Server Error' |
| 64 |
| 65 def test_infra_error(): |
| 66 with api.context(infra_steps=True): |
| 67 api.url.get_text(TEST_ERROR_URL, step_name='infra error', |
| 68 transient_retry=False) |
| 69 exc = raises(test_infra_error, api.url.InfraHTTPError) |
| 70 assert exc.response.error_body == '500 Internal Server Error' |
| 71 |
| 72 def test_auth_over_http(): |
| 73 api.url.get_text('http://foo/bar/text/error', |
| 74 headers={'Authorization': 'SECRET'}) |
| 75 raises(test_auth_over_http, ValueError) |
| 76 |
| 77 for bad_cert in TEST_BAD_CERTS: |
| 78 def test_bad_cert(): |
| 79 api.url.get_text(bad_cert) |
| 80 raises(test_bad_cert, api.step.StepFailure) |
| 81 |
| 82 |
| 83 def GenTests(api): |
| 84 def get(url): |
| 85 return 'GET %s' % (url,) |
| 86 |
| 87 test = ( |
| 88 api.test('basic') + |
| 89 api.url.text(get(TEST_HTTPS_URL), 'ohai') + |
| 90 api.url.text(get(TEST_HTTP_URL), '<html>The Chromium Projects</html>') + |
| 91 api.url.json(get(TEST_JSON_URL), {'is_json': True}) + |
| 92 |
| 93 api.url.error('error', 500, body='500 Internal Server Error') + |
| 94 api.url.error('infra error', 500, body='500 Internal Server Error') |
| 95 ) |
| 96 for bad_cert in TEST_BAD_CERTS: |
| 97 test += api.step_data('GET ' + bad_cert, retcode=1) |
| 98 yield test |
OLD | NEW |