diff --git a/test-cnm/pyproject.toml b/test-cnm/pyproject.toml index b8f3eea..f0d5911 100644 --- a/test-cnm/pyproject.toml +++ b/test-cnm/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api" [tool.poetry] name = "test-cnm" -version = "0.8.2" +version = "0.9.0" description = "" authors = ["Rohan Weeden "] readme = "README.md" diff --git a/test-cnm/test_cnm/commands/cmd_configure.py b/test-cnm/test_cnm/commands/cmd_configure.py new file mode 100644 index 0000000..e6ee5fc --- /dev/null +++ b/test-cnm/test_cnm/commands/cmd_configure.py @@ -0,0 +1,70 @@ +import argparse +import logging + +from test_cnm.config import ConfigBasic +from test_cnm.metadata import Metadata +from test_cnm.tester.collector import BucketTestCollector + +log = logging.getLogger(__name__) + + +def add_parser( + subparsers: argparse._SubParsersAction, +) -> argparse.ArgumentParser: + parser_update_metadata = subparsers.add_parser( + "configure", + help="Update metadata file to set test level configuration", + ) + parser_update_metadata.add_argument( + "filter", + help="Glob pattern to filter tests by. Can include '*', '?' and '[]' expressions", + nargs="*", + default=[], + ) + properties_group = parser_update_metadata.add_argument_group("properties") + properties_group.add_argument( + "--cnm-ingest-queue", + help="Override the cnm_ingest_queue that the CNM-S will be sent to for the tests", + ) + properties_group.add_argument( + "--cnm-response-queue", + help="Override the cnm_response_queue that the CNM client will poll for the CNM-R response for the tests", + ) + parser_update_metadata.set_defaults( + func=cmd_update_metadata, + config_cls=ConfigBasic, + ) + + return parser_update_metadata + + +def cmd_update_metadata( + parser: argparse.ArgumentParser, + args: argparse.Namespace, + config: ConfigBasic, +): + filters = args.filter + + session = config.session() + + collector = BucketTestCollector( + session, + config.test_bucket, + ) + tests = collector.collect_tests(filters) + + with Metadata(session, config.test_bucket) as metadata: + for test in tests.values(): + test_id = test.get_id() + cfg = metadata.test_config[test_id] + + for attr in ("cnm_ingest_queue", "cnm_response_queue"): + value = getattr(args, attr) + if value: + log.debug("%s setting %s to %s", test_id, attr, value) + cfg[attr] = value + + if cfg: + log.info("%s:", test_id) + for attr, value in cfg.items(): + log.info(" %s: %s", attr, value) diff --git a/test-cnm/test_cnm/commands/cmd_list.py b/test-cnm/test_cnm/commands/cmd_list.py index 4e2a06d..8f45a87 100644 --- a/test-cnm/test_cnm/commands/cmd_list.py +++ b/test-cnm/test_cnm/commands/cmd_list.py @@ -59,8 +59,7 @@ def cmd_list( for collection, grouped_tests in tests_by_collection.items(): log.info("%s:", collection) for test in grouped_tests: - prefix = f"{collection}/" - test_id = test.get_id().removeprefix(prefix) + test_id = test.get_id() log.info(" - %s (%d files)", test_id, len(test.files)) if args.files: last_idx = len(test.files) - 1 diff --git a/test-cnm/test_cnm/commands/cmd_test.py b/test-cnm/test_cnm/commands/cmd_test.py index bf2d4b2..2b735f4 100644 --- a/test-cnm/test_cnm/commands/cmd_test.py +++ b/test-cnm/test_cnm/commands/cmd_test.py @@ -6,7 +6,6 @@ from test_cnm.tester.cnm_generator import CnmSGenerator from test_cnm.tester.collector import BucketTestCollector from test_cnm.tester.executor import TestExecutor -from test_cnm.tester.ingest_client import CnmIngestClient from tqdm import tqdm from tqdm.contrib.logging import logging_redirect_tqdm @@ -46,24 +45,15 @@ def cmd_test( metadata = Metadata(session, config.test_bucket) metadata.load() - collector = BucketTestCollector( - session, - config.test_bucket, - ) - ingest_client = CnmIngestClient( + executor = TestExecutor( session=session, - make_cnm_s=CnmSGenerator( - provider=args.provider or "ASF-TESTCNM", - trace=config.trace, - metadata=metadata, + collector=BucketTestCollector( + session, + config.test_bucket, ), - start_queue=config.cnm_ingest_queue_name(), - response_queue=config.cnm_response_queue_name(), - ) - executor = TestExecutor( - collector, - ingest_client, - config.default_data_version, + make_cnm_s=CnmSGenerator(metadata), + metadata=metadata, + config=config, ) log.info("Executing tests on %s", config.stack_name) diff --git a/test-cnm/test_cnm/commands/cmd_tidy.py b/test-cnm/test_cnm/commands/cmd_tidy.py index 90ad308..7dbf0be 100644 --- a/test-cnm/test_cnm/commands/cmd_tidy.py +++ b/test-cnm/test_cnm/commands/cmd_tidy.py @@ -43,6 +43,7 @@ def cmd_tidy( with Metadata(session, config.test_bucket) as metadata: extra_metadata = dict(metadata.metadata) + extra_metadata.pop(metadata.test_config_key, None) total = 0 deleted = 0 diff --git a/test-cnm/test_cnm/commands/cmd_update_metadata.py b/test-cnm/test_cnm/commands/cmd_update_metadata.py index ba454ef..d26cc8e 100644 --- a/test-cnm/test_cnm/commands/cmd_update_metadata.py +++ b/test-cnm/test_cnm/commands/cmd_update_metadata.py @@ -7,7 +7,8 @@ import boto3 from test_cnm.config import ConfigBasic -from test_cnm.metadata import CHECKSUM_PATTERN, ChecksumWriter, Metadata +from test_cnm.metadata import CHECKSUM_PATTERN, Metadata +from test_cnm.uploader import ChecksumWriter log = logging.getLogger(__name__) diff --git a/test-cnm/test_cnm/config.py b/test-cnm/test_cnm/config.py index d1ed963..be33c5b 100644 --- a/test-cnm/test_cnm/config.py +++ b/test-cnm/test_cnm/config.py @@ -1,8 +1,14 @@ import argparse import configparser -from dataclasses import MISSING, dataclass, fields +from collections.abc import Mapping +from dataclasses import MISSING, dataclass, field, fields, replace from typing import Optional, Union +try: + from typing import Self +except ImportError: + Self = "ConfigBase" + import boto3 DEFAULT = "default" @@ -14,12 +20,31 @@ class ConfigError(Exception): @dataclass class ConfigBase: + _options: Optional[dict[str, str]] = field( + default=None, + init=False, + repr=False, + compare=False, + ) + _section: Optional[dict[str, str]] = field( + default=None, + init=False, + repr=False, + compare=False, + ) + _default_section: Optional[dict[str, str]] = field( + default=None, + init=False, + repr=False, + compare=False, + ) + @classmethod def from_file( cls, filenames: Union[str, list[str]], args: Optional[argparse.Namespace] = None, - ) -> "ConfigBase": + ) -> Self: config = configparser.ConfigParser(default_section=None) config.read(filenames) @@ -32,31 +57,71 @@ def from_file( section = config[env] if config.has_section(env) else {} default_section = config[DEFAULT] if config.has_section(DEFAULT) else {} + kwargs = cls._resolve_kwargs( + options=options, + section=section, + default_section=default_section, + ) + missing = [ + # ruff hint + k + for k, v in kwargs.items() + if v is MISSING + ] + if missing: + raise ConfigError(f"missing values for {missing} for environment '{env}'") + + obj = cls(**kwargs) + obj._options = options + obj._section = {} if env == DEFAULT else dict(section) + obj._default_section = dict(default_section) + + return obj + + def dynamic_config(self, cfg: dict) -> Self: + if self._options is None or self._section is None or self._default_section is None: + obj = replace(self, **cfg) + else: + kwargs = self._resolve_kwargs( + options=self._options, + section=self._section, + default_section=self._default_section, + dynamic_config=cfg, + ) + obj = self.__class__(**kwargs) + + obj._options = self._options + obj._section = self._section + obj._default_section = self._default_section + + return obj + + @classmethod + def _resolve_kwargs( + cls, + options: Mapping[str, str], + section: Mapping[str, str], + default_section: Mapping[str, str], + dynamic_config: Optional[Mapping[str, str]] = None, + ) -> dict[str, str]: def _get_value(key: str, default=None): if (val := options.get(key)) is not None: return val if (val := section.get(key)) is not None: return val + if dynamic_config and (val := dynamic_config.get(key)) is not None: + return val if (val := default_section.get(key)) is not None: return val return default - kwargs = { + return { # ruff hint field.name: _get_value(field.name, default=field.default) for field in fields(cls) + if not field.name.startswith("_") } - missing = [ - # ruff hint - k - for field, (k, v) in zip(fields(cls), kwargs.items()) - if v is MISSING - ] - if missing: - raise ConfigError(f"missing values for {missing} for environment '{env}'") - - return cls(**kwargs) @dataclass @@ -86,13 +151,14 @@ def session(self) -> boto3.Session: return boto3.Session(profile_name=self.profile) def cnm_ingest_queue_name(self) -> str: - if self.stack_name and not self.cnm_ingest_queue.startswith(self.stack_name): - return f"{self.stack_name}-{self.cnm_ingest_queue}" - - return self.cnm_ingest_queue + return get_queue_name(self.stack_name, self.cnm_ingest_queue) def cnm_response_queue_name(self) -> str: - if self.stack_name and not self.cnm_response_queue.startswith(self.stack_name): - return f"{self.stack_name}-{self.cnm_response_queue}" + return get_queue_name(self.stack_name, self.cnm_response_queue) + + +def get_queue_name(stack_name: Optional[str], queue_name: str) -> str: + if stack_name and not queue_name.startswith(stack_name): + return f"{stack_name}-{queue_name}" - return self.cnm_response_queue + return queue_name diff --git a/test-cnm/test_cnm/main.py b/test-cnm/test_cnm/main.py index 73b4956..ecf8b28 100644 --- a/test-cnm/test_cnm/main.py +++ b/test-cnm/test_cnm/main.py @@ -16,6 +16,7 @@ from typing import Optional from test_cnm.commands import ( + cmd_configure, cmd_list, cmd_move, cmd_test, @@ -95,6 +96,7 @@ def get_parser() -> argparse.ArgumentParser: dest="command", ) + cmd_configure.add_parser(subparsers) cmd_list.add_parser(subparsers) cmd_move.add_parser(subparsers) cmd_test.add_parser(subparsers) diff --git a/test-cnm/test_cnm/metadata.py b/test-cnm/test_cnm/metadata.py index 2871770..ece28ce 100644 --- a/test-cnm/test_cnm/metadata.py +++ b/test-cnm/test_cnm/metadata.py @@ -1,12 +1,18 @@ import codecs +import hashlib import io import json import logging import re -from typing import IO +from collections import defaultdict import boto3 +try: + from typing import Self +except ImportError: + Self = "Metadata" + log = logging.getLogger(__name__) @@ -23,8 +29,19 @@ def __init__( self.session = session self.bucket = bucket self.key = key + self.test_config_key = "$testconfig" + + self.metadata = defaultdict(dict) + self._metadata_hash = "" - self.metadata = {} + def _get_metadata_hash(self) -> str: + return hashlib.md5( + json.dumps(self.metadata, sort_keys=True).encode(), + ).hexdigest() + + @property + def test_config(self) -> dict: + return self[self.test_config_key] def load(self): client = self.session.client("s3") @@ -43,7 +60,12 @@ def load(self): Key=self.key, ) buf.seek(0) - self.metadata = json.load(buf) + self.metadata = defaultdict(dict, json.load(buf)) + self.metadata[self.test_config_key] = defaultdict( + dict, + self.metadata[self.test_config_key], + ) + self._metadata_hash = self._get_metadata_hash() except Exception as e: log.error("Failed to load metadata file: %s", e) log.debug( @@ -56,6 +78,11 @@ def load(self): def save(self): client = self.session.client("s3") + new_metadata_hash = self._get_metadata_hash() + if new_metadata_hash == self._metadata_hash: + log.debug("No changes to metadata file. Skipping save()") + return + log.debug( "Saving metadata file to s3://%s/%s", self.bucket, @@ -63,6 +90,8 @@ def save(self): ) try: + if not self[self.test_config_key]: + del self[self.test_config_key] # Json requires a StringIO, but boto3 wants a BytesIO StreamWriter = codecs.getwriter("utf-8") @@ -74,6 +103,7 @@ def save(self): Bucket=self.bucket, Key=self.key, ) + self._metadata_hash = new_metadata_hash except Exception as e: log.error("Failed to save checksums file: %s", e) log.debug( @@ -94,7 +124,7 @@ def delete(self, key: str, item: str): if not entry: del self.metadata[key] - def __enter__(self) -> "Metadata": + def __enter__(self) -> Self: self.load() return self @@ -108,34 +138,10 @@ def __delitem__(self, key: str): del self.metadata[key] def __getitem__(self, key: str) -> dict: - if key not in self.metadata: - self.metadata[key] = {} + if key == self.test_config_key and key not in self.metadata: + self.metadata[key] = defaultdict(dict) return self.metadata[key] def __setitem__(self, key: str, value: dict): self.metadata[key] = value - - -class ChecksumReaderProxy: - """Compute a checksum while reading from a file-like object""" - - def __init__(self, f: IO[bytes], hash_obj): - self.f = f - self.hash_obj = hash_obj - - def read(self, n: int = -1) -> bytes: - data = self.f.read(n) - self.hash_obj.update(data) - return data - - -class ChecksumWriter: - """A file-like object that computes a checksum when consuming data""" - - def __init__(self, hash_obj): - self.hash_obj = hash_obj - - def write(self, data: bytes) -> int: - self.hash_obj.update(data) - return len(data) diff --git a/test-cnm/test_cnm/tester/cnm_generator.py b/test-cnm/test_cnm/tester/cnm_generator.py index 009e522..96641af 100644 --- a/test-cnm/test_cnm/tester/cnm_generator.py +++ b/test-cnm/test_cnm/tester/cnm_generator.py @@ -1,6 +1,8 @@ +import uuid +from collections.abc import Callable from datetime import datetime, timezone from pathlib import Path -from typing import Callable, Optional +from typing import Optional from test_cnm.metadata import CHECKSUM_PATTERN, Metadata @@ -22,18 +24,14 @@ ".xml": "data", } -CnmSGeneratorType = Callable[[str, str, str, list], dict] +CnmSGeneratorType = Callable[[str, str, str, list, str, Optional[str]], dict] class CnmSGenerator: def __init__( self, - provider: str, - trace: Optional[str] = None, metadata: Optional[Metadata] = None, ): - self.provider = provider - self.trace = trace self.metadata = metadata def __call__( @@ -42,9 +40,11 @@ def __call__( data_version: str, name: str, files: list, + provider: str, + trace: Optional[str] = None, ) -> dict: cnm_s = { - "identifier": name, + "identifier": str(uuid.uuid4()), "collection": collection, "version": "1.3", "submissionTime": datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ"), @@ -63,10 +63,10 @@ def __call__( for file in files ], }, - "provider": self.provider, + "provider": provider, } - if self.trace: - cnm_s["trace"] = self.trace + if trace: + cnm_s["trace"] = trace return cnm_s diff --git a/test-cnm/test_cnm/tester/collector.py b/test-cnm/test_cnm/tester/collector.py index d1980df..2fe713f 100644 --- a/test-cnm/test_cnm/tester/collector.py +++ b/test-cnm/test_cnm/tester/collector.py @@ -2,46 +2,17 @@ import logging import re from collections import defaultdict -from dataclasses import dataclass from pathlib import Path -from typing import Optional, Protocol, TypedDict +from typing import Optional import boto3 +from test_cnm.tester.types import FileDict, TestInfo log = logging.getLogger(__name__) DATA_VERSION_PATTERN = re.compile(r"^\d+\.\d+|\d$") -class FileDict(TypedDict): - Bucket: str - Key: str - Size: int - ETag: str - - -@dataclass -class TestInfo: - collection: str - data_version: Optional[str] - name: str - files: list[FileDict] - cnm_s: Optional[dict] = None - - def get_id(self) -> str: - if self.data_version: - return self.get_full_id(self.data_version) - - return f"{self.collection}/{self.name}" - - def get_full_id(self, default_data_version: str) -> str: - return f"{self.collection}/{self.data_version or default_data_version}/{self.name}" - - -class TestCollector(Protocol): - def collect_tests(self, filters: list[str]) -> dict[str, TestInfo]: ... - - class BucketTestCollector: def __init__(self, session: boto3.Session, bucket: str): self.session = session @@ -64,7 +35,7 @@ def collect_tests(self, filters: list[str]) -> dict[str, TestInfo]: collection = path.parts[0] data_version = path.parts[1] if DATA_VERSION_PATTERN.fullmatch(path.parts[1]) else None name = path.parts[-2] - s3_entries[(collection, data_version, name)].append( + s3_entries[collection, data_version, name].append( { "Bucket": response["Name"], "Key": key, diff --git a/test-cnm/test_cnm/tester/executor.py b/test-cnm/test_cnm/tester/executor.py index 4f0c993..0e5def9 100644 --- a/test-cnm/test_cnm/tester/executor.py +++ b/test-cnm/test_cnm/tester/executor.py @@ -2,8 +2,11 @@ import logging from collections.abc import Generator -from test_cnm.tester.collector import TestCollector, TestInfo +from test_cnm.config import ConfigFull +from test_cnm.metadata import Metadata +from test_cnm.tester.cnm_generator import CnmSGenerator from test_cnm.tester.ingest_client import CnmIngestClient +from test_cnm.tester.types import ExecutableTest, TestCollector, TestInfo log = logging.getLogger(__name__) @@ -11,13 +14,17 @@ class TestExecutor: def __init__( self, + session, collector: TestCollector, - ingest_client: CnmIngestClient, - default_data_version: str, + make_cnm_s: CnmSGenerator, + metadata: Metadata, + config: ConfigFull, ): + self.session = session self.collector = collector - self.ingest_client = ingest_client - self.default_data_version = default_data_version + self.make_cnm_s = make_cnm_s + self.metadata = metadata + self.config = config def new_test_run(self, filters: list[str]) -> "TestRun": return TestRun(self, filters) @@ -25,14 +32,36 @@ def new_test_run(self, filters: list[str]) -> "TestRun": def run(self, filters: list[str]): self.new_test_run(filters).run() + def resolve_test(self, test: TestInfo) -> ExecutableTest: + config = self.config.dynamic_config(self.metadata.test_config[test.get_id()]) + log.debug("Config for test %s: %s", test.get_id(), config) + + return ExecutableTest( + collection=test.collection, + data_version=test.data_version, + resolved_data_version=test.data_version or config.default_data_version, + name=test.name, + files=test.files, + cnm_ingest_queue=config.cnm_ingest_queue_name(), + cnm_response_queue=config.cnm_response_queue_name(), + provider=config.provider, + trace=config.trace, + ) + class TestRun: def __init__(self, executor: TestExecutor, filters: list[str]): self.executor = executor self.filters = filters - self.tests = {} + self.ingest_clients: dict[str, CnmIngestClient] = {} + self.tests: list[ExecutableTest] = [] self.pending_tests = {} + # For deduplicating based on name. This is no longer strictly required + # but may be desirable for now as it was the existing behavior. Cumulus + # has also since pushed out an update to allow repeated granule ids + # across multiple collections. + self.pending_tests_by_name = {} # Stats self.num_skipped = 0 @@ -46,6 +75,36 @@ def __init__(self, executor: TestExecutor, filters: list[str]): def num_completed(self) -> int: return self.num_succeeded + self.num_failed + def _get_ingest_client(self, test: ExecutableTest) -> CnmIngestClient: + key = test.cnm_response_queue + + if key not in self.ingest_clients: + self.ingest_clients[key] = CnmIngestClient( + session=self.executor.session, + make_cnm_s=self.executor.make_cnm_s, + cnm_response_queue=test.cnm_response_queue, + ) + + return self.ingest_clients[key] + + def _iter_ingest_client_responses(self) -> Generator[dict]: + while True: + clients_waiting = [ + # ruff hint + client + for client in self.ingest_clients.values() + if client.is_waiting + ] + + if not clients_waiting: + return + + # Short poll first + for ingest_client in clients_waiting[1:]: + yield from ingest_client.process_messages() + + yield from clients_waiting[0].process_messages(wait_time_seconds=5) + def run(self): self.collect_tests() for _ in self.iter_start_tests(): @@ -57,44 +116,46 @@ def run(self): def collect_tests(self): assert self._state == "not_started", "Tests must be collected only once" - # Collect - self.tests = self.executor.collector.collect_tests(self.filters) + self.tests = [ + # ruff hint + self.executor.resolve_test(test) + for test in self.executor.collector.collect_tests(self.filters).values() + ] self._state = "tests_collected" - def iter_start_tests(self) -> Generator[TestInfo]: + def iter_start_tests(self) -> Generator[ExecutableTest]: assert self._state == "tests_collected", "Tests must be collected first" self.pending_tests.clear() - for test in self.tests.values(): - if test.name in self.pending_tests: + self.pending_tests_by_name.clear() + for test in self.tests: + if test.name in self.pending_tests_by_name: log.warning( "Skipping %s as the product name conflicts with already started test %s", test.get_id(), - self.pending_tests[test.name].get_id(), + self.pending_tests_by_name[test.name].get_id(), ) self.num_skipped += 1 continue log.info("Starting: %s", test.get_id()) - test.cnm_s = self.executor.ingest_client.submit_request( - test.collection, - test.data_version or self.executor.default_data_version, - test.name, - test.files, - ) + test.cnm_s = self._get_ingest_client(test).submit_request(test) self.num_started += 1 - self.pending_tests[test.name] = test + key = (test.cnm_s["identifier"], test.cnm_s["submissionTime"]) + self.pending_tests[key] = test + self.pending_tests_by_name[test.name] = test yield test self._state = "tests_started" - def iter_responses(self) -> Generator[tuple[TestInfo, dict]]: + def iter_responses(self) -> Generator[tuple[ExecutableTest, dict]]: assert self._state == "tests_started", "Tests must be started first" - for name, cnm_r in self.executor.ingest_client.iter_responses(): - test = self.pending_tests.pop(name) + for cnm_r in self._iter_ingest_client_responses(): + key = (cnm_r["identifier"], cnm_r["submissionTime"]) + test = self.pending_tests.pop(key) response = cnm_r.get("response", {}) status = response.get("status") diff --git a/test-cnm/test_cnm/tester/ingest_client.py b/test-cnm/test_cnm/tester/ingest_client.py index 19e479d..a058095 100644 --- a/test-cnm/test_cnm/tester/ingest_client.py +++ b/test-cnm/test_cnm/tester/ingest_client.py @@ -1,8 +1,11 @@ import json import logging +from collections.abc import Generator +from typing import Optional import boto3 from test_cnm.tester.cnm_generator import CnmSGeneratorType +from test_cnm.tester.types import ExecutableTest log = logging.getLogger(__name__) @@ -12,61 +15,93 @@ def __init__( self, session: boto3.Session, make_cnm_s: CnmSGeneratorType, - start_queue: str, - response_queue: str, + cnm_response_queue: str, ): self.session = session self.make_cnm_s = make_cnm_s - self.start_queue = start_queue - self.response_queue = response_queue + self.cnm_response_queue = cnm_response_queue self.client = session.client("sqs") self._requests = {} - def submit_request( - self, - collection: str, - data_version: str, - name: str, - files: list, - ): - cnm_s = self.make_cnm_s(collection, data_version, name, files) - self._requests[(name, cnm_s["submissionTime"])] = True + @property + def is_waiting(self) -> bool: + return bool(self._requests) + + def submit_request(self, test: ExecutableTest) -> dict: + assert test.cnm_response_queue == self.cnm_response_queue + + cnm_s = self.make_cnm_s( + test.collection, + test.resolved_data_version, + test.name, + test.files, + test.provider, + test.trace, + ) message_body = json.dumps(cnm_s, indent=2) - log.debug("Sending CNM-S: %s", message_body) + log.debug("Sending CNM-S to %s: %s", test.cnm_ingest_queue, message_body) self.client.send_message( - QueueUrl=self.start_queue, + QueueUrl=test.cnm_ingest_queue, MessageBody=message_body, ) + + key = (cnm_s["identifier"], cnm_s["submissionTime"]) + if key in self._requests: + log.warning("CNM-S identifier conflicts with in-flight request %s", key) + self._requests[key] = True + return cnm_s - def get_responses(self): - return { - # ruff hint - name: cnm_r - for name, cnm_r in self.iter_responses() - } - - def iter_responses(self): - while self._requests: - response = self.client.receive_message( - QueueUrl=self.response_queue, - WaitTimeSeconds=5, + def iter_responses(self) -> Generator[dict]: + while self.is_waiting: + yield from self.process_messages(wait_time_seconds=5) + + def process_messages(self, wait_time_seconds: Optional[int] = None) -> Generator[dict]: + if not self.is_waiting: + return + + response = self._receive_message(wait_time_seconds=wait_time_seconds) + + for message in response.get("Messages", ()): + cnm_r = json.loads(message["Body"]) + request = self._requests.pop( + (cnm_r["identifier"], cnm_r["submissionTime"]), + None, + ) + if request is None: + continue + + log.debug( + "Received CNM-R from %s: %s", + self.cnm_response_queue, + message["Body"], + ) + + self.client.delete_message( + QueueUrl=self.cnm_response_queue, + ReceiptHandle=message["ReceiptHandle"], ) - for message in response.get("Messages", ()): - cnm_r = json.loads(message["Body"]) - name = cnm_r["identifier"] - request = self._requests.pop((name, cnm_r["submissionTime"]), None) - if request is None: - continue - log.debug("Received CNM-R: %s", message["Body"]) + yield cnm_r - self.client.delete_message( - QueueUrl=self.response_queue, - ReceiptHandle=message["ReceiptHandle"], - ) + 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( + QueueUrl=self.cnm_response_queue, + MaxNumberOfMessages=10, + ) - yield name, cnm_r + log.debug( + "Long polling %s for %s seconds", + self.cnm_response_queue, + wait_time_seconds, + ) + return self.client.receive_message( + QueueUrl=self.cnm_response_queue, + MaxNumberOfMessages=10, + WaitTimeSeconds=wait_time_seconds, + ) diff --git a/test-cnm/test_cnm/tester/types.py b/test-cnm/test_cnm/tester/types.py new file mode 100644 index 0000000..ebd2fa9 --- /dev/null +++ b/test-cnm/test_cnm/tester/types.py @@ -0,0 +1,37 @@ +from dataclasses import dataclass +from typing import Optional, Protocol, TypedDict + + +class FileDict(TypedDict): + Bucket: str + Key: str + Size: int + ETag: str + + +@dataclass +class TestInfo: + collection: str + data_version: Optional[str] + name: str + files: list[FileDict] + + def get_id(self) -> str: + if self.data_version: + return f"{self.collection}/{self.data_version}/{self.name}" + + return f"{self.collection}/{self.name}" + + +@dataclass +class ExecutableTest(TestInfo): + resolved_data_version: str + cnm_ingest_queue: str + cnm_response_queue: str + provider: str + trace: Optional[str] = None + cnm_s: Optional[dict] = None + + +class TestCollector(Protocol): + def collect_tests(self, filters: list[str]) -> dict[str, TestInfo]: ... diff --git a/test-cnm/test_cnm/uploader.py b/test-cnm/test_cnm/uploader.py index a3bb410..dd941f6 100644 --- a/test-cnm/test_cnm/uploader.py +++ b/test-cnm/test_cnm/uploader.py @@ -1,14 +1,38 @@ import hashlib import logging from pathlib import Path -from typing import Optional +from typing import IO, Optional import boto3 -from test_cnm.metadata import ChecksumReaderProxy, Metadata +from test_cnm.metadata import Metadata log = logging.getLogger(__name__) +class ChecksumReaderProxy: + """Compute a checksum while reading from a file-like object""" + + def __init__(self, f: IO[bytes], hash_obj): + self.f = f + self.hash_obj = hash_obj + + def read(self, n: int = -1) -> bytes: + data = self.f.read(n) + self.hash_obj.update(data) + return data + + +class ChecksumWriter: + """A file-like object that computes a checksum when consuming data""" + + def __init__(self, hash_obj): + self.hash_obj = hash_obj + + def write(self, data: bytes) -> int: + self.hash_obj.update(data) + return len(data) + + class Uploader: def __init__( self, diff --git a/test-cnm/tests/conftest.py b/test-cnm/tests/conftest.py index b690c6a..f0f517c 100644 --- a/test-cnm/tests/conftest.py +++ b/test-cnm/tests/conftest.py @@ -1,6 +1,7 @@ import logging import os from pathlib import Path +from typing import Optional import boto3 import pytest @@ -26,18 +27,66 @@ def aws_credentials(): os.environ["AWS_DEFAULT_REGION"] = "us-east-1" -@pytest.fixture(scope="session") -def get_client(): - return boto3.client +@pytest.fixture +def boto_session(): + with mock_aws(): + yield boto3.Session() @pytest.fixture -def s3_resource(): - with mock_aws(): - yield boto3.resource("s3") +def s3_resource(boto_session): + return boto_session.resource("s3") @pytest.fixture -def sqs_client(): - with mock_aws(): - yield boto3.client("sqs") +def sqs_client(boto_session): + return boto_session.client("sqs") + + +@pytest.fixture +def test_bucket(s3_resource): + bucket = s3_resource.Bucket("test-bucket") + bucket.create() + + return bucket + + +@pytest.fixture +def mock_make_cnm_s(): + uid = 0 + + def mock_make_cnm_s( + collection: str, + data_version: str, + name: str, + files: list, + provider: str, + trace: Optional[str], + ) -> dict: + nonlocal uid + uid += 1 + + return { + "identifier": str(uid), + "collection": collection, + "version": "1.3", + "submissionTime": "2026-01-01T00:00:00.000Z", + "product": { + "name": name, + "dataVersion": data_version, + "files": [ + { + "name": Path(file["Key"]).name, + "type": "data", + "uri": f"s3://{file['Bucket']}/{file['Key']}", + "size": file["Size"], + "checksum": file["ETag"], + "checksumType": "md5", + } + for file in files + ], + }, + "provider": provider, + } + + return mock_make_cnm_s diff --git a/test-cnm/tests/test_cli/conftest.py b/test-cnm/tests/test_cli/conftest.py index 491035d..d56aa9f 100644 --- a/test-cnm/tests/test_cli/conftest.py +++ b/test-cnm/tests/test_cli/conftest.py @@ -5,14 +5,6 @@ import pytest -@pytest.fixture -def test_bucket(s3_resource): - bucket = s3_resource.Bucket("test-bucket") - bucket.create() - - return bucket - - @pytest.fixture(autouse=True) def home_directory(tmp_path, test_bucket, monkeypatch): home_path = tmp_path / "home" diff --git a/test-cnm/tests/test_cli/test_cmd_list.py b/test-cnm/tests/test_cli/test_cmd_list.py index 33f263a..0d878fd 100644 --- a/test-cnm/tests/test_cli/test_cmd_list.py +++ b/test-cnm/tests/test_cli/test_cmd_list.py @@ -44,16 +44,16 @@ def test_list_all(capcli): Using config: ConfigBasic(test_bucket='test-bucket', profile=None) Collecting tests from bucket test-bucket COLLECTION_1: - - PRODUCT_1 (2 files) - - PRODUCT_2 (2 files) + - COLLECTION_1/PRODUCT_1 (2 files) + - COLLECTION_1/PRODUCT_2 (2 files) COLLECTION_2: - - PRODUCT_1 (4 files) + - COLLECTION_2/PRODUCT_1 (4 files) DATA_VERSION_1: - - 1.0/PRODUCT_1 (2 files) + - DATA_VERSION_1/1.0/PRODUCT_1 (2 files) DATA_VERSION_2: - - 1/PRODUCT_1 (2 files) + - DATA_VERSION_2/1/PRODUCT_1 (2 files) OTHER_COLLECTION: - - PRODUCT_1 (2 files) + - OTHER_COLLECTION/PRODUCT_1 (2 files) Totals: 5 Collections; 6 Tests """.lstrip("\n") @@ -68,8 +68,8 @@ def test_list_prefix(capcli): Using config: ConfigBasic(test_bucket='test-bucket', profile=None) Collecting tests from bucket test-bucket COLLECTION_1: - - PRODUCT_1 (2 files) - - PRODUCT_2 (2 files) + - COLLECTION_1/PRODUCT_1 (2 files) + - COLLECTION_1/PRODUCT_2 (2 files) Totals: 1 Collections; 2 Tests """.lstrip("\n") @@ -84,10 +84,10 @@ def test_list_prefix_multiple(capcli): Using config: ConfigBasic(test_bucket='test-bucket', profile=None) Collecting tests from bucket test-bucket COLLECTION_1: - - PRODUCT_1 (2 files) - - PRODUCT_2 (2 files) + - COLLECTION_1/PRODUCT_1 (2 files) + - COLLECTION_1/PRODUCT_2 (2 files) COLLECTION_2: - - PRODUCT_1 (4 files) + - COLLECTION_2/PRODUCT_1 (4 files) Totals: 2 Collections; 3 Tests """.lstrip("\n") @@ -102,28 +102,28 @@ def test_list_files(capcli): Using config: ConfigBasic(test_bucket='test-bucket', profile=None) Collecting tests from bucket test-bucket COLLECTION_1: - - PRODUCT_1 (2 files) + - COLLECTION_1/PRODUCT_1 (2 files) ├── file1.txt └── file2.txt - - PRODUCT_2 (2 files) + - COLLECTION_1/PRODUCT_2 (2 files) ├── file1.txt └── file2.txt COLLECTION_2: - - PRODUCT_1 (4 files) + - COLLECTION_2/PRODUCT_1 (4 files) ├── file1.txt ├── file2.txt ├── file3.txt └── file4.txt DATA_VERSION_1: - - 1.0/PRODUCT_1 (2 files) + - DATA_VERSION_1/1.0/PRODUCT_1 (2 files) ├── file1.txt └── file2.txt DATA_VERSION_2: - - 1/PRODUCT_1 (2 files) + - DATA_VERSION_2/1/PRODUCT_1 (2 files) ├── file1.txt └── file2.txt OTHER_COLLECTION: - - PRODUCT_1 (2 files) + - OTHER_COLLECTION/PRODUCT_1 (2 files) ├── file1.txt └── file2.txt diff --git a/test-cnm/tests/test_cli/test_cmd_move.py b/test-cnm/tests/test_cli/test_cmd_move.py index 7e44058..365c9df 100644 --- a/test-cnm/tests/test_cli/test_cmd_move.py +++ b/test-cnm/tests/test_cli/test_cmd_move.py @@ -13,6 +13,11 @@ def test_bucket(test_bucket): metadata.put( Body=json.dumps( { + "$testconfig": { + "COLLECTION_1/PRODUCT_1": { + "foo": "bar", + }, + }, obj1.key: { "checksum": "11111111111111111111111111111111", "type": "data", @@ -35,3 +40,9 @@ def test_move_file(test_bucket): "checksum": "11111111111111111111111111111111", "type": "data", } + # Testconfig section is not updated + assert metadata_dict["$testconfig"] == { + "COLLECTION_1/PRODUCT_1": { + "foo": "bar", + }, + } diff --git a/test-cnm/tests/test_cli/test_cmd_test.py b/test-cnm/tests/test_cli/test_cmd_test.py index 65e54be..1889581 100644 --- a/test-cnm/tests/test_cli/test_cmd_test.py +++ b/test-cnm/tests/test_cli/test_cmd_test.py @@ -1,7 +1,7 @@ import configparser import json import threading -from datetime import datetime +from datetime import datetime, timezone import pytest from test_cnm.main import main @@ -115,7 +115,7 @@ def cnm_response_worker(self): cnm_r = { "version": cnm_s["version"], "receivedTime": cnm_s.get("receivedTime"), - "processCompleteTime": datetime.utcnow().strftime( + "processCompleteTime": datetime.now(timezone.utc).strftime( "%Y-%m-%d %H:%M:%SZ", ), "product": { diff --git a/test-cnm/tests/test_cli/test_cmd_tidy.py b/test-cnm/tests/test_cli/test_cmd_tidy.py index 3bd12ad..9922a73 100644 --- a/test-cnm/tests/test_cli/test_cmd_tidy.py +++ b/test-cnm/tests/test_cli/test_cmd_tidy.py @@ -19,6 +19,11 @@ def test_bucket(test_bucket): metadata.put( Body=json.dumps( { + "$testconfig": { + "COLLECTION_1/PRODUCT_1": { + "foo": "bar", + }, + }, obj1.key: { "checksum": "11111111111111111111111111111111", }, @@ -55,6 +60,38 @@ def test_tidy(test_bucket): metadata_obj = test_bucket.Object("metadata.json") metadata_dict = json.loads(metadata_obj.get()["Body"].read()) + assert metadata_dict == { + "$testconfig": { + "COLLECTION_1/PRODUCT_1": { + "foo": "bar", + }, + }, + "COLLECTION_1/PRODUCT_1/file1.txt": { + "checksum": "33333333333333333333333333333333", + "type": "data", + }, + "COLLECTION_1/PRODUCT_1/file2.txt": { + "checksum": "44444444444444444444444444444444", + "type": "qa", + }, + } + + +def test_tidy_no_test_config(test_bucket): + metadata_obj = test_bucket.Object("metadata.json") + metadata_dict = json.loads(metadata_obj.get()["Body"].read()) + del metadata_dict["$testconfig"] + metadata_obj.put(Body=json.dumps(metadata_dict)) + + main(["tidy"]) + + assert [obj.key for obj in test_bucket.objects.all()] == [ + "COLLECTION_1/PRODUCT_1/file1.txt", + "COLLECTION_1/PRODUCT_1/file2.txt", + "metadata.json", + ] + metadata_dict = json.loads(metadata_obj.get()["Body"].read()) + assert metadata_dict == { "COLLECTION_1/PRODUCT_1/file1.txt": { "checksum": "33333333333333333333333333333333", @@ -79,6 +116,11 @@ def test_tidy_keep_metadata(test_bucket): metadata_dict = json.loads(metadata_obj.get()["Body"].read()) assert metadata_dict == { + "$testconfig": { + "COLLECTION_1/PRODUCT_1": { + "foo": "bar", + }, + }, "COLLECTION_1/": { "checksum": "11111111111111111111111111111111", }, diff --git a/test-cnm/tests/test_config.py b/test-cnm/tests/test_config.py index 6017786..b3fbbca 100644 --- a/test-cnm/tests/test_config.py +++ b/test-cnm/tests/test_config.py @@ -9,24 +9,60 @@ def test_from_file_default(data_path): assert config.profile == "sbx" assert config.test_bucket == "asf-cumulus-dev-tests-e2e" + assert config._options == {} + assert config._section == {} + assert config._default_section == { + "profile": "sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } def test_from_file_args_override(data_path): - args = argparse.Namespace() - args.test_bucket = "override-test-bucket" - config = ConfigBasic.from_file(data_path / "testcnm.cfg", args) + config = ConfigBasic.from_file( + data_path / "testcnm.cfg", + argparse.Namespace(test_bucket="override-test-bucket"), + ) assert config.profile == "sbx" assert config.test_bucket == "override-test-bucket" + assert config._options == { + "test_bucket": "override-test-bucket", + } + assert config._section == {} + assert config._default_section == { + "profile": "sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } def test_from_file_environment(data_path): - args = argparse.Namespace() - args.environment = "sitenv" - config = ConfigBasic.from_file(data_path / "testcnm.cfg", args) + config = ConfigBasic.from_file( + data_path / "testcnm.cfg", + argparse.Namespace(environment="sitenv"), + ) assert config.profile == "sit" assert config.test_bucket == "asf-cumulus-int-tests-e2e" + assert config._options == { + "environment": "sitenv", + } + assert config._section == { + "profile": "sit", + "test_bucket": "asf-cumulus-int-tests-e2e", + } + assert config._default_section == { + "profile": "sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } def test_from_file_inheritance(data_path): @@ -39,6 +75,116 @@ def test_from_file_inheritance(data_path): assert config.profile == "home-sbx" assert config.test_bucket == "asf-cumulus-dev-tests-e2e" + assert config._options == {} + assert config._section == {} + assert config._default_section == { + "profile": "home-sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } + + +def test_dynamic_config(data_path): + config = ConfigFull.from_file(data_path / "testcnm.cfg") + + config = config.dynamic_config({"cnm_ingest_queue": "foo-bar"}) + + assert config.profile == "sbx" + assert config.test_bucket == "asf-cumulus-dev-tests-e2e" + assert config.cnm_ingest_queue == "foo-bar" + assert config._options == {} + assert config._section == {} + assert config._default_section == { + "profile": "sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } + + +def test_dynamic_config_with_args(data_path): + config = ConfigFull.from_file( + data_path / "testcnm.cfg", + argparse.Namespace(cnm_ingest_queue="cnm-ingest-queue-from-args"), + ) + + config = config.dynamic_config({"cnm_ingest_queue": "foo-bar"}) + + assert config.profile == "sbx" + assert config.test_bucket == "asf-cumulus-dev-tests-e2e" + assert config.cnm_ingest_queue == "cnm-ingest-queue-from-args" + assert config._options == { + "cnm_ingest_queue": "cnm-ingest-queue-from-args", + } + assert config._section == {} + assert config._default_section == { + "profile": "sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } + + +def test_dynamic_config_with_env(data_path): + config = ConfigFull.from_file( + data_path / "testcnm.cfg", + argparse.Namespace(environment="sitenv"), + ) + + config = config.dynamic_config({"cnm_ingest_queue": "cnm-ingest-queue-from-dynamic"}) + + assert config.profile == "sit" + assert config.test_bucket == "asf-cumulus-int-tests-e2e" + assert config.cnm_ingest_queue == "cnm-ingest-queue-from-dynamic" + assert config._options == { + "environment": "sitenv", + } + assert config._section == { + "profile": "sit", + "test_bucket": "asf-cumulus-int-tests-e2e", + } + assert config._default_section == { + "profile": "sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } + + +def test_dynamic_config_with_env_and_args(data_path): + config = ConfigFull.from_file( + data_path / "testcnm.cfg", + argparse.Namespace( + environment="sitenv", + cnm_ingest_queue="cnm-ingest-queue-from-args", + ), + ) + + config = config.dynamic_config({"cnm_ingest_queue": "cnm-ingest-queue-from-dynamic"}) + + assert config.profile == "sit" + assert config.test_bucket == "asf-cumulus-int-tests-e2e" + assert config.cnm_ingest_queue == "cnm-ingest-queue-from-args" + assert config._options == { + "environment": "sitenv", + "cnm_ingest_queue": "cnm-ingest-queue-from-args", + } + assert config._section == { + "profile": "sit", + "test_bucket": "asf-cumulus-int-tests-e2e", + } + assert config._default_section == { + "profile": "sbx", + "test_bucket": "asf-cumulus-dev-tests-e2e", + "cnm_ingest_queue": "asf-cumulus-dev-opera-cnm-ingest-queue", + "cnm_response_queue": "asf-cumulus-dev-opera-mock-jpl-response-queue", + "provider": "JPL-OPERA", + } def test_from_file_empty(data_path): diff --git a/test-cnm/tests/test_executor.py b/test-cnm/tests/test_executor.py new file mode 100644 index 0000000..d98dcde --- /dev/null +++ b/test-cnm/tests/test_executor.py @@ -0,0 +1,398 @@ +import json +from unittest import mock + +import pytest +from test_cnm.config import ConfigFull +from test_cnm.metadata import Metadata +from test_cnm.tester.executor import TestExecutor +from test_cnm.tester.types import ExecutableTest, TestCollector, TestInfo + + +@pytest.fixture +def executor(boto_session, test_bucket, mock_make_cnm_s, ingest_queue_1, response_queue): + return TestExecutor( + session=boto_session, + collector=mock.create_autospec(TestCollector), + make_cnm_s=mock_make_cnm_s, + metadata=Metadata(boto_session, test_bucket.name), + config=ConfigFull( + test_bucket=test_bucket.name, + cnm_ingest_queue=ingest_queue_1[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ) + + +@pytest.fixture() +def ingest_queue_1(sqs_client): + queue_name = "test-cnm-ingest-queue-1" + response = sqs_client.create_queue(QueueName=queue_name) + return queue_name, response["QueueUrl"] + + +@pytest.fixture() +def ingest_queue_2(sqs_client): + queue_name = "test-cnm-ingest-queue-2" + response = sqs_client.create_queue(QueueName=queue_name) + return queue_name, response["QueueUrl"] + + +@pytest.fixture() +def response_queue(sqs_client): + queue_name = "test-cnm-response-queue" + response = sqs_client.create_queue(QueueName=queue_name) + return queue_name, response["QueueUrl"] + + +def test_collect_tests(executor, ingest_queue_1, response_queue): + executor.collector.collect_tests.return_value = { + "TEST_1": TestInfo( + collection="TEST_COLLECTION", + data_version="1.0", + name="TEST_1", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_1/TEST_1.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + ), + "TEST_2": TestInfo( + collection="TEST_COLLECTION", + data_version="1.0", + name="TEST_2", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_2/TEST_2.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + ), + } + test_run = executor.new_test_run([]) + + test_run.collect_tests() + + assert test_run.tests == [ + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_1", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_1/TEST_1.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_1[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_2", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_2/TEST_2.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_1[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ] + + +def test_iter_start_tests(executor, sqs_client, ingest_queue_1, response_queue): + test_run = executor.new_test_run([]) + + test_run.tests = [ + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_1", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_1/TEST_1.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_1[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_2", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_2/TEST_2.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_1[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ] + test_run._state = "tests_collected" + + for _ in test_run.iter_start_tests(): + pass + + assert len(test_run.pending_tests) == 2 + assert test_run.num_started == 2 + + response = sqs_client.receive_message(QueueUrl=ingest_queue_1[1], MaxNumberOfMessages=10) + messages = [json.loads(msg["Body"]) for msg in response.get("Messages", ())] + assert messages == [ + { + "identifier": "1", + "collection": "TEST_COLLECTION", + "version": "1.3", + "submissionTime": "2026-01-01T00:00:00.000Z", + "product": { + "name": "TEST_1", + "dataVersion": "1.0", + "files": [ + { + "name": "TEST_1.json", + "type": "data", + "uri": "s3://test/TEST_COLLECTION/TEST_1/TEST_1.json", + "size": 100, + "checksum": '"foobarbaz"', + "checksumType": "md5", + } + ], + }, + "provider": "test-provider", + }, + { + "identifier": "2", + "collection": "TEST_COLLECTION", + "version": "1.3", + "submissionTime": "2026-01-01T00:00:00.000Z", + "product": { + "name": "TEST_2", + "dataVersion": "1.0", + "files": [ + { + "name": "TEST_2.json", + "type": "data", + "uri": "s3://test/TEST_COLLECTION/TEST_2/TEST_2.json", + "size": 100, + "checksum": '"foobarbaz"', + "checksumType": "md5", + } + ], + }, + "provider": "test-provider", + }, + ] + + +def test_iter_start_tests_multiple_clients( + executor, + sqs_client, + ingest_queue_1, + ingest_queue_2, + response_queue, +): + test_run = executor.new_test_run([]) + + test_run.tests = [ + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_1", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_1/TEST_1.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_1[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_2", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_2/TEST_2.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_2[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ] + test_run._state = "tests_collected" + + for _ in test_run.iter_start_tests(): + pass + + assert len(test_run.pending_tests) == 2 + assert test_run.num_started == 2 + + response = sqs_client.receive_message(QueueUrl=ingest_queue_1[1], MaxNumberOfMessages=10) + messages = [json.loads(msg["Body"]) for msg in response.get("Messages", ())] + assert messages == [ + { + "identifier": "1", + "collection": "TEST_COLLECTION", + "version": "1.3", + "submissionTime": "2026-01-01T00:00:00.000Z", + "product": { + "name": "TEST_1", + "dataVersion": "1.0", + "files": [ + { + "name": "TEST_1.json", + "type": "data", + "uri": "s3://test/TEST_COLLECTION/TEST_1/TEST_1.json", + "size": 100, + "checksum": '"foobarbaz"', + "checksumType": "md5", + } + ], + }, + "provider": "test-provider", + }, + ] + + response = sqs_client.receive_message(QueueUrl=ingest_queue_2[1], MaxNumberOfMessages=10) + messages = [json.loads(msg["Body"]) for msg in response.get("Messages", ())] + assert messages == [ + { + "identifier": "2", + "collection": "TEST_COLLECTION", + "version": "1.3", + "submissionTime": "2026-01-01T00:00:00.000Z", + "product": { + "name": "TEST_2", + "dataVersion": "1.0", + "files": [ + { + "name": "TEST_2.json", + "type": "data", + "uri": "s3://test/TEST_COLLECTION/TEST_2/TEST_2.json", + "size": 100, + "checksum": '"foobarbaz"', + "checksumType": "md5", + } + ], + }, + "provider": "test-provider", + }, + ] + + +def test_iter_responses( + executor, + sqs_client, + ingest_queue_1, + ingest_queue_2, + response_queue, +): + test_run = executor.new_test_run([]) + + test_run.tests = [ + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_1", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_1/TEST_1.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_1[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ExecutableTest( + collection="TEST_COLLECTION", + data_version="1.0", + resolved_data_version="1.0", + name="TEST_2", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_2/TEST_2.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue_2[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ), + ] + test_run._state = "tests_collected" + + for _ in test_run.iter_start_tests(): + pass + + sqs_client.send_message( + QueueUrl=response_queue[1], + MessageBody=json.dumps( + { + "identifier": "1", + "submissionTime": "2026-01-01T00:00:00.000Z", + "response": { + "status": "SUCCESS", + }, + } + ), + ) + sqs_client.send_message( + QueueUrl=response_queue[1], + MessageBody=json.dumps( + { + "identifier": "2", + "submissionTime": "2026-01-01T00:00:00.000Z", + "response": { + "status": "FAILURE", + }, + } + ), + ) + + _ = list(test_run.iter_responses()) + assert test_run.num_succeeded == 1 + assert test_run.num_failed == 1 + assert not test_run.pending_tests diff --git a/test-cnm/tests/test_ingest_client.py b/test-cnm/tests/test_ingest_client.py new file mode 100644 index 0000000..cc7af28 --- /dev/null +++ b/test-cnm/tests/test_ingest_client.py @@ -0,0 +1,130 @@ +import json + +import pytest +from test_cnm.tester.ingest_client import CnmIngestClient +from test_cnm.tester.types import ExecutableTest + + +@pytest.fixture() +def ingest_queue(sqs_client): + queue_name = "test-cnm-ingest-queue" + response = sqs_client.create_queue(QueueName=queue_name) + return queue_name, response["QueueUrl"] + + +@pytest.fixture() +def response_queue(sqs_client): + queue_name = "test-cnm-response-queue" + response = sqs_client.create_queue(QueueName=queue_name) + return queue_name, response["QueueUrl"] + + +def test_submit_request(boto_session, sqs_client, mock_make_cnm_s, ingest_queue): + ingest_client = CnmIngestClient( + session=boto_session, + make_cnm_s=mock_make_cnm_s, + cnm_response_queue="", + ) + + cnm_s = ingest_client.submit_request( + ExecutableTest( + collection="TEST_COLLECTION", + data_version=None, + resolved_data_version="1.0", + name="TEST_1", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_1/TEST_1.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue[1], + cnm_response_queue="", + provider="test-provider", + ) + ) + + expected = { + "identifier": "1", + "collection": "TEST_COLLECTION", + "version": "1.3", + "submissionTime": "2026-01-01T00:00:00.000Z", + "product": { + "name": "TEST_1", + "dataVersion": "1.0", + "files": [ + { + "name": "TEST_1.json", + "type": "data", + "uri": "s3://test/TEST_COLLECTION/TEST_1/TEST_1.json", + "size": 100, + "checksum": '"foobarbaz"', + "checksumType": "md5", + } + ], + }, + "provider": "test-provider", + } + + assert cnm_s == expected + + response = sqs_client.receive_message(QueueUrl=ingest_queue[1]) + messages = [json.loads(msg["Body"]) for msg in response.get("Messages", ())] + assert messages == [expected] + + +def test_process_messages(boto_session, sqs_client, mock_make_cnm_s, ingest_queue, response_queue): + ingest_client = CnmIngestClient( + session=boto_session, + make_cnm_s=mock_make_cnm_s, + cnm_response_queue=response_queue[1], + ) + + cnm_s = ingest_client.submit_request( + ExecutableTest( + collection="TEST_COLLECTION", + data_version=None, + resolved_data_version="1.0", + name="TEST_1", + files=[ + { + "Bucket": "test", + "Key": "TEST_COLLECTION/TEST_1/TEST_1.json", + "Size": 100, + "ETag": '"foobarbaz"', + } + ], + cnm_ingest_queue=ingest_queue[1], + cnm_response_queue=response_queue[1], + provider="test-provider", + ) + ) + + assert list(ingest_client.process_messages()) == [] + assert ingest_client.is_waiting + + sqs_client.send_message( + QueueUrl=response_queue[1], + MessageBody=json.dumps( + { + "identifier": cnm_s["identifier"], + "submissionTime": cnm_s["submissionTime"], + "response": { + "status": "SUCCESS", + }, + } + ), + ) + + assert list(ingest_client.process_messages()) == [ + { + "identifier": cnm_s["identifier"], + "submissionTime": cnm_s["submissionTime"], + "response": { + "status": "SUCCESS", + }, + } + ] + assert not ingest_client.is_waiting diff --git a/test-cnm/tests/test_metadata.py b/test-cnm/tests/test_metadata.py new file mode 100644 index 0000000..c3039a3 --- /dev/null +++ b/test-cnm/tests/test_metadata.py @@ -0,0 +1,86 @@ +import json +from unittest import mock + +import pytest +from test_cnm.metadata import Metadata + + +@pytest.fixture +def metadata(boto_session, test_bucket): + return Metadata( + session=boto_session, + bucket=test_bucket.name, + ) + + +def test_load(test_bucket, metadata): + test_bucket.Object("metadata.json").put( + Body=json.dumps({"foo": {"checksum": "bar"}}), + ) + + metadata.load() + + assert metadata.metadata == { + "$testconfig": {}, + "foo": {"checksum": "bar"}, + } + + +def test_save(test_bucket, metadata): + metadata["foo"]["checksum"] = "baz" + metadata["bar"]["checksum"] = "qux" + + metadata.save() + + contents = json.load(test_bucket.Object("metadata.json").get()["Body"]) + assert contents == { + "foo": { + "checksum": "baz", + }, + "bar": { + "checksum": "qux", + }, + } + + +def test_save_when_unmodified(test_bucket, metadata): + test_bucket.Object("metadata.json").put( + Body=json.dumps({"foo": {"checksum": "bar"}}), + ) + + metadata.load() + metadata.session = mock.Mock() + metadata.save() + + metadata.session.client().upload_fileobj.assert_not_called() + + +def test_contextmanager(test_bucket, metadata): + test_bucket.Object("metadata.json").put( + Body=json.dumps({"foo": {"checksum": "bar"}}), + ) + + with metadata: + assert metadata["foo"] == {"checksum": "bar"} + + metadata["foo"]["checksum"] = "baz" + metadata["bar"]["checksum"] = "qux" + + contents = json.load(test_bucket.Object("metadata.json").get()["Body"]) + assert contents == { + "foo": { + "checksum": "baz", + }, + "bar": { + "checksum": "qux", + }, + } + + +def test_crud(metadata): + assert "foo" not in metadata + metadata["foo"]["bar"] = "baz" + assert "foo" in metadata + assert metadata["foo"] == {"bar": "baz"} + del metadata["foo"] + assert "foo" not in metadata