Add unit tests to twine.repository

tests/test_repository.py

@@ -21,7 +21,17 @@
 from twine import utils
 
 
+@pytest.fixture()
+def default_repo():
+    return repository.Repository(
+        repository_url=utils.DEFAULT_REPOSITORY,
+        username="username",
+        password="password",
+    )
+
+
 def test_gpg_signature_structure_is_preserved():
+    """Test that gpg signature structure doesn't change"""
     data = {
         "gpg_signature": ("filename.asc", "filecontent"),
     }
@@ -31,6 +41,7 @@ def test_gpg_signature_structure_is_preserved():
 
 
 def test_content_structure_is_preserved():
+    """Test that content structure doesn't change"""
     data = {
         "content": ("filename", "filecontent"),
     }
@@ -40,6 +51,7 @@ def test_content_structure_is_preserved():
 
 
 def test_iterables_are_flattened():
+    """Test that iterables structures are changed and flattened"""
     data = {
         "platform": ["UNKNOWN"],
     }
@@ -55,47 +67,29 @@ def test_iterables_are_flattened():
     assert tuples == [("platform", "UNKNOWN"), ("platform", "ANOTHERPLATFORM")]
 
 
-def test_set_client_certificate():
-    repo = repository.Repository(
-        repository_url=utils.DEFAULT_REPOSITORY,
-        username="username",
-        password="password",
-    )
-
-    assert repo.session.cert is None
-
-    repo.set_client_certificate(("/path/to/cert", "/path/to/key"))
-    assert repo.session.cert == ("/path/to/cert", "/path/to/key")
-
+def test_set_client_certificate(default_repo):
+    """Test that setting client certificate is successful"""
+    assert default_repo.session.cert is None
 
-def test_set_certificate_authority():
-    repo = repository.Repository(
-        repository_url=utils.DEFAULT_REPOSITORY,
-        username="username",
-        password="password",
-    )
+    default_repo.set_client_certificate(("/path/to/cert", "/path/to/key"))
+    assert default_repo.session.cert == ("/path/to/cert", "/path/to/key")
 
-    assert repo.session.verify is True
 
-    repo.set_certificate_authority("/path/to/cert")
-    assert repo.session.verify == "/path/to/cert"
+def test_set_certificate_authority(default_repo):
+    """Test that setting certificate authority is successful"""
+    assert default_repo.session.verify is True
 
+    default_repo.set_certificate_authority("/path/to/cert")
+    assert default_repo.session.verify == "/path/to/cert"
 
-def test_make_user_agent_string():
-    repo = repository.Repository(
-        repository_url=utils.DEFAULT_REPOSITORY,
-        username="username",
-        password="password",
-    )
 
-    assert "User-Agent" in repo.session.headers
+def test_make_user_agent_string(default_repo):
+    """Test that dependencies are present in user agent string"""
+    assert "User-Agent" in default_repo.session.headers
 
-    user_agent = repo.session.headers["User-Agent"]
-    assert "twine/" in user_agent
-    assert "requests/" in user_agent
-    assert "requests-toolbelt/" in user_agent
-    assert "pkginfo/" in user_agent
-    assert "setuptools/" in user_agent
+    user_agent = default_repo.session.headers["User-Agent"]
+    packages = ("twine/", "requests/", "requests-toolbelt/", "pkginfo/", "setuptools/")
+    assert all(p in user_agent for p in packages)
 
 
 def response_with(**kwattrs):
@@ -107,37 +101,89 @@ def response_with(**kwattrs):
     return resp
 
 
-def test_package_is_uploaded_404s():
-    repo = repository.Repository(
-        repository_url=utils.DEFAULT_REPOSITORY,
-        username="username",
-        password="password",
+def test_package_is_uploaded_404s(default_repo):
+    """Test that a package upload fails with 404"""
+    default_repo.session = pretend.stub(
+        get=lambda url, headers: response_with(status_code=404)
     )
-    repo.session = pretend.stub(get=lambda url, headers: response_with(status_code=404))
     package = pretend.stub(safe_name="fake", metadata=pretend.stub(version="2.12.0"),)
 
-    assert repo.package_is_uploaded(package) is False
+    assert default_repo.package_is_uploaded(package) is False
 
 
-def test_package_is_uploaded_200s_with_no_releases():
-    repo = repository.Repository(
-        repository_url=utils.DEFAULT_REPOSITORY,
-        username="username",
-        password="password",
-    )
-    repo.session = pretend.stub(
+def test_package_is_uploaded_200s_with_no_releases(default_repo):
+    """Test that a package upload succeeds with 200 but has no releases"""
+    default_repo.session = pretend.stub(
         get=lambda url, headers: response_with(
             status_code=200, _content=b'{"releases": {}}', _content_consumed=True
         ),
     )
     package = pretend.stub(safe_name="fake", metadata=pretend.stub(version="2.12.0"),)
 
-    assert repo.package_is_uploaded(package) is False
+    assert default_repo.package_is_uploaded(package) is False
+
+
+def test_package_is_uploaded_with_releases_using_cache(default_repo):
+    """Test that a package upload succeeds without bypassing cache"""
+    default_repo._releases_json_data = {"fake": {"0.1": [{"filename": "fake.whl"}]}}
+    package = pretend.stub(
+        safe_name="fake", basefilename="fake.whl", metadata=pretend.stub(version="0.1"),
+    )
+
+    assert default_repo.package_is_uploaded(package) is True
+
+
+def test_package_is_uploaded_with_releases_not_using_cache(default_repo):
+    """Test that a package upload succeeds bypassing cache"""
+    default_repo.session = pretend.stub(
+        get=lambda url, headers: response_with(
+            status_code=200,
+            _content=b'{"releases": {"0.1": [{"filename": "fake.whl"}]}}',
+            _content_consumed=True,
+        ),
+    )
+    package = pretend.stub(
+        safe_name="fake", basefilename="fake.whl", metadata=pretend.stub(version="0.1"),
+    )
+
+    assert default_repo.package_is_uploaded(package, bypass_cache=True) is True
+
+
+def test_package_is_uploaded_different_filenames(default_repo):
+    """Test that a package upload fails as safe name and basefilename differ"""
+    default_repo.session = pretend.stub(
+        get=lambda url, headers: response_with(
+            status_code=200,
+            _content=b'{"releases": {"0.1": [{"filename": "fake.whl"}]}}',
+            _content_consumed=True,
+        ),
+    )
+    package = pretend.stub(
+        safe_name="fake", basefilename="foo.whl", metadata=pretend.stub(version="0.1"),
+    )
+
+    assert default_repo.package_is_uploaded(package) is False
+
+
+def test_package_is_registered(default_repo):
+    """Test that a package is registered successfully"""
+    package = pretend.stub(
+        basefilename="fake.whl", metadata_dictionary=lambda: {"name": "fake"}
+    )
+
+    resp = response_with(status_code=200)
+    setattr(resp, "raw", pretend.stub())
+    setattr(resp.raw, "close", lambda: None)
+    default_repo.session = pretend.stub(
+        post=lambda url, data, allow_redirects, headers: resp
+    )
+
+    assert default_repo.register(package)
 
 
 @pytest.mark.parametrize("disable_progress_bar", [True, False])
 def test_disable_progress_bar_is_forwarded_to_tqdm(
-    monkeypatch, tmpdir, disable_progress_bar
+    monkeypatch, tmpdir, disable_progress_bar, default_repo
 ):
     """Test whether the disable flag is passed to tqdm
         when the disable_progress_bar option is passed to the
@@ -151,14 +197,9 @@ def progressbarstub(*args, **kwargs):
         yield
 
     monkeypatch.setattr(repository, "ProgressBar", progressbarstub)
-    repo = repository.Repository(
-        repository_url=utils.DEFAULT_REPOSITORY,
-        username="username",
-        password="password",
-        disable_progress_bar=disable_progress_bar,
-    )
+    default_repo.disable_progress_bar = disable_progress_bar
 
-    repo.session = pretend.stub(
+    default_repo.session = pretend.stub(
         post=lambda url, data, allow_redirects, headers: response_with(status_code=200)
     )
 
@@ -176,7 +217,43 @@ def dictfunc():
         metadata_dictionary=dictfunc,
     )
 
-    repo.upload(package)
+    default_repo.upload(package)
+
+
+def test_upload_retry(tmpdir, default_repo, capsys):
+    """Test that retry works while uploading"""
+    default_repo.disable_progress_bar = True
+
+    default_repo.session = pretend.stub(
+        post=lambda url, data, allow_redirects, headers: response_with(
+            status_code=500, reason="Internal server error"
+        )
+    )
+
+    fakefile = tmpdir.join("fake.whl")
+    fakefile.write(".")
+
+    package = pretend.stub(
+        safe_name="fake",
+        metadata=pretend.stub(version="2.12.0"),
+        basefilename="fake.whl",
+        filename=str(fakefile),
+        metadata_dictionary=lambda: {"name": "fake"},
+    )
+
+    default_repo.upload(package)
+
+    msg = [
+        (
+            "Uploading fake.whl\n"
+            'Received "500: Internal server error" '
+            f"Package upload appears to have failed.  Retry {i} of 5"
+        )
+        for i in range(1, 6)  # default max_redirects == 5
+    ]
+
+    captured = capsys.readouterr()
+    assert captured.out == "\n".join(msg) + "\n"
 
 
 @pytest.mark.parametrize(
@@ -216,6 +293,7 @@ def dictfunc():
     ],
 )
 def test_release_urls(package_meta, repository_url, release_urls):
+    """Test that the correct release urls are read"""
     packages = [
         pretend.stub(safe_name=name, metadata=pretend.stub(version=version),)
         for name, version in package_meta
@@ -226,3 +304,16 @@ def test_release_urls(package_meta, repository_url, release_urls):
     )
 
     assert repo.release_urls(packages) == release_urls
+
+
+def test_package_is_uploaded_incorrect_repo_url():
+    """Test that a package upload fails on providing wrong repo url"""
+    repo = repository.Repository(
+        repository_url="https://bad.repo.com/legacy",
+        username="username",
+        password="password",
+    )
+
+    repo.url = "https://bad.repo.com/legacy"
+
+    assert repo.package_is_uploaded(None) is False