Fix TestKeypairAdmin tests and compute.*_keypair

The get_keypair and delete_keypair calls need to specify the user_id if
the keypair belongs to another user. This was only working by accident
because the first element returned by list_users was the current user.
However, the ordering of the user list is now different following a
change in Keystone [1].

This failing test uncovered additionally that keypair methods are not
properly setting user_id into the query parameters. Overload fetch and
delete methods to properly pass user_id as a parameter.

[1] https://review.opendev.org/c/openstack/keystone/+/938814

Closes-Bug: #2095312
Co-Authored-By: Artem Goncharov <artem.goncharov@gmail.com>

Change-Id: Ic552dee83d56278d2b866de0cb365a0c394fe26a
This commit is contained in:
Pierre Riteau
2025-01-18 10:31:13 +01:00
parent 059ae3f8f8
commit da17a2762e
4 changed files with 131 additions and 23 deletions

View File

@@ -643,13 +643,17 @@ class Proxy(proxy.Proxy):
:returns: ``None``
"""
attrs = {'user_id': user_id} if user_id else {}
self._delete(
_keypair.Keypair,
keypair,
ignore_missing=ignore_missing,
**attrs,
)
# NOTE(gtema): it is necessary to overload normal logic since query
# parameters are not properly respected in typical DELETE case
res = self._get_resource(_keypair.Keypair, keypair)
try:
delete_params = {'user_id': user_id} if user_id else {}
res.delete(self, params=delete_params)
except exceptions.NotFoundException:
if ignore_missing:
return None
raise
def get_keypair(self, keypair, user_id=None):
"""Get a single keypair
@@ -662,8 +666,14 @@ class Proxy(proxy.Proxy):
:raises: :class:`~openstack.exceptions.NotFoundException`
when no resource can be found.
"""
attrs = {'user_id': user_id} if user_id else {}
return self._get(_keypair.Keypair, keypair, **attrs)
# NOTE(gtema): it is necessary to overload normal logic since query
# parameters are not properly respected in typical fetch case
res = self._get_resource(_keypair.Keypair, keypair)
get_params = {'user_id': user_id} if user_id else {}
return res.fetch(
self, error_message=f"No Keypair found for {keypair}", **get_params
)
def find_keypair(self, name_or_id, ignore_missing=True, *, user_id=None):
"""Find a single keypair

View File

@@ -65,12 +65,12 @@ class TestKeypairAdmin(base.BaseFunctionalTest):
self._keypair = sot
def tearDown(self):
sot = self.conn.compute.delete_keypair(self._keypair)
sot = self.conn.compute.delete_keypair(self.NAME, user_id=self.USER.id)
self.assertIsNone(sot)
super().tearDown()
def test_get(self):
sot = self.conn.compute.get_keypair(self.NAME)
sot = self.conn.compute.get_keypair(self.NAME, user_id=self.USER.id)
self.assertEqual(self.NAME, sot.name)
self.assertEqual(self.NAME, sot.id)
self.assertEqual(self.USER.id, sot.user_id)

View File

@@ -12,6 +12,7 @@
import contextlib
import datetime
import fixtures
from unittest import mock
import uuid
import warnings
@@ -40,6 +41,7 @@ from openstack.compute.v2 import usage
from openstack.compute.v2 import volume_attachment
from openstack.identity.v3 import project
from openstack import proxy as proxy_base
from openstack.tests.unit import base
from openstack.tests.unit import test_proxy_base
from openstack import warnings as os_warnings
@@ -266,18 +268,32 @@ class TestKeyPair(TestComputeProxy):
self.verify_create(self.proxy.create_keypair, keypair.Keypair)
def test_keypair_delete(self):
self.verify_delete(self.proxy.delete_keypair, keypair.Keypair, False)
self._verify(
"openstack.compute.v2.keypair.Keypair.delete",
self.proxy.delete_keypair,
method_args=["value"],
expected_args=[self.proxy],
expected_kwargs={"params": {}},
)
def test_keypair_delete_ignore(self):
self.verify_delete(self.proxy.delete_keypair, keypair.Keypair, True)
self._verify(
"openstack.compute.v2.keypair.Keypair.delete",
self.proxy.delete_keypair,
method_args=["value", True],
method_kwargs={"user_id": "fake_user"},
expected_args=[self.proxy],
expected_kwargs={"params": {"user_id": "fake_user"}},
)
def test_keypair_delete_user_id(self):
self.verify_delete(
self._verify(
"openstack.compute.v2.keypair.Keypair.delete",
self.proxy.delete_keypair,
keypair.Keypair,
True,
method_kwargs={'user_id': 'fake_user'},
expected_kwargs={'user_id': 'fake_user'},
method_args=["value"],
method_kwargs={"user_id": "fake_user"},
expected_args=[self.proxy],
expected_kwargs={"params": {"user_id": "fake_user"}},
)
def test_keypair_find(self):
@@ -292,14 +308,28 @@ class TestKeyPair(TestComputeProxy):
)
def test_keypair_get(self):
self.verify_get(self.proxy.get_keypair, keypair.Keypair)
self._verify(
"openstack.compute.v2.keypair.Keypair.fetch",
self.proxy.get_keypair,
method_args=["value"],
method_kwargs={},
expected_args=[self.proxy],
expected_kwargs={
"error_message": "No Keypair found for value",
},
)
def test_keypair_get_user_id(self):
self.verify_get(
self._verify(
"openstack.compute.v2.keypair.Keypair.fetch",
self.proxy.get_keypair,
keypair.Keypair,
method_kwargs={'user_id': 'fake_user'},
expected_kwargs={'user_id': 'fake_user'},
method_args=["value"],
method_kwargs={"user_id": "fake_user"},
expected_args=[self.proxy],
expected_kwargs={
"error_message": "No Keypair found for value",
"user_id": "fake_user",
},
)
def test_keypairs(self):
@@ -314,6 +344,68 @@ class TestKeyPair(TestComputeProxy):
)
class TestKeyPairUrl(base.TestCase):
def setUp(self):
super().setUp()
self.useFixture(
fixtures.MonkeyPatch(
"openstack.utils.maximum_supported_microversion",
lambda *args, **kwargs: "2.10",
)
)
def test_keypair_find_user_id(self):
self.register_uris(
[
dict(
method="GET",
uri=self.get_mock_url(
"compute",
"public",
append=["os-keypairs", "fake_keypair"],
qs_elements=["user_id=fake_user"],
),
),
]
)
self.cloud.compute.find_keypair("fake_keypair", user_id="fake_user")
def test_keypair_get_user_id(self):
self.register_uris(
[
dict(
method="GET",
uri=self.get_mock_url(
"compute",
"public",
append=["os-keypairs", "fake_keypair"],
qs_elements=["user_id=fake_user"],
),
),
]
)
self.cloud.compute.get_keypair("fake_keypair", user_id="fake_user")
def test_keypair_delete_user_id(self):
self.register_uris(
[
dict(
method="DELETE",
uri=self.get_mock_url(
"compute",
"public",
append=["os-keypairs", "fake_keypair"],
qs_elements=["user_id=fake_user"],
),
),
]
)
self.cloud.compute.delete_keypair("fake_keypair", user_id="fake_user")
class TestAggregate(TestComputeProxy):
def test_aggregate_create(self):
self.verify_create(self.proxy.create_aggregate, aggregate.Aggregate)

View File

@@ -0,0 +1,6 @@
---
fixes:
- |
Fix the ``delete_keypair``, ``get_keypair`` and ``find_keypair`` methods
not including the optional ``user_id`` parameter in API queries.
[`bug 2095312 <https://bugs.launchpad.net/openstacksdk/+bug/2095312>`_]