• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Tests for distutils.command.upload."""
2import os
3import unittest
4import unittest.mock as mock
5from urllib.request import HTTPError
6
7from test.support import run_unittest
8
9from distutils.command import upload as upload_mod
10from distutils.command.upload import upload
11from distutils.core import Distribution
12from distutils.errors import DistutilsError
13from distutils.log import ERROR, INFO
14
15from distutils.tests.test_config import PYPIRC, BasePyPIRCCommandTestCase
16
17PYPIRC_LONG_PASSWORD = """\
18[distutils]
19
20index-servers =
21    server1
22    server2
23
24[server1]
25username:me
26password:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
27
28[server2]
29username:meagain
30password: secret
31realm:acme
32repository:http://another.pypi/
33"""
34
35
36PYPIRC_NOPASSWORD = """\
37[distutils]
38
39index-servers =
40    server1
41
42[server1]
43username:me
44"""
45
46class FakeOpen(object):
47
48    def __init__(self, url, msg=None, code=None):
49        self.url = url
50        if not isinstance(url, str):
51            self.req = url
52        else:
53            self.req = None
54        self.msg = msg or 'OK'
55        self.code = code or 200
56
57    def getheader(self, name, default=None):
58        return {
59            'content-type': 'text/plain; charset=utf-8',
60            }.get(name.lower(), default)
61
62    def read(self):
63        return b'xyzzy'
64
65    def getcode(self):
66        return self.code
67
68
69class uploadTestCase(BasePyPIRCCommandTestCase):
70
71    def setUp(self):
72        super(uploadTestCase, self).setUp()
73        self.old_open = upload_mod.urlopen
74        upload_mod.urlopen = self._urlopen
75        self.last_open = None
76        self.next_msg = None
77        self.next_code = None
78
79    def tearDown(self):
80        upload_mod.urlopen = self.old_open
81        super(uploadTestCase, self).tearDown()
82
83    def _urlopen(self, url):
84        self.last_open = FakeOpen(url, msg=self.next_msg, code=self.next_code)
85        return self.last_open
86
87    def test_finalize_options(self):
88
89        # new format
90        self.write_file(self.rc, PYPIRC)
91        dist = Distribution()
92        cmd = upload(dist)
93        cmd.finalize_options()
94        for attr, waited in (('username', 'me'), ('password', 'secret'),
95                             ('realm', 'pypi'),
96                             ('repository', 'https://upload.pypi.org/legacy/')):
97            self.assertEqual(getattr(cmd, attr), waited)
98
99    def test_saved_password(self):
100        # file with no password
101        self.write_file(self.rc, PYPIRC_NOPASSWORD)
102
103        # make sure it passes
104        dist = Distribution()
105        cmd = upload(dist)
106        cmd.finalize_options()
107        self.assertEqual(cmd.password, None)
108
109        # make sure we get it as well, if another command
110        # initialized it at the dist level
111        dist.password = 'xxx'
112        cmd = upload(dist)
113        cmd.finalize_options()
114        self.assertEqual(cmd.password, 'xxx')
115
116    def test_upload(self):
117        tmp = self.mkdtemp()
118        path = os.path.join(tmp, 'xxx')
119        self.write_file(path)
120        command, pyversion, filename = 'xxx', '2.6', path
121        dist_files = [(command, pyversion, filename)]
122        self.write_file(self.rc, PYPIRC_LONG_PASSWORD)
123
124        # lets run it
125        pkg_dir, dist = self.create_dist(dist_files=dist_files)
126        cmd = upload(dist)
127        cmd.show_response = 1
128        cmd.ensure_finalized()
129        cmd.run()
130
131        # what did we send ?
132        headers = dict(self.last_open.req.headers)
133        self.assertEqual(headers['Content-length'], '2162')
134        content_type = headers['Content-type']
135        self.assertTrue(content_type.startswith('multipart/form-data'))
136        self.assertEqual(self.last_open.req.get_method(), 'POST')
137        expected_url = 'https://upload.pypi.org/legacy/'
138        self.assertEqual(self.last_open.req.get_full_url(), expected_url)
139        self.assertTrue(b'xxx' in self.last_open.req.data)
140        self.assertIn(b'protocol_version', self.last_open.req.data)
141
142        # The PyPI response body was echoed
143        results = self.get_logs(INFO)
144        self.assertEqual(results[-1], 75 * '-' + '\nxyzzy\n' + 75 * '-')
145
146    def test_upload_fails(self):
147        self.next_msg = "Not Found"
148        self.next_code = 404
149        self.assertRaises(DistutilsError, self.test_upload)
150
151    def test_wrong_exception_order(self):
152        tmp = self.mkdtemp()
153        path = os.path.join(tmp, 'xxx')
154        self.write_file(path)
155        dist_files = [('xxx', '2.6', path)]  # command, pyversion, filename
156        self.write_file(self.rc, PYPIRC_LONG_PASSWORD)
157
158        pkg_dir, dist = self.create_dist(dist_files=dist_files)
159        tests = [
160            (OSError('oserror'), 'oserror', OSError),
161            (HTTPError('url', 400, 'httperror', {}, None),
162             'Upload failed (400): httperror', DistutilsError),
163        ]
164        for exception, expected, raised_exception in tests:
165            with self.subTest(exception=type(exception).__name__):
166                with mock.patch('distutils.command.upload.urlopen',
167                                new=mock.Mock(side_effect=exception)):
168                    with self.assertRaises(raised_exception):
169                        cmd = upload(dist)
170                        cmd.ensure_finalized()
171                        cmd.run()
172                    results = self.get_logs(ERROR)
173                    self.assertIn(expected, results[-1])
174                    self.clear_logs()
175
176
177def test_suite():
178    return unittest.makeSuite(uploadTestCase)
179
180if __name__ == "__main__":
181    run_unittest(test_suite())
182