opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix SA20 syntax in test_vault_file_source, refactor

Browse Source
This commit is contained in:
John Davis
2023-07-21 13:04:23 -04:00
parent e54a2ec897
commit bce033e3fd
1 changed files with 18 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/integration/test_vault_file_source.py
+18 -32
View File
@@ -1,6 +1,8 @@
import os
import tempfile
from sqlalchemy import select
from galaxy.security.vault import UserVaultWrapper
from galaxy_test.base import api_asserts
from galaxy_test.base.populators import DatasetPopulator
@@ -31,14 +33,11 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
This file source performs a user vault lookup. The secret stored for the first user is a
valid path and should succeed, while the second user's stored secret should fail.
"""
app = self._app
with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
app = self._app
user = (
app.model.context.query(app.model.User)
.filter(app.model.User.email == self.USER_1_APP_VAULT_ENTRY)
.first()
)
user_vault = UserVaultWrapper(self._app.vault, user)
user = self._get_user_by_email(self.USER_1_APP_VAULT_ENTRY)
user_vault = UserVaultWrapper(app.vault, user)
# use a valid symlink path so the posix list succeeds
user_vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
@@ -49,13 +48,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
print(remote_files)
with self._different_user(email=self.USER_2_APP_VAULT_ENTRY):
app = self._app
user = (
app.model.context.query(app.model.User)
.filter(app.model.User.email == self.USER_2_APP_VAULT_ENTRY)
.first()
)
user_vault = UserVaultWrapper(self._app.vault, user)
user = self._get_user_by_email(self.USER_2_APP_VAULT_ENTRY)
user_vault = UserVaultWrapper(app.vault, user)
# use an invalid symlink path so the posix list fails
user_vault.write_secret("posix/root_path", "/invalid/root")
@@ -75,12 +69,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
app.vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
user = (
app.model.context.query(app.model.User)
.filter(app.model.User.email == self.USER_1_APP_VAULT_ENTRY)
.first()
)
user_vault = UserVaultWrapper(self._app.vault, user)
user = self._get_user_by_email(self.USER_1_APP_VAULT_ENTRY)
user_vault = UserVaultWrapper(app.vault, user)
# use a valid symlink path so the posix list succeeds
user_vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
@@ -91,12 +81,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
print(remote_files)
with self._different_user(email=self.USER_2_APP_VAULT_ENTRY):
user = (
app.model.context.query(app.model.User)
.filter(app.model.User.email == self.USER_2_APP_VAULT_ENTRY)
.first()
)
user_vault = UserVaultWrapper(self._app.vault, user)
user = self._get_user_by_email(self.USER_2_APP_VAULT_ENTRY)
user_vault = UserVaultWrapper(app.vault, user)
# use an invalid symlink path so the posix list would fail if used
user_vault.write_secret("posix/root_path", "/invalid/root")
@@ -109,12 +95,8 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
def test_upload_file_from_remote_source(self):
with self._different_user(email=self.USER_1_APP_VAULT_ENTRY):
app = self._app
user = (
app.model.context.query(app.model.User)
.filter(app.model.User.email == self.USER_1_APP_VAULT_ENTRY)
.first()
)
user_vault = UserVaultWrapper(self._app.vault, user)
user = self._get_user_by_email(self.USER_1_APP_VAULT_ENTRY)
user_vault = UserVaultWrapper(app.vault, user)
# use a valid symlink path so the posix list succeeds
user_vault.write_secret("posix/root_path", app.config.user_library_import_symlink_allowlist[0])
data = {"target": "gxfiles://test_user_vault"}
@@ -138,3 +120,7 @@ class TestVaultFileSourceIntegration(integration_util.IntegrationTestCase, integ
new_dataset = self.dataset_populator.fetch(payload, assert_ok=True).json()["outputs"][0]
content = self.dataset_populator.get_history_dataset_content(history_id, dataset=new_dataset)
assert content == "I require access to the vault", content
def _get_user_by_email(self, email):
stmt = select(self._app.model.User).filter(self._app.model.User.email == email).limit(1)
return self._app.model.session.scalars(stmt).first()