WIP Fix tests for new bindings version

pulpcore/pytest_plugin.py

@@ -585,7 +585,7 @@ def _random_artifact_factory(size=32, pulp_domain=None):
             kwargs["pulp_domain"] = pulp_domain
         temp_file = tmp_path / str(uuid.uuid4())
         temp_file.write_bytes(os.urandom(size))
-        return pulpcore_bindings.ArtifactsApi.create(temp_file, **kwargs)
+        return pulpcore_bindings.ArtifactsApi.create(str(temp_file), **kwargs)
 
     return _random_artifact_factory
 

pulpcore/tests/functional/api/test_auth.py

@@ -4,11 +4,10 @@
 <https://docs.pulpproject.org/restapi.html#section/Authentication>`_.
 """
 
-import pytest
 import json
-
 from base64 import b64encode
-from pulpcore.client.pulpcore import ApiException
+
+import pytest
 
 from pulpcore.app import settings
 
@@ -20,9 +19,9 @@ def test_base_auth_success(pulpcore_bindings, pulp_admin_user):
     Assert that a response indicating success is returned.
     """
     with pulp_admin_user:
-        response, status, headers = pulpcore_bindings.ArtifactsApi.list_with_http_info()
-    assert status == 200
-    assert headers["Content-Type"] == "application/json"
+        response = pulpcore_bindings.ArtifactsApi.list_with_http_info()
+    assert response.status_code == 200
+    assert response.headers["Content-Type"] == "application/json"
     # Maybe test correlation ID as well?
 
 
@@ -33,7 +32,7 @@ def test_base_auth_failure(pulpcore_bindings, invalid_user):
     Assert that a response indicating failure is returned.
     """
     with invalid_user:
-        with pytest.raises(ApiException) as e:
+        with pytest.raises(pulpcore_bindings.ApiException) as e:
             pulpcore_bindings.ArtifactsApi.list()
 
     assert e.value.status == 401
@@ -50,7 +49,7 @@ def test_base_auth_required(pulpcore_bindings, anonymous_user):
     Assert that a response indicating failure is returned.
     """
     with anonymous_user:
-        with pytest.raises(ApiException) as e:
+        with pytest.raises(pulpcore_bindings.ApiException) as e:
             pulpcore_bindings.ArtifactsApi.list()
 
     assert e.value.status == 401
@@ -79,8 +78,8 @@ def test_jq_header_remote_auth(pulpcore_bindings, anonymous_user):
         encoded_header = b64encode(bytes(header_content, "ascii"))
 
         pulpcore_bindings.ArtifactsApi.api_client.default_headers["x-rh-identity"] = encoded_header
-        _, status, _ = pulpcore_bindings.ArtifactsApi.list_with_http_info()
-    assert status == 200
+        response = pulpcore_bindings.ArtifactsApi.list_with_http_info()
+    assert response.status_code == 200
 
 
 @pytest.mark.parallel
@@ -106,7 +105,7 @@ def test_jq_header_remote_auth_denied_by_wrong_header(pulpcore_bindings, anonymo
             encoded_header
         )
 
-        with pytest.raises(ApiException) as exception:
+        with pytest.raises(pulpcore_bindings.ApiException) as exception:
             pulpcore_bindings.ArtifactsApi.list()
 
     assert exception.value.status == 401
@@ -127,7 +126,7 @@ def test_jq_header_remote_auth_denied_by_wrong_content(pulpcore_bindings, anonym
 
         pulpcore_bindings.ArtifactsApi.api_client.default_headers["x-rh-identity"] = encoded_header
 
-        with pytest.raises(ApiException) as exception:
+        with pytest.raises(pulpcore_bindings.ApiException) as exception:
             pulpcore_bindings.ArtifactsApi.list()
 
     assert exception.value.status == 401

pulpcore/tests/functional/api/test_correlation_id.py

@@ -1,7 +1,6 @@
 def test_correlation_id(cid, pulpcore_bindings, monitor_task):
     """Test that a correlation can be passed as a header and logged."""
-    response, status, headers = pulpcore_bindings.OrphansCleanupApi.cleanup_with_http_info({})
-    monitor_task(response.task)
-    task = pulpcore_bindings.TasksApi.read(response.task)
-    assert headers["Correlation-ID"] == cid
+    response = pulpcore_bindings.OrphansCleanupApi.cleanup_with_http_info({})
+    task = monitor_task(response.data.task)
+    assert response.headers["Correlation-ID"] == cid
     assert task.logging_cid == cid

pulpcore/tests/functional/api/test_crd_artifacts.py

@@ -7,7 +7,6 @@
 import pytest
 
 from django.conf import settings
-from pulpcore.client.pulpcore import ApiException
 
 
 @pytest.fixture
@@ -21,12 +20,12 @@ def pulpcore_random_file(tmp_path):
     return {"name": name, "size": 1024, "digest": digest}
 
 
-def _do_upload_valid_attrs(artifact_api, file, data):
+def _do_upload_valid_attrs(pulpcore_bindings, file, data):
     """Upload a file with the given attributes."""
-    artifact = artifact_api.create(file, **data)
+    artifact = pulpcore_bindings.ArtifactsApi.create(str(file), **data)
     # assumes ALLOWED_CONTENT_CHECKSUMS does NOT contain "md5"
     assert artifact.md5 is None, "MD5 {}".format(artifact.md5)
-    read_artifact = artifact_api.read(artifact.pulp_href)
+    read_artifact = pulpcore_bindings.ArtifactsApi.read(artifact.pulp_href)
     # assumes ALLOWED_CONTENT_CHECKSUMS does NOT contain "md5"
     assert read_artifact.md5 is None
     for key, val in artifact.to_dict().items():
@@ -54,9 +53,7 @@ def test_upload_valid_attrs(pulpcore_bindings, pulpcore_random_file, monitor_tas
             monitor_task(
                 pulpcore_bindings.OrphansCleanupApi.cleanup({"orphan_protection_time": 0}).task
             )
-            _do_upload_valid_attrs(
-                pulpcore_bindings.ArtifactsApi, pulpcore_random_file["name"], data
-            )
+            _do_upload_valid_attrs(pulpcore_bindings, pulpcore_random_file["name"], data)
 
 
 def test_upload_empty_file(pulpcore_bindings, tmp_path, monitor_task):
@@ -79,7 +76,7 @@ def test_upload_empty_file(pulpcore_bindings, tmp_path, monitor_task):
                 pulpcore_bindings.OrphansCleanupApi.cleanup({"orphan_protection_time": 0}).task
             )
             data = {key: file_attrs[key] for key in keys}
-            _do_upload_valid_attrs(pulpcore_bindings.ArtifactsApi, file, data)
+            _do_upload_valid_attrs(pulpcore_bindings, file, data)
 
 
 @pytest.mark.parallel
@@ -98,16 +95,16 @@ def test_upload_invalid_attrs(pulpcore_bindings, pulpcore_random_file):
     for i in range(1, len(file_attrs) + 1):
         for keys in itertools.combinations(file_attrs, i):
             data = {key: file_attrs[key] for key in keys}
-            _do_upload_invalid_attrs(pulpcore_bindings.ArtifactsApi, pulpcore_random_file, data)
+            _do_upload_invalid_attrs(pulpcore_bindings, pulpcore_random_file, data)
 
 
-def _do_upload_invalid_attrs(artifact_api, file, data):
+def _do_upload_invalid_attrs(pulpcore_bindings, file, data):
     """Upload a file with the given attributes."""
-    with pytest.raises(ApiException) as e:
-        artifact_api.create(file["name"], **data)
+    with pytest.raises(pulpcore_bindings.ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.create(str(file["name"]), **data)
 
     assert e.value.status == 400
-    artifacts = artifact_api.list()
+    artifacts = pulpcore_bindings.ArtifactsApi.list()
     for artifact in artifacts.results:
         assert artifact.sha256 != file["digest"]
 
@@ -119,8 +116,8 @@ def test_upload_md5(pulpcore_bindings, pulpcore_random_file):
     Assumes ALLOWED_CONTENT_CHECKSUMS does NOT contain ``md5``
     """
     file_attrs = {"md5": str(uuid.uuid4()), "size": pulpcore_random_file["size"]}
-    with pytest.raises(ApiException) as e:
-        pulpcore_bindings.ArtifactsApi.create(pulpcore_random_file["name"], **file_attrs)
+    with pytest.raises(pulpcore_bindings.ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.create(str(pulpcore_random_file["name"]), **file_attrs)
 
     assert e.value.status == 400
 
@@ -141,7 +138,7 @@ def test_upload_mixed_attrs(pulpcore_bindings, pulpcore_random_file):
         {"sha256": str(uuid.uuid4()), "size": pulpcore_random_file["size"]},
     )
     for data in invalid_data:
-        _do_upload_invalid_attrs(pulpcore_bindings.ArtifactsApi, pulpcore_random_file, data)
+        _do_upload_invalid_attrs(pulpcore_bindings, pulpcore_random_file, data)
 
 
 @pytest.mark.parallel
@@ -152,19 +149,19 @@ def test_delete_artifact(pulpcore_bindings, pulpcore_random_file, gen_user):
         pytest.skip("this test only works for filesystem storage")
     media_root = settings.MEDIA_ROOT
 
-    artifact = pulpcore_bindings.ArtifactsApi.create(pulpcore_random_file["name"])
+    artifact = pulpcore_bindings.ArtifactsApi.create(str(pulpcore_random_file["name"]))
     path_to_file = os.path.join(media_root, artifact.file)
     file_exists = os.path.exists(path_to_file)
     assert file_exists
 
     # try to delete as a regular (non-admin) user
     regular_user = gen_user()
-    with regular_user, pytest.raises(ApiException) as e:
+    with regular_user, pytest.raises(pulpcore_bindings.ApiException) as e:
         pulpcore_bindings.ArtifactsApi.delete(artifact.pulp_href)
     assert e.value.status == 403
 
     # destroy artifact api is not allowed, even for admins
-    with pytest.raises(ApiException) as e:
+    with pytest.raises(pulpcore_bindings.ApiException) as e:
         pulpcore_bindings.ArtifactsApi.delete(artifact.pulp_href)
     assert e.value.status == 403
 
@@ -173,8 +170,8 @@ def test_delete_artifact(pulpcore_bindings, pulpcore_random_file, gen_user):
 def test_upload_artifact_as_a_regular_user(pulpcore_bindings, gen_user, pulpcore_random_file):
     """Regular users do not have permission to upload artifacts."""
     regular_user = gen_user()
-    with regular_user, pytest.raises(ApiException) as e:
-        pulpcore_bindings.ArtifactsApi.create(pulpcore_random_file["name"])
+    with regular_user, pytest.raises(pulpcore_bindings.ApiException) as e:
+        pulpcore_bindings.ArtifactsApi.create(str(pulpcore_random_file["name"]))
     assert e.value.status == 403
 
 
@@ -184,14 +181,14 @@ def test_list_and_retrieve_artifact_as_a_regular_user(
 ):
     """Regular users are not allowed to list and/or retrieve artifacts."""
     regular_user = gen_user()
-    artifact = pulpcore_bindings.ArtifactsApi.create(pulpcore_random_file["name"])
+    artifact = pulpcore_bindings.ArtifactsApi.create(str(pulpcore_random_file["name"]))
 
     # check if list is not allowed
-    with regular_user, pytest.raises(ApiException) as e:
+    with regular_user, pytest.raises(pulpcore_bindings.ApiException) as e:
         pulpcore_bindings.ArtifactsApi.list()
     assert e.value.status == 403
 
     # check if retrieve is also not allowed
-    with regular_user, pytest.raises(ApiException) as e:
+    with regular_user, pytest.raises(pulpcore_bindings.ApiException) as e:
         pulpcore_bindings.ArtifactsApi.read(artifact.pulp_href)
     assert e.value.status == 403

pulpcore/tests/functional/api/using_plugin/test_crud_repos.py

@@ -262,25 +262,11 @@ def _compare_results(data, received):
     new_remote = file_bindings.RemotesFileApi.read(remote.pulp_href)
     _compare_results(data, new_remote)
 
-    # Test invalid float < 0
-    data = {
-        "total_timeout": -1.0,
-    }
-    with pytest.raises(ApiException):
-        file_bindings.RemotesFileApi.partial_update(remote.pulp_href, data)
-
-    # Test invalid non-float
-    data = {
-        "connect_timeout": "abc",
-    }
-    with pytest.raises(ApiException):
-        file_bindings.RemotesFileApi.partial_update(remote.pulp_href, data)
-
     # Test reset to empty
     data = {
-        "total_timeout": False,
+        "total_timeout": None,
         "connect_timeout": None,
-        "sock_connect_timeout": False,
+        "sock_connect_timeout": None,
         "sock_read_timeout": None,
     }
     response = file_bindings.RemotesFileApi.partial_update(remote.pulp_href, data)
@@ -289,12 +275,6 @@ def _compare_results(data, received):
     _compare_results(data, new_remote)
 
     # Test that headers value must be a list of dicts
-    data = {"headers": {"Connection": "keep-alive"}}
-    with pytest.raises(ApiException):
-        file_bindings.RemotesFileApi.partial_update(remote.pulp_href, data)
-    data = {"headers": [1, 2, 3]}
-    with pytest.raises(ApiException):
-        file_bindings.RemotesFileApi.partial_update(remote.pulp_href, data)
     data = {"headers": [{"Connection": "keep-alive"}]}
     response = file_bindings.RemotesFileApi.partial_update(remote.pulp_href, data)
     monitor_task(response.task)

pulpcore/tests/functional/api/using_plugin/test_distributions.py

@@ -4,11 +4,6 @@
 import json
 from uuid import uuid4
 
-from pulpcore.client.pulp_file import (
-    RepositorySyncURL,
-    FileFileDistribution,
-    FileFilePublication,
-)
 from pulpcore.client.pulp_file.exceptions import ApiException
 
 
@@ -24,7 +19,7 @@ def test_crud_publication_distribution(
 ):
     # Create a remote and sync from it to create the first repository version
     remote = file_remote_ssl_factory(manifest_path=basic_manifest_path, policy="on_demand")
-    body = RepositorySyncURL(remote=remote.pulp_href)
+    body = file_bindings.RepositorySyncURL(remote=remote.pulp_href)
     monitor_task(file_bindings.RepositoriesFileApi.sync(file_repo.pulp_href, body).task)
 
     # Remove content to create two more repository versions
@@ -44,7 +39,7 @@ def test_crud_publication_distribution(
 
     # Create a publication from version 2
     repo_versions = file_bindings.RepositoriesFileVersionsApi.list(file_repo.pulp_href).results
-    publish_data = FileFilePublication(repository_version=repo_versions[2].pulp_href)
+    publish_data = file_bindings.FileFilePublication(repository_version=repo_versions[2].pulp_href)
     publication = gen_object_with_cleanup(file_bindings.PublicationsFileApi, publish_data)
     distribution_data = {
         "publication": publication.pulp_href,
@@ -90,7 +85,7 @@ def test_crud_publication_distribution(
     new_name = str(uuid4())
     distribution.name = new_name
     monitor_task(
-        file_bindings.DistributionsFileApi.update(distribution.pulp_href, distribution).task
+        file_bindings.DistributionsFileApi.update(distribution.pulp_href, distribution.dict()).task
     )
     distribution = file_bindings.DistributionsFileApi.read(distribution.pulp_href)
     assert distribution.name == new_name
@@ -99,7 +94,7 @@ def test_crud_publication_distribution(
     new_base_path = str(uuid4())
     distribution.base_path = new_base_path
     monitor_task(
-        file_bindings.DistributionsFileApi.update(distribution.pulp_href, distribution).task
+        file_bindings.DistributionsFileApi.update(distribution.pulp_href, distribution.dict()).task
     )
     distribution = file_bindings.DistributionsFileApi.read(distribution.pulp_href)
     assert distribution.base_path == new_base_path
@@ -180,7 +175,7 @@ def generate_repo_with_content():
         repo = file_repository_factory()
         repo_manifest_path = write_3_iso_file_fixture_data_factory(str(uuid4()))
         remote = file_remote_factory(manifest_path=repo_manifest_path, policy="on_demand")
-        body = RepositorySyncURL(remote=remote.pulp_href)
+        body = file_bindings.RepositorySyncURL(remote=remote.pulp_href)
         task_response = file_bindings.RepositoriesFileApi.sync(repo.pulp_href, body).task
         version_href = monitor_task(task_response).created_resources[0]
         content = file_bindings.ContentFilesApi.list(repository_version_added=version_href).results[
@@ -190,11 +185,11 @@ def generate_repo_with_content():
 
     repo1, content1 = generate_repo_with_content()
 
-    publish_data = FileFilePublication(repository=repo1.pulp_href)
+    publish_data = file_bindings.FileFilePublication(repository=repo1.pulp_href)
     publication = gen_object_with_cleanup(file_bindings.PublicationsFileApi, publish_data)
 
     # test if a publication attached to a distribution exposes the published content
-    data = FileFileDistribution(
+    data = file_bindings.FileFileDistribution(
         name=str(uuid4()), base_path=str(uuid4()), publication=publication.pulp_href
     )
     distribution_pub1 = gen_object_with_cleanup(file_bindings.DistributionsFileApi, data)
@@ -203,9 +198,9 @@ def generate_repo_with_content():
     assert [distribution_pub1] == results
 
     # test if a publication pointing to repository version no. 0 does not expose any content
-    publish_data = FileFilePublication(repository_version=repo1.versions_href + "0/")
+    publish_data = file_bindings.FileFilePublication(repository_version=repo1.versions_href + "0/")
     publication_version_0 = gen_object_with_cleanup(file_bindings.PublicationsFileApi, publish_data)
-    data = FileFileDistribution(
+    data = file_bindings.FileFileDistribution(
         name=str(uuid4()), base_path=str(uuid4()), publication=publication_version_0.pulp_href
     )
     gen_object_with_cleanup(file_bindings.DistributionsFileApi, data)
@@ -215,7 +210,7 @@ def generate_repo_with_content():
 
     # test if a repository assigned to a distribution exposes the content available in the latest
     # publication for that repository's versions
-    data = FileFileDistribution(
+    data = file_bindings.FileFileDistribution(
         name=str(uuid4()), base_path=str(uuid4()), repository=repo1.pulp_href
     )
     distribution_repopub = gen_object_with_cleanup(file_bindings.DistributionsFileApi, data)
@@ -237,20 +232,20 @@ def generate_repo_with_content():
     monitor_task(response.task)
     assert [] == file_bindings.DistributionsFileApi.list(with_content=content2.pulp_href).results
 
-    publish_data = FileFilePublication(repository=repo1.pulp_href)
+    publish_data = file_bindings.FileFilePublication(repository=repo1.pulp_href)
     new_publication = gen_object_with_cleanup(file_bindings.PublicationsFileApi, publish_data)
 
     # test later (20 lines below) if the publication now exposes the recently added content in the
     # affected distributions (i.e., the distribution with the reference to a repository and the
     # new one)
-    data = FileFileDistribution(
+    data = file_bindings.FileFileDistribution(
         name="pub3", base_path="pub3", publication=new_publication.pulp_href
     )
     distribution_pub3 = gen_object_with_cleanup(file_bindings.DistributionsFileApi, data)
 
     # test if a repository without any attached publication does not expose any kind of content
     # to a user even though the content is still present in the latest repository version
-    data = FileFileDistribution(
+    data = file_bindings.FileFileDistribution(
         name=str(uuid4()), base_path=str(uuid4()), repository=repo2.pulp_href
     )
     distribution_repo_only = gen_object_with_cleanup(file_bindings.DistributionsFileApi, data)
@@ -262,7 +257,7 @@ def generate_repo_with_content():
     assert {distribution_pub3.pulp_href, distribution_repopub.pulp_href} == results
 
     # create a publication to see whether the content of the second repository is now served or not
-    publish_data = FileFilePublication(repository=repo2.pulp_href)
+    publish_data = file_bindings.FileFilePublication(repository=repo2.pulp_href)
     gen_object_with_cleanup(file_bindings.PublicationsFileApi, publish_data)
 
     results = set(