[PR-7651] Update TCNM to work with two AWS accounts

README.md

@@ -76,6 +76,12 @@ To upgrade a specific tool:
 ```
 pipx upgrade destroy-cumulus
 ```
+
+To test local changes, without missing with the global install:
+```
+pipx run --pip-args=--no-cache-dir --spec ~/Documents/cloud-tools/test-cnm tcnm <args>
+```
+
 </details>
 
 ### For local development

test-cnm/README.md

@@ -85,6 +85,13 @@ If no environment is specified the `default` environment will be used. If a
 config value is missing for the currently specified environment, the value will
 be pulled from the entry in the `[default]` section of the config instead.
 
+If the CNM response queue lives in a different AWS account than the ingest
+queue, set both `response_stack_name` (e.g. `cnsld-cumulus-uat2`) and
+`response_profile` (or pass `--response-stack-name` and `--response-profile`).
+Messages are still submitted to `<stack_name>-<cnm_ingest_queue>` using
+`profile`, but responses are read from
+`<response_stack_name>-<cnm_response_queue>` using `response_profile`.
+
 ## Adding New Data to the Bucket
 
 - Get the Granule S3 Path

test-cnm/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [project]
 name = "test-cnm"
-version = "0.10.1"
+version = "0.10.2"
 description = ""
 license = "Apache-2.0"
 authors = [

test-cnm/test_cnm/commands/cmd_test.py

@@ -41,6 +41,7 @@ def cmd_test(
     filters = args.filter
 
     session = config.session()
+    response_session = config.response_session()
 
     metadata = Metadata(session, config.test_bucket)
     metadata.load()
@@ -54,6 +55,7 @@ def cmd_test(
         make_cnm_s=CnmSGenerator(metadata),
         metadata=metadata,
         config=config,
+        response_session=response_session,
     )
 
     log.info("Executing tests on %s", config.stack_name)

test-cnm/test_cnm/config.py

@@ -2,6 +2,7 @@
 import configparser
 from collections.abc import Mapping
 from dataclasses import MISSING, dataclass, field, fields, replace
+from functools import cache
 from typing import Optional, Union
 
 try:
@@ -18,6 +19,11 @@ class ConfigError(Exception):
     pass
 
 
+@cache
+def _get_session(profile: Optional[str]) -> boto3.Session:
+    return boto3.Session(profile_name=profile)
+
+
 @dataclass
 class ConfigBase:
     _options: Optional[dict[str, str]] = field(
@@ -143,18 +149,32 @@ class ConfigFull(ConfigBase):
     provider: str
     default_data_version: str = "1.0"
     stack_name: Optional[str] = None
+    response_stack_name: Optional[str] = None
     trace: Optional[str] = None
     # TODO(reweeden): python3.10 refactor duplication
     profile: Optional[str] = None
+    response_profile: Optional[str] = None
 
     def session(self) -> boto3.Session:
-        return boto3.Session(profile_name=self.profile)
+        return _get_session(self.profile)
+
+    def response_session(self) -> boto3.Session:
+        if bool(self.response_stack_name) != bool(self.response_profile):
+            raise ConfigError("response_stack_name and response_profile must be set together, if set.")
+
+        if self.response_profile:
+            return _get_session(self.response_profile)
+
+        return self.session()
 
     def cnm_ingest_queue_name(self) -> str:
         return get_queue_name(self.stack_name, self.cnm_ingest_queue)
 
     def cnm_response_queue_name(self) -> str:
-        return get_queue_name(self.stack_name, self.cnm_response_queue)
+        return get_queue_name(
+            self.response_stack_name or self.stack_name,
+            self.cnm_response_queue,
+        )
 
 
 def get_queue_name(stack_name: Optional[str], queue_name: str) -> str:

test-cnm/test_cnm/main.py

@@ -88,6 +88,18 @@ def get_parser() -> argparse.ArgumentParser:
         "--trace",
         help="Value for the CNM-S trace element",
     )
+    response_group = parser.add_argument_group(
+        "response account",
+        "Options for reading the CNM response queue from a different account. Must be set together.",
+    )
+    response_group.add_argument(
+        "--response-stack-name",
+        help="Cumulus prefix of the second account e.g. cnsld-cumulus-uat2",
+    )
+    response_group.add_argument(
+        "--response-profile",
+        help="AWS profile name for the account containing the CNM response queue",
+    )
 
     subparsers = parser.add_subparsers(
         title="command",

test-cnm/test_cnm/tester/executor.py

@@ -1,7 +1,9 @@
 import json
 import logging
 from collections.abc import Generator
+from typing import Optional
 
+import boto3
 from test_cnm.config import ConfigFull
 from test_cnm.metadata import Metadata
 from test_cnm.tester.cnm_generator import CnmSGenerator
@@ -14,13 +16,15 @@
 class TestExecutor:
     def __init__(
         self,
-        session,
+        session: boto3.Session,
         collector: TestCollector,
         make_cnm_s: CnmSGenerator,
         metadata: Metadata,
         config: ConfigFull,
+        response_session: Optional[boto3.Session] = None,
     ):
         self.session = session
+        self.response_session = response_session or session
         self.collector = collector
         self.make_cnm_s = make_cnm_s
         self.metadata = metadata
@@ -83,6 +87,7 @@ def _get_ingest_client(self, test: ExecutableTest) -> CnmIngestClient:
                 session=self.executor.session,
                 make_cnm_s=self.executor.make_cnm_s,
                 cnm_response_queue=test.cnm_response_queue,
+                response_session=self.executor.response_session,
             )
 
         return self.ingest_clients[key]

test-cnm/test_cnm/tester/ingest_client.py

@@ -16,12 +16,15 @@ def __init__(
         session: boto3.Session,
         make_cnm_s: CnmSGeneratorType,
         cnm_response_queue: str,
+        response_session: Optional[boto3.Session] = None,
     ):
         self.session = session
+        self.response_session = response_session or session
         self.make_cnm_s = make_cnm_s
         self.cnm_response_queue = cnm_response_queue
 
         self.client = session.client("sqs")
+        self.response_client = self.response_session.client("sqs")
         self._requests = {}
 
     @property
@@ -80,7 +83,7 @@ def process_messages(self, wait_time_seconds: Optional[int] = None) -> Generator
                 message["Body"],
             )
 
-            self.client.delete_message(
+            self.response_client.delete_message(
                 QueueUrl=self.cnm_response_queue,
                 ReceiptHandle=message["ReceiptHandle"],
             )
@@ -90,7 +93,7 @@ def process_messages(self, wait_time_seconds: Optional[int] = None) -> Generator
     def _receive_message(self, wait_time_seconds: Optional[int] = None) -> dict:
         if wait_time_seconds is None:
             log.debug("Short polling %s", self.cnm_response_queue)
-            return self.client.receive_message(
+            return self.response_client.receive_message(
                 QueueUrl=self.cnm_response_queue,
                 MaxNumberOfMessages=10,
             )
@@ -100,7 +103,7 @@ def _receive_message(self, wait_time_seconds: Optional[int] = None) -> dict:
             self.cnm_response_queue,
             wait_time_seconds,
         )
-        return self.client.receive_message(
+        return self.response_client.receive_message(
             QueueUrl=self.cnm_response_queue,
             MaxNumberOfMessages=10,
             WaitTimeSeconds=wait_time_seconds,

test-cnm/tests/test_config.py

@@ -1,7 +1,13 @@
 import argparse
+from unittest import mock
 
 import pytest
-from test_cnm.config import ConfigBasic, ConfigError, ConfigFull
+from test_cnm.config import ConfigBasic, ConfigError, ConfigFull, _get_session
+
+
+@pytest.fixture(autouse=True)
+def clear_session_cache():
+    _get_session.cache_clear()
 
 
 def test_from_file_default(data_path):
@@ -240,3 +246,100 @@ def test_cnm_ingest_queue_name_postfix_with_stack_name():
     )
 
     assert config.cnm_ingest_queue_name() == "sds-n-cumulus-dev-nisar-workflow-queue"
+
+
+def test_response_session_defaults_to_session(monkeypatch):
+    config = ConfigFull(
+        profile="sbx",
+        test_bucket="",
+        cnm_ingest_queue="",
+        cnm_response_queue="",
+        provider="",
+    )
+    session = mock.Mock()
+    monkeypatch.setattr(config, "session", session)
+
+    assert config.response_session() is session.return_value
+    session.assert_called_once_with()
+
+
+def test_response_session_uses_response_profile(monkeypatch):
+    session_cls = mock.Mock()
+    monkeypatch.setattr("test_cnm.config.boto3.Session", session_cls)
+    config = ConfigFull(
+        profile="sbx",
+        response_profile="response_profile",
+        response_stack_name="cnsld-cumulus-uat2",
+        test_bucket="",
+        cnm_ingest_queue="",
+        cnm_response_queue="",
+        provider="",
+    )
+
+    assert config.response_session() is session_cls.return_value
+    session_cls.assert_called_once_with(profile_name="response_profile")
+
+
+def test_cnm_response_queue_name_uses_stack_name_by_default():
+    config = ConfigFull(
+        test_bucket="",
+        cnm_ingest_queue="",
+        cnm_response_queue="aria-mock-response-queue",
+        provider="",
+        stack_name="asf-cumulus-test",
+    )
+
+    assert config.cnm_response_queue_name() == "asf-cumulus-test-aria-mock-response-queue"
+
+
+def test_cnm_response_queue_name_response_stack_name():
+    config = ConfigFull(
+        test_bucket="",
+        cnm_ingest_queue="aria-workflow-queue",
+        cnm_response_queue="aria-mock-response-queue",
+        provider="",
+        stack_name="asf-cumulus-test",
+        response_stack_name="cnsld-cumulus-uat2",
+    )
+
+    assert config.cnm_ingest_queue_name() == "asf-cumulus-test-aria-workflow-queue"
+    assert config.cnm_response_queue_name() == "cnsld-cumulus-uat2-aria-mock-response-queue"
+
+
+def test_cnm_response_queue_name_response_stack_name_prefix():
+    config = ConfigFull(
+        test_bucket="",
+        cnm_ingest_queue="",
+        cnm_response_queue="cnsld-cumulus-uat2-aria-mock-response-queue",
+        provider="",
+        stack_name="asf-cumulus-test",
+        response_stack_name="cnsld-cumulus-uat2",
+    )
+
+    assert config.cnm_response_queue_name() == "cnsld-cumulus-uat2-aria-mock-response-queue"
+
+
+def test_response_session_response_profile_requires_response_stack_name():
+    config = ConfigFull(
+        test_bucket="",
+        cnm_ingest_queue="",
+        cnm_response_queue="",
+        provider="",
+        response_profile="response_profile",
+    )
+
+    with pytest.raises(ConfigError, match="must be set together"):
+        config.response_session()
+
+
+def test_response_session_response_stack_name_requires_response_profile():
+    config = ConfigFull(
+        test_bucket="",
+        cnm_ingest_queue="",
+        cnm_response_queue="",
+        provider="",
+        response_stack_name="cnsld-cumulus-uat2",
+    )
+
+    with pytest.raises(ConfigError, match="must be set together"):
+        config.response_session()

test-cnm/tests/test_ingest_client.py

@@ -1,5 +1,6 @@
 import json
 
+import boto3
 import pytest
 from test_cnm.tester.ingest_client import CnmIngestClient
 from test_cnm.tester.types import ExecutableTest
@@ -128,3 +129,25 @@ def test_process_messages(boto_session, sqs_client, mock_make_cnm_s, ingest_queu
         }
     ]
     assert not ingest_client.is_waiting
+
+
+def test_response_session(boto_session, mock_make_cnm_s):
+    ingest_client = CnmIngestClient(
+        session=boto_session,
+        make_cnm_s=mock_make_cnm_s,
+        cnm_response_queue="<unused>",
+        response_session=boto3.Session(region_name="us-west-2"),
+    )
+
+    assert ingest_client.client.meta.region_name == "us-east-1"
+    assert ingest_client.response_client.meta.region_name == "us-west-2"
+
+
+def test_response_session_defaults_to_session(boto_session, mock_make_cnm_s):
+    ingest_client = CnmIngestClient(
+        session=boto_session,
+        make_cnm_s=mock_make_cnm_s,
+        cnm_response_queue="<unused>",
+    )
+
+    assert ingest_client.response_client.meta.region_name == "us-east-1"