Skip to content

Instantly share code, notes, and snippets.

@b10n1k
Created December 13, 2021 21:22
Show Gist options
  • Save b10n1k/a02c2b3d25369a253a43f3f06410a749 to your computer and use it in GitHub Desktop.
Save b10n1k/a02c2b3d25369a253a43f3f06410a749 to your computer and use it in GitHub Desktop.
============================= test session starts ==============================
platform linux -- Python 3.8.12, pytest-6.2.5, py-1.11.0, pluggy-1.0.0 -- /home/iob/Envs/pcwenv/bin/python
cachedir: .pytest_cache
django: settings: webui.settings (from ini)
rootdir: /home/iob/suserepos/pcw, configfile: pytest.ini
plugins: django-4.5.1, cov-2.5.0, Faker-9.9.0
collecting ... collected 3 items
tests/test_gce.py::test_parse_image_name PASSED [ 33%]
tests/test_gce.py::test_cleanup_all PASSED [ 66%]
tests/test_gce.py::test_cleanup_serviceAccounts FAILED [100%]
=================================== FAILURES ===================================
_________________________ test_cleanup_serviceAccounts _________________________
monkeypatch = <_pytest.monkeypatch.MonkeyPatch object at 0x7f013870d4c0>
def test_cleanup_serviceAccounts(monkeypatch):
service_account_older_than_max_age = (datetime.now(timezone.utc) - timedelta(hours=gce_max_service_account_age_hours+1)).isoformat()
service_account_older_than_min_age = (datetime.now(timezone.utc) - timedelta(hours=gce_min_service_account_age_hours+1)).isoformat()
fake_vault_service_accounts = FakeServiceAccounts([
FakeRequest({ # on serviceAccounts().list()
"accounts": [
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]",
"email": "[email protected]"},
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]",
"email": "[email protected]"},
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]",
"email": "[email protected]"},
]
})
])
fake_vault_service_accounts_keys = FakeServiceAccounts([
FakeRequest({ # on serviceAccounts().keys().list()
"keys": [
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]/keys/c718317203fdacf1833db761b45c0a8beb6f7bbd",
"validAfterTime": service_account_older_than_max_age},
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]/keys/c718317203fdacf1833db761b45c0a8beb6f7bbd",
"validAfterTime": service_account_older_than_max_age},
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]/keys/c718317203fdacf1833db761b45c0a8beb6f7bbd",
"validAfterTime": service_account_older_than_min_age},
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]/keys/c718317203fdacf1833db761b45c0a8beb6f7bbd",
"validAfterTime": service_account_older_than_max_age},
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]/keys/c718317203fdacf1833db761b45c0a8beb6f7bbd",
"validAfterTime": service_account_older_than_max_age},
{"name": "projects/suse-sle-qa/serviceAccounts/[email protected]/keys/c718317203fdacf1833db761b45c0a8beb6f7bbd",
"validAfterTime": service_account_older_than_min_age}
]
})
])
def mocked_iam_client():
#pass
#return MockedServiceAccountsClient(fake_vault_service_accounts, fake_vault_service_accounts_keys)
#return MockedServiceAccountsClient()
return FakeServiceAccounts()
# def mock_projects():
# return mock_serviceAccounts()
# def mock_serviceAccounts():
# return FakeServiceAccounts()
monkeypatch.setattr(PCWConfig, 'get_feature_property', mock_get_feature_property)
# iam_service.projects().serviceAccounts().list(name="projects/suse-sle-qa")
#mocked_iam_client.projects = lambda self: mock_projects
#mocked_iam_client.serviceAccounts = lambda self: mock_serviceAccounts
monkeypatch.setattr(GCE, 'iam_client', lambda self: mocked_iam_client)
mocked_iam_client.projects = lambda *args, **kwargs: mocked_iam_client
mocked_iam_client.serviceAccounts = lambda *args, **kwargs: fake_vault_service_accounts
mocked_iam_client.keys = lambda *args, **kwargs: fake_vault_service_accounts_keys
# mocked_iam_client.serviceAccounts = lambda *args, **kwargs: mocked_iam_client
# mocked_iam_client.list = lambda *args, **kwargs: {}
# #mocked_iam_client.list.execute = lambda *args, **kwargs: {}
# print("mocking keys")
# #mocked_iam_client.keys = lambda *args, **kwargs: fake_vault_service_accounts_keys
# mocked_iam_client.keys.list = lambda *args, **kwargs: {}
# #mocked_iam_client.keys.execute = lambda *args, **kwargs: {}
gce = GCE("fake")
> gce.cleanup_vaultopenqa_serviceaccounts()
tests/test_gce.py:217:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
ocw/lib/gce.py:253: in cleanup_vaultopenqa_serviceaccounts
service_accounts_emails_filtered = self.filter_email_service_accounts(service, service_accounts_emails)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <ocw.lib.gce.GCE object at 0x7f01386f4eb0>
service = <function test_cleanup_serviceAccounts.<locals>.mocked_iam_client at 0x7f013a215310>
vault_account_tuple = ['[email protected]', '[email protected]', '[email protected]']
time_limit = datetime.datetime(2021, 12, 12, 21, 22, 1, 161917)
def filter_email_service_accounts(self, service, vault_account_tuple, time_limit=24):
'''Returns a list of the emails of the Service Accounts filtered by _time_limit_
Makes a API request to the ServiceAccount to list all the keys associated with
the corresponding email of the service.
The respond returns a json in the format
_{'keys': [
{'name': 'projects/suse-sle-qa/serviceAccounts/{ACCOUNT}/keys/{KEY_HASH} # the full path for a key,
'validAfterTime': '2021-11-15T12:13:43Z', # age of the key
'validBeforeTime': '2021-12-02T12:13:43Z',
'keyAlgorithm': 'KEY_ALG_RSA_2048',
'keyOrigin': 'GOOGLE_PROVIDED',
'keyType': 'USER_MANAGED'}
]}
Using `validAfterTime` the function calculate which Service Accounts
should be deleted. The actual delete function takes the email in its
request, so the `filter_email_service_accounts` has to returned them
corelated email addresses of the `service_resp['keys']`
Parameters
----------
service : googleapiclient.discovery.Resource, required
The IAM Service Account instance
vault_account_tuple : googleapiclient.discovery.Resource, required
The IAM Service Account instance
time_limit : int
the number of hours where the Service Accounts are keep been
ignored from the cleanup job
'''
from datetime import datetime, timedelta
time_limit = datetime.now() - timedelta(hours=time_limit)
dt_frm = datetime.fromisoformat
filtered_service_accounts = []
for email in vault_account_tuple:
> service_resp = service.projects().serviceAccounts().keys().list(
name='projects/-/serviceAccounts/%s' % email).execute()
E AttributeError: 'list' object has no attribute 'list'
ocw/lib/gce.py:291: AttributeError
----------------------------- Captured stdout call -----------------------------
<class 'list'>
<class 'tests.test_gce.FakeRequest'>
<class 'dict'>
----------------------------- Captured stderr call -----------------------------
2021-12-13 21:22:01,161 ocw.lib.gce INFO [fake] ## GCE Vault key cleaning - None ##
=========================== short test summary info ============================
FAILED tests/test_gce.py::test_cleanup_serviceAccounts - AttributeError: 'lis...
========================= 1 failed, 2 passed in 0.31s ==========================
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment