import re

from test.support import os_helper

from test.support import warnings_helper

from test.support.testcase import ExtraAssertions

import time

import unittest

import urllib.request

self.assertIsNone(cookie.expires)

class LWPCookieTests(unittest.TestCase):

class LWPCookieTests(unittest.TestCase, ExtraAssertions):

# Tests taken from libwww-perl, with a few modifications and additions.

def test_netscape_example_1(self):

h = req.get_header("Cookie")

self.assertIn("PART_NUMBER=ROCKET_LAUNCHER_0001", h)

self.assertIn("CUSTOMER=WILE_E_COYOTE", h)

self.assertTrue(h.startswith("SHIPPING=FEDEX;"))

self.assertStartsWith(h,"SHIPPING=FEDEX;")

def test_netscape_example_2(self):

# Second Example transaction sequence:

from http import cookies

import pickle

from test import support

from test.support.testcase import ExtraAssertions

class CookieTests(unittest.TestCase):

class CookieTests(unittest.TestCase, ExtraAssertions):

def test_basic(self):

cases = [

C = cookies.SimpleCookie('Customer="WILE_E_COYOTE"')

C['Customer']['expires'] = 0

# can't test exact output, it always depends on current date/time

self.assertTrue(C.output().endswith('GMT'))

self.assertEndsWith(C.output(),'GMT')

# loading 'expires'

C = cookies.SimpleCookie()

from test import support

from test.support import os_helper

from test.support import socket_helper

from test.support.testcase import ExtraAssertions

support.requires_working_socket(module=True)

def create_connection(self, *pos, **kw):

return FakeSocket(*self.fake_socket_args)

class HeaderTests(TestCase):

class HeaderTests(TestCase, ExtraAssertions):

def test_auto_headers(self):

# Some headers are added automatically, but should not be added by

# .request() if they are explicitly set.

sock = FakeSocket('')

conn.sock = sock

conn.request('GET', '/foo')

self.assertTrue(sock.data.startswith(expected))

self.assertStartsWith(sock.data,expected)

expected = b'GET /foo HTTP/1.1\r\nHost: [2001:102A::]\r\n' \

b'Accept-Encoding: identity\r\n\r\n'

conn = client.HTTPConnection('[2001:102A::]')

sock = FakeSocket('')

conn.sock = sock

conn.request('GET', '/foo')

self.assertTrue(sock.data.startswith(expected))

self.assertStartsWith(sock.data,expected)

expected = b'GET /foo HTTP/1.1\r\nHost: [fe80::]\r\n' \

b'Accept-Encoding: identity\r\n\r\n'

conn = client.HTTPConnection('[fe80::%2]')

sock = FakeSocket('')

conn.sock = sock

conn.request('GET', '/foo')

self.assertTrue(sock.data.startswith(expected))

self.assertStartsWith(sock.data,expected)

expected = b'GET /foo HTTP/1.1\r\nHost: [fe80::]:81\r\n' \

b'Accept-Encoding: identity\r\n\r\n'

conn = client.HTTPConnection('[fe80::%2]:81')

sock = FakeSocket('')

conn.sock = sock

conn.request('GET', '/foo')

self.assertTrue(sock.data.startswith(expected))

self.assertStartsWith(sock.data,expected)

def test_malformed_headers_coped_with(self):

# Issue 19996

self.assertIsNotNone(resp.getheader('obs-text'))

self.assertIn('obs-text', resp.msg)

for folded in (resp.getheader('obs-fold'), resp.msg['obs-fold']):

self.assertTrue(folded.startswith('text'))

self.assertStartsWith(folded,'text')

self.assertIn(' folded with space', folded)

self.assertTrue(folded.endswith('folded with tab'))

self.assertEndsWith(folded,'folded with tab')

def test_invalid_headers(self):

conn = client.HTTPConnection('example.com')

return b''.join(body)

class BasicTest(TestCase):

class BasicTest(TestCase, ExtraAssertions):

def test_dir_with_added_behavior_on_status(self):

# see issue40084

self.assertTrue({'description', 'name', 'phrase', 'value'} <= set(dir(HTTPStatus(404))))

sock = FakeSocket(body)

conn.sock = sock

conn.request('GET', '/foo', body)

self.assertTrue(sock.data.startswith(expected), '%r != %r' %

(sock.data[:len(expected)], expected))

self.assertStartsWith(sock.data, expected)

def test_send(self):

expected = b'this is a test this is only a test'

conn.putrequest('GET', '/☃')

class ExtendedReadTest(TestCase):

class ExtendedReadTest(TestCase, ExtraAssertions):

"""

Test peek(), read1(), readline()

"""

# then unbounded peek

p2 = resp.peek()

self.assertGreaterEqual(len(p2), len(p))

self.assertTrue(p2.startswith(p))

self.assertStartsWith(p2, p)

next = resp.read(len(p2))

self.assertEqual(next, p2)

else:

line = readline(limit)

if line and line != b"foo":

if len(line) < 5:

self.assertTrue(line.endswith(b"\n"))

self.assertEndsWith(line,b"\n")

all.append(line)

if not line:

break

raise

class OfflineTest(TestCase):

class OfflineTest(TestCase, ExtraAssertions):

def test_all(self):

# Documented objects defined in the module should be in __all__

expected = {"responses"} # Allowlist documented dict() object

]

for const in expected:

with self.subTest(constant=const):

self.assertTrue(hasattr(client, const))

self.assertHasAttr(client, const)

class SourceAddressTest(TestCase):

header = self.resp.getheader('No-Such-Header',default=42)

self.assertEqual(header, 42)

class TunnelTests(TestCase):

class TunnelTests(TestCase, ExtraAssertions):

def setUp(self):

response_text = (

'HTTP/1.1 200 OK\r\n\r\n' # Reply to CONNECT

msg=f'unexpected number of send calls: {mock_send.mock_calls}')

proxy_setup_data_sent = mock_send.mock_calls[0][1][0]

self.assertIn(b'CONNECT destination.com', proxy_setup_data_sent)

self.assertTrue(

proxy_setup_data_sent.endswith(b'\r\n\r\n'),

self.assertEndsWith(proxy_setup_data_sent, b'\r\n\r\n',

msg=f'unexpected proxy data sent {proxy_setup_data_sent!r}')

def test_connect_put_request(self):

from test.support import (

is_apple, os_helper, requires_subprocess, threading_helper

)

from test.support.testcase import ExtraAssertions

support.requires_working_socket(module=True)

self.join()

class BaseTestCase(unittest.TestCase):

class BaseTestCase(unittest.TestCase, ExtraAssertions):

def setUp(self):

self._threads = threading_helper.threading_setup()

os.environ = os_helper.EnvironmentVarGuard()

self.con.request('GET', '/')

self.con.getresponse()

self.assertTrue(

err.getvalue().endswith('"GET / HTTP/1.1" 200 -\n'))

self.assertEndsWith(err.getvalue(), '"GET / HTTP/1.1" 200 -\n')

def test_err(self):

self.con = http.client.HTTPConnection(self.HOST, self.PORT)

self.con.getresponse()

lines = err.getvalue().split('\n')

self.assertTrue(lines[0].endswith('code 404, message File not found'))

self.assertTrue(lines[1].endswith('"ERROR / HTTP/1.1" 404 -'))

self.assertEndsWith(lines[0],'code 404, message File not found')

self.assertEndsWith(lines[1],'"ERROR / HTTP/1.1" 404 -')

class SimpleHTTPServerTestCase(BaseTestCase):

response = self.request(attack_url)

self.check_status_and_reason(response, HTTPStatus.MOVED_PERMANENTLY)

location = response.getheader('Location')

self.assertFalse(location.startswith('//'), msg=location)

self.assertNotStartsWith(location,'//')

self.assertEqual(location, expected_location,

msg='Expected Location header to start with a single / and '

'end with a / as this is a directory redirect.')

# We're just ensuring that the scheme and domain make it through, if

# there are or aren't multiple slashes at the start of the path that

# follows that isn't important in this Location: header.

self.assertTrue(location.startswith('https://pypi.org/'), msg=location)

self.assertStartsWith(location,'https://pypi.org/')

def test_get(self):

#constructs the path relative to the root directory of the HTTPServer

return len(self.datas)

class BaseHTTPRequestHandlerTestCase(unittest.TestCase):

class BaseHTTPRequestHandlerTestCase(unittest.TestCase, ExtraAssertions):

"""Test the functionality of the BaseHTTPServer.

Test the support for the Expect 100-continue header.

b'Host: dummy\r\n'

b'\r\n'

)

self.assertTrue(result[0].startswith(b'HTTP/1.1 400 '))

self.assertStartsWith(result[0],b'HTTP/1.1 400 ')

self.verify_expected_headers(result[1:result.index(b'\r\n')])

self.assertFalse(self.handler.get_called)