refactor(secrets): adapt to reana-commons secret-handling changes (re…

…anahub#583)

Closes reanahub/reana-commons#455

reana_workflow_controller/consumer.py

@@ -23,7 +23,7 @@
  current_k8s_batchv1_api_client,
  current_k8s_corev1_api_client,
 )
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore
 from reana_commons.utils import (
  calculate_file_access_time,
  calculate_hash_of_dir,
@@ -180,8 +180,16 @@ def _update_commit_status(workflow, status):
  state = "canceled"
  else:
  state = "running"
- secret_store = REANAUserSecretsStore(workflow.owner_id)
- gitlab_access_token = secret_store.get_secret_value("gitlab_access_token")
+
+ secret_store = UserSecretsStore.fetch(workflow.owner_id)
+ gitlab_access_token_secret = secret_store.get_secret("gitlab_access_token")
+ if not gitlab_access_token_secret:
+ logging.error(
+ f"Skipping updating commit status for workflow {workflow.id_}: GitLab access token not found."
+ )
+ return
+ gitlab_access_token = gitlab_access_token_secret.value_str
+
  target_url = f"https://{REANA_HOSTNAME}/api/workflows/{workflow.id_}/logs"
  workflow_name = urlparse.quote_plus(workflow.git_repo)
  system_name = "reana"
@@ -190,7 +198,13 @@ def _update_commit_status(workflow, status):
  f"{workflow.git_ref}?access_token={gitlab_access_token}&state={state}&"
  f"target_url={target_url}&name={system_name}"
  )
- requests.post(commit_status_url)
+
+ res = requests.post(commit_status_url)
+ if res.status_code >= 400:
+ logging.error(
+ f"Failed to update commit status for workflow {workflow.id_}: "
+ f"status code {res.status_code}, content {res.text}"
+ )
 
 
 def _update_run_progress(workflow_uuid, msg):

reana_workflow_controller/rest/utils.py

@@ -33,7 +33,7 @@
 from kubernetes.client.rest import ApiException
 from reana_commons import workspace
 from reana_commons.config import REANA_WORKFLOW_UMASK, WORKFLOW_TIME_FORMAT
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore
 from reana_commons.utils import (
  get_workflow_status_change_verb,
  remove_upper_level_references,
@@ -367,8 +367,11 @@ def create_workflow_workspace(
  os.umask(REANA_WORKFLOW_UMASK)
  os.makedirs(path, exist_ok=True)
  if git_url and git_ref:
- secret_store = REANAUserSecretsStore(user_id)
- gitlab_access_token = secret_store.get_secret_value("gitlab_access_token")
+ secret_store = UserSecretsStore.fetch(user_id)
+ gitlab_access_token_secret = secret_store.get_secret("gitlab_access_token")
+ if not gitlab_access_token_secret:
+ raise Exception("GitLab access token not found.")
+ gitlab_access_token = gitlab_access_token_secret.value_str
  url = "https://oauth2:{0}@{1}/{2}.git".format(
  gitlab_access_token, REANA_GITLAB_HOST, git_url
  )

reana_workflow_controller/workflow_run_manager.py

@@ -36,7 +36,7 @@
 )
 from reana_commons.k8s.api_client import current_k8s_batchv1_api_client
 from reana_commons.k8s.kerberos import get_kerberos_k8s_config
-from reana_commons.k8s.secrets import REANAUserSecretsStore
+from reana_commons.k8s.secrets import UserSecretsStore
 from reana_commons.k8s.volumes import (
  create_cvmfs_persistent_volume_claim,
  get_workspace_volume,
@@ -502,8 +502,7 @@ def _create_job_spec(
  namespace=REANA_RUNTIME_KUBERNETES_NAMESPACE,
  )
 
- secrets_store = REANAUserSecretsStore(owner_id)
-
+ secrets_store = UserSecretsStore.fetch(owner_id)
  kerberos = None
  if self.requires_kerberos():
  kerberos = get_kerberos_k8s_config(
@@ -566,7 +565,8 @@ def _create_job_spec(
 
  job_controller_env_secrets = secrets_store.get_env_secrets_as_k8s_spec()
 
- user = secrets_store.get_secret_value("CERN_USER") or WORKFLOW_RUNTIME_USER_NAME
+ user_secret = secrets_store.get_secret("CERN_USER")
+ user = user_secret.value_str if user_secret else WORKFLOW_RUNTIME_USER_NAME
 
  job_controller_container = client.V1Container(
  name=current_app.config["JOB_CONTROLLER_NAME"],

tests/test_consumer.py

@@ -14,9 +14,10 @@
 from reana_commons.config import MQ_DEFAULT_QUEUES
 from reana_commons.consumer import BaseConsumer
 from reana_commons.publisher import WorkflowStatusPublisher
+from reana_commons.k8s.secrets import UserSecrets, Secret
 from reana_db.models import RunStatus
 
-from reana_workflow_controller.consumer import JobStatusConsumer
+from reana_workflow_controller.consumer import JobStatusConsumer, _update_commit_status
 
 
 def test_workflow_finish_and_kubernetes_not_available(
@@ -45,3 +46,67 @@ def test_workflow_finish_and_kubernetes_not_available(
  ):
  consume_queue(job_status_consumer, limit=1)
  assert sample_serial_workflow_in_db.status == next_status
+
+
+@pytest.mark.parametrize(
+ "status,gitlab_status",
+ [
+ (RunStatus.created, "running"),
+ (RunStatus.deleted, "canceled"),
+ (RunStatus.failed, "failed"),
+ (RunStatus.finished, "success"),
+ (RunStatus.pending, "running"),
+ (RunStatus.queued, "running"),
+ (RunStatus.running, "running"),
+ (RunStatus.stopped, "canceled"),
+ ],
+)
+def test_update_commit_status(
+ session, sample_serial_workflow_in_db, status, gitlab_status
+):
+ """Test update commit status."""
+ workflow = sample_serial_workflow_in_db
+ workflow.git_repo = "foo/bar"
+ session.add(workflow)
+ session.commit()
+
+ post_mock = Mock()
+ post_mock.return_value.status_code = 200
+ secrets = UserSecrets(
+ user_id=str(workflow.owner_id),
+ k8s_secret_name="k8s-secret",
+ secrets=[Secret(name="gitlab_access_token", type_="env", value="my-token")],
+ )
+ fetch_mock = Mock()
+ fetch_mock.return_value = secrets
+ with patch("requests.post", post_mock), patch(
+ "reana_commons.k8s.secrets.UserSecretsStore.fetch", fetch_mock
+ ):
+ _update_commit_status(workflow, status)
+ fetch_mock.assert_called_once_with(workflow.owner_id)
+ post_mock.assert_called_once()
+ url = post_mock.call_args.args[0]
+ assert "access_token=my-token" in url
+ assert f"state={gitlab_status}" in url
+
+
+def test_update_commit_status_without_token(session, sample_serial_workflow_in_db):
+ """Test updating commit status without valid GitLab token."""
+ workflow = sample_serial_workflow_in_db
+ workflow.git_repo = "foo/bar"
+ session.add(workflow)
+ session.commit()
+
+ post_mock = Mock()
+ secrets = UserSecrets(
+ user_id=str(workflow.owner_id),
+ k8s_secret_name="k8s-secret",
+ )
+ fetch_mock = Mock()
+ fetch_mock.return_value = secrets
+ with patch("requests.post", post_mock), patch(
+ "reana_commons.k8s.secrets.UserSecretsStore.fetch", fetch_mock
+ ):
+ _update_commit_status(workflow, RunStatus.finished)
+ fetch_mock.assert_called_once_with(workflow.owner_id)
+ post_mock.assert_not_called()

tests/test_views.py

@@ -1155,8 +1155,6 @@ def test_start_workflow_db_failure(
  app,
  session,
  default_user,
- user_secrets,
- corev1_api_client_with_user_secrets,
  sample_serial_workflow_in_db,
 ):
  """Test starting workflow with a DB failure."""
@@ -1193,8 +1191,6 @@ def test_start_workflow_kubernetes_failure(
  app,
  session,
  default_user,
- user_secrets,
- corev1_api_client_with_user_secrets,
  sample_serial_workflow_in_db,
 ):
  """Test starting workflow with a Kubernetes failure when creating jobs."""

tests/test_workflow_run_manager.py

@@ -19,8 +19,6 @@
  RunStatus,
  InteractiveSession,
  InteractiveSessionType,
- JobStatus,
- Job,
 )
 
 from reana_workflow_controller.errors import REANAInteractiveSessionError