#!/usr/bin/env python # # Copyright (C) 2009 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # This module is used for version 2 of the Google Data APIs. __author__ = 'j.s@google.com (Jeff Scudder)' import unittest import atom.http_core import StringIO class UriTest(unittest.TestCase): def test_parse_uri(self): uri = atom.http_core.parse_uri('http://www.google.com/test?q=foo&z=bar') self.assert_(uri.scheme == 'http') self.assert_(uri.host == 'www.google.com') self.assert_(uri.port is None) self.assert_(uri.path == '/test') self.assert_(uri.query == {'z':'bar', 'q':'foo'}) def test_static_parse_uri(self): uri = atom.http_core.Uri.parse_uri('http://test.com/?token=foo&x=1') self.assertEqual(uri.scheme, 'http') self.assertEqual(uri.host, 'test.com') self.assert_(uri.port is None) self.assertEqual(uri.query, {'token':'foo', 'x':'1'}) def test_modify_request_no_request(self): uri = atom.http_core.parse_uri('http://www.google.com/test?q=foo&z=bar') request = uri.modify_request() self.assert_(request.uri.scheme == 'http') self.assert_(request.uri.host == 'www.google.com') # If no port was provided, the HttpClient is responsible for determining # the default. self.assert_(request.uri.port is None) self.assert_(request.uri.path.startswith('/test')) self.assertEqual(request.uri.query, {'z': 'bar', 'q': 'foo'}) self.assert_(request.method is None) self.assert_(request.headers == {}) self.assert_(request._body_parts == []) def test_modify_request_http_with_set_port(self): request = atom.http_core.HttpRequest(uri=atom.http_core.Uri(port=8080), method='POST') request.add_body_part('hello', 'text/plain') uri = atom.http_core.parse_uri('//example.com/greet') self.assert_(uri.query == {}) self.assert_(uri._get_relative_path() == '/greet') self.assert_(uri.host == 'example.com') self.assert_(uri.port is None) uri.ModifyRequest(request) self.assert_(request.uri.host == 'example.com') # If no scheme was provided, the URI will not add one, but the HttpClient # should assume the request is HTTP. self.assert_(request.uri.scheme is None) self.assert_(request.uri.port == 8080) self.assert_(request.uri.path == '/greet') self.assert_(request.method == 'POST') self.assert_(request.headers['Content-Type'] == 'text/plain') def test_modify_request_use_default_ssl_port(self): request = atom.http_core.HttpRequest( uri=atom.http_core.Uri(scheme='https'), method='PUT') request.add_body_part('hello', 'text/plain') uri = atom.http_core.parse_uri('/greet') uri.modify_request(request) self.assert_(request.uri.host is None) self.assert_(request.uri.scheme == 'https') # If no port was provided, leave the port as None, it is up to the # HttpClient to set the correct default port. self.assert_(request.uri.port is None) self.assert_(request.uri.path == '/greet') self.assert_(request.method == 'PUT') self.assert_(request.headers['Content-Type'] == 'text/plain') self.assert_(len(request._body_parts) == 1) self.assert_(request._body_parts[0] == 'hello') def test_to_string(self): uri = atom.http_core.Uri(host='www.google.com', query={'q':'sippycode'}) uri_string = uri._to_string() self.assert_(uri_string == 'http://www.google.com/?q=sippycode') class HttpRequestTest(unittest.TestCase): def test_request_with_one_body_part(self): request = atom.http_core.HttpRequest() self.assert_(len(request._body_parts) == 0) self.assert_('Content-Length' not in request.headers) self.assert_(not 'Content-Type' in request.headers) self.assert_(not 'Content-Length' in request.headers) request.add_body_part('this is a test', 'text/plain') self.assert_(len(request._body_parts) == 1) self.assert_(request.headers['Content-Type'] == 'text/plain') self.assert_(request._body_parts[0] == 'this is a test') self.assert_(request.headers['Content-Length'] == str(len( 'this is a test'))) def test_add_file_without_size(self): virtual_file = StringIO.StringIO('this is a test') request = atom.http_core.HttpRequest() try: request.add_body_part(virtual_file, 'text/plain') self.fail('We should have gotten an UnknownSize error.') except atom.http_core.UnknownSize: pass request.add_body_part(virtual_file, 'text/plain', len('this is a test')) self.assert_(len(request._body_parts) == 1) self.assert_(request.headers['Content-Type'] == 'text/plain') self.assert_(request._body_parts[0].read() == 'this is a test') self.assert_(request.headers['Content-Length'] == str(len( 'this is a test'))) def test_copy(self): request = atom.http_core.HttpRequest( uri=atom.http_core.Uri(scheme='https', host='www.google.com'), method='POST', headers={'test':'1', 'ok':'yes'}) request.add_body_part('body1', 'text/plain') request.add_body_part('body2', 'text/html') copied = request._copy() self.assert_(request.uri.scheme == copied.uri.scheme) self.assert_(request.uri.host == copied.uri.host) self.assert_(request.method == copied.method) self.assert_(request.uri.path == copied.uri.path) self.assert_(request.headers == copied.headers) self.assert_(request._body_parts == copied._body_parts) copied.headers['test'] = '2' copied._body_parts[1] = 'body3' self.assert_(request.headers != copied.headers) self.assert_(request._body_parts != copied._body_parts) def suite(): return unittest.TestSuite((unittest.makeSuite(UriTest,'test'), unittest.makeSuite(HttpRequestTest,'test'))) if __name__ == '__main__': unittest.main()