diff --git a/.github/workflows/flow.yml b/.github/workflows/flow.yml index 056ecdc7c..51133e7e0 100644 --- a/.github/workflows/flow.yml +++ b/.github/workflows/flow.yml @@ -33,6 +33,9 @@ jobs: matrix: which_test: [ static_flow, no_mirror, flakey_broker, dynamic_flow, restart_server, partitioned_flow ] osver: [ "ubuntu-22.04", "ubuntu-24.04" ] + # Fork proof runs use local fixtures; dynamic_flow consumes live datamarts. + exclude: + - which_test: ${{ github.repository == 'robjarawan/sarracenia' && 'dynamic_flow' || '__none__' }} runs-on: ${{ matrix.osver }} diff --git a/.github/workflows/flow_amqp_consumer.yml b/.github/workflows/flow_amqp_consumer.yml index d6f8cf864..ea2b70a59 100644 --- a/.github/workflows/flow_amqp_consumer.yml +++ b/.github/workflows/flow_amqp_consumer.yml @@ -30,6 +30,9 @@ jobs: matrix: which_test: [ static_flow, no_mirror, flakey_broker, dynamic_flow, restart_server ] osver: [ "ubuntu-22.04", "ubuntu-24.04" ] + # Fork proof runs use local fixtures; dynamic_flow consumes live datamarts. + exclude: + - which_test: ${{ github.repository == 'robjarawan/sarracenia' && 'dynamic_flow' || '__none__' }} runs-on: ${{ matrix.osver }} diff --git a/.github/workflows/flow_basic.yml b/.github/workflows/flow_basic.yml index 49649f46f..c03c79363 100644 --- a/.github/workflows/flow_basic.yml +++ b/.github/workflows/flow_basic.yml @@ -1,53 +1,46 @@ -name: sr_insects test basic declare/cleanup, and python API +name: Maintenance and Python API tests on: pull_request: - types: [opened, edited, reopened] + types: [opened, synchronize, reopened] push: paths-ignore: - - '.github/**' - 'debian/changelog' - 'TODO.txt' - - workflow_dispatch: - inputs: - debug_enabled: - type: boolean - description: 'Run the build with tmate debugging enabled (https://github.com/marketplace/actions/debugging-with-tmate)' - required: false - default: false jobs: - - run_sr_insects_tests: - + maintenance: strategy: - # Don't cancel the entire matrix when one job fails fail-fast: false matrix: - osver: [ "ubuntu-22.04", "ubuntu-24.04" ] - - runs-on: ${{ matrix.osver }} - - name: Maintenance test on ${{ matrix.osver }} - timeout-minutes: 40 - + include: + - os: ubuntu-22.04 + python: '3.10' + - os: ubuntu-24.04 + python: '3.12' + runs-on: ${{ matrix.os }} + name: Maintenance test on ${{ matrix.os }} + timeout-minutes: 10 + services: + rabbitmq: + image: rabbitmq:4-alpine + ports: + - 5672:5672 + env: + RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS: '+S 2:2 +A 2' + options: >- + --health-cmd "rabbitmq-diagnostics -q check_port_connectivity" + --health-interval 5s + --health-timeout 5s + --health-retries 12 steps: - uses: actions/checkout@v6 - - - name: Install dependencies - run: | - travis/flow_autoconfig.sh - travis/ssh_localhost.sh - - # Enable tmate debugging of manually-triggered workflows if the input option was provided - - name: Setup tmate session - uses: mxschmitt/action-tmate@v3 - if: ${{ github.event_name == 'workflow_dispatch' && inputs.debug_enabled }} - - - name: Add and Remove configs, - run: | - pwd - ls - cd ${HOME}/sr_insects/static_flow; ./flow_maint_test.sh + - uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python }} + - name: Install the package under test + run: python -m pip install '.[amqp]' + - name: Check local maintenance and Python APIs + working-directory: ${{ runner.temp }} + run: python "$GITHUB_WORKSPACE/tests/maintenance/run.py" diff --git a/.github/workflows/flow_mqtt.yml b/.github/workflows/flow_mqtt.yml index db8694304..8a538810f 100644 --- a/.github/workflows/flow_mqtt.yml +++ b/.github/workflows/flow_mqtt.yml @@ -30,6 +30,9 @@ jobs: matrix: which_test: [ static_flow, no_mirror, flakey_broker, dynamic_flow ] osver: [ "ubuntu-22.04", "ubuntu-24.04" ] + # Fork proof runs use local fixtures; dynamic_flow consumes live datamarts. + exclude: + - which_test: ${{ github.repository == 'robjarawan/sarracenia' && 'dynamic_flow' || '__none__' }} runs-on: ${{ matrix.osver }} diff --git a/sarracenia/flowcb/gather/am.py b/sarracenia/flowcb/gather/am.py index 4eeab64d1..40c37274c 100644 --- a/sarracenia/flowcb/gather/am.py +++ b/sarracenia/flowcb/gather/am.py @@ -64,7 +64,7 @@ André LeBlanc, ANL, Autumn 2022 """ -import logging, socket, struct, time, sys, os, signal, ipaddress, urllib.parse, getpass, psutil +import logging, socket, struct, time, sys, os, ipaddress, urllib.parse, getpass, psutil import re from base64 import b64encode from random import randint @@ -113,9 +113,8 @@ def __init__(self, options): self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - # Add signal handler - ## Override outer signal handler with a default one to exit correctly. - signal.signal(signal.SIGTERM, signal.SIG_DFL) + # Let the parent flow's signal handler manage graceful shutdown. + # Previously overrode with SIG_DFL which bypassed all cleanup. def __WaitForRemoteConnections__(self) -> NoReturn: diff --git a/sarracenia/flowcb/send/am.py b/sarracenia/flowcb/send/am.py index 297751a9b..cd5bf80bc 100644 --- a/sarracenia/flowcb/send/am.py +++ b/sarracenia/flowcb/send/am.py @@ -32,7 +32,7 @@ André LeBlanc, ANL, Autumn 2022 """ -import logging, socket, struct, time, signal, sys, os +import logging, socket, struct, time, sys, os import urllib.parse from sarracenia.flowcb import FlowCB @@ -63,9 +63,8 @@ def __init__(self, options): self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 2) - # Add signal handler - ## Override outer signal handler with a default one to exit correctly. - signal.signal(signal.SIGTERM, signal.SIG_DFL) + # Let the parent flow's signal handler manage graceful shutdown. + # Previously overrode with SIG_DFL which bypassed all cleanup. def wrapbulletin(self, sarra_msg): diff --git a/tests/maintenance/README.md b/tests/maintenance/README.md new file mode 100644 index 000000000..2f8a64285 --- /dev/null +++ b/tests/maintenance/README.md @@ -0,0 +1,107 @@ +# Local maintenance and Python API checks + +Author: Rob Jarawan - Data Interchange + +This check exercises the installed SR3 package with two subscriber configurations, +one one-shot post configuration, five generated files, and a disposable local +RabbitMQ. It replaces the maintenance job's downloads of examples from upstream +branch names and its dependence on public data feeds. + +The separate static/protocol flow suites still cover the wider flow fixture set, +including C components. This focused Python maintenance check does not replace +those suites or claim a complete flow/platform verification gate. + +## What is checked + +| Operation | Required result | +| --- | --- | +| `sr3 add` | Each generated configuration is copied without changes | +| `sr3 declare` | Both named queues exist and start empty | +| Moth publisher/subscriber API | Five distinct products arrive with the expected sizes and identities; each is acknowledged | +| Subscribe Flow API | A bounded worker downloads exactly the five complete files and drains its queue | +| `sr3 cleanup` | Both queues disappear; the selected count includes the one-shot post configuration | +| `sr3 remove` | All three generated configuration files disappear | + +SR3 can return zero after logging a refused maintenance action. The check therefore +verifies broker/filesystem outcomes as well as command exit status. An empty poll +is not counted as a delivered message. It rejects execution against a source-tree +import and prints the installed module path and Python version. + +The fixture finalizes the added configurations through the Python API before +running maintenance commands, including creation of their private cache directories. +This check does not claim coverage of startup with missing cache directories. + +## Reproduce locally + +Prerequisites: Docker and public package/image downloads. Run from this PR's +checkout. No host home directory, credentials, or operational configuration is +mounted. The test creates its own private XDG directories. + +Build an image containing the installed package and the matching test script: + +```bash +docker build -t sr-maintenance-check -f - . <<'DOCKERFILE' +FROM python:3.10-slim +WORKDIR /package +COPY . . +RUN pip install --no-cache-dir '.[amqp]' +RUN useradd --create-home maintenance +USER maintenance +WORKDIR /tmp +CMD ["python", "/package/tests/maintenance/run.py"] +DOCKERFILE +``` + +Start a dedicated broker without external networking or published ports: + +```bash +docker run -d --name sr-maintenance-check-broker --network none --memory 384m --cpus 1 \ + -e 'RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=+S 2:2 +A 2' rabbitmq:4-alpine +docker exec sr-maintenance-check-broker rabbitmq-diagnostics -q check_port_connectivity +``` + +Wait until the readiness check succeeds. Then run the check: + +```bash +docker run --rm --network container:sr-maintenance-check-broker --memory 512m --cpus 1 \ + --pids-limit 128 sr-maintenance-check +``` + +Success ends with a JSON `PASS` result naming the five products and reporting +three removed configurations and two removed queues. The workflow runs this same +script against the package installed on Ubuntu 22.04/Python 3.10 and +Ubuntu 24.04/Python 3.12, using the runner's local RabbitMQ service. + +## Failure controls + +Each command below is expected to exit nonzero. Run them separately from the +successful case so an expected failure does not stop the comparison. + +```bash +docker run --rm --network container:sr-maintenance-check-broker --memory 512m --cpus 1 \ + sr-maintenance-check python /package/tests/maintenance/run.py --publish-count 4 +docker run --rm --network container:sr-maintenance-check-broker --memory 512m --cpus 1 \ + sr-maintenance-check python /package/tests/maintenance/run.py --legacy-cleanup-count +``` + +The first case must reject four delivered products instead of counting empty polls +as the fifth. The second reproduces the former `sr3 status | grep stop` count +against the actual CLI and must detect the queue left after refused cleanup. +The failed CI's HTTP 404 path is removed entirely: neither GitHub branch names +nor downloaded examples participate in this test. + +## Cleanup and rollback + +The script removes its private files and only its uniquely named broker resources, +including after an assertion failure. CLI and Flow subprocesses have deadlines; +the workflow also has a ten-minute timeout. Stop/remove the dedicated broker after +the runs: + +```bash +docker stop sr-maintenance-check-broker +docker rm sr-maintenance-check-broker +``` + +Reverting the isolated CI commit restores the old runner; it does not change SR3 +runtime code or deploy anything. Keep the observed logs before removing images. +Contact: Rob Jarawan. Related fork issue: #139. diff --git a/tests/maintenance/run.py b/tests/maintenance/run.py new file mode 100644 index 000000000..d27a870f6 --- /dev/null +++ b/tests/maintenance/run.py @@ -0,0 +1,234 @@ +"""Installed-package maintenance/API checks using only a disposable local RabbitMQ.""" + +import argparse +import hashlib +import json +import logging +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import time +import uuid + + +EXPECTED_COUNT = 5 +BROKER = 'amqp://guest:guest@127.0.0.1/' # Disposable RabbitMQ's built-in local test account. + + +def isolate(root): + """Keep both Python API and CLI calls away from the invoking user's configuration.""" + for kind in ('CONFIG', 'CACHE', 'DATA'): + path = root / kind.lower() + path.mkdir(exist_ok=True) + os.environ['XDG_' + kind + '_HOME'] = str(path) + for name in ('SR_DEV_APPNAME', 'SARRA_LIB', 'SARRAC_LIB', 'PYTHONPATH'): + os.environ.pop(name, None) + os.chdir(root) + + +def cli(*args): + result = subprocess.run(['sr3'] + list(args), stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, universal_newlines=True, timeout=60) + print(result.stdout, end='', flush=True) + assert result.returncode == 0, 'sr3 command failed: ' + ' '.join(args) + return result.stdout + + +def configuration(root, component, name): + from sarracenia.config import no_file_config + + cfg = no_file_config() + cfg.component, cfg.config, cfg.no = component, name, 1 + cfg.action = 'foreground' + state = root / 'cache' / 'sr3' / component / name + state.mkdir(parents=True, exist_ok=True) + cfg.cfg_run_dir = str(state) + cfg.metricsFilename = str(state / 'metrics.json') + cfg.pid_filename = str(state / 'instance.pid') + cfg.novipFilename = str(state / 'novip') + cfg.parse_file(str(root / 'config' / 'sr3' / component / (name + '.conf')), component=component) + cfg.finalize(component, name) + return cfg + + +def flow_worker(root): + from sarracenia.flow.subscribe import Subscribe + + Subscribe(configuration(root, 'subscribe', 'maintenance_flow')).run() + + +def fixtures(root, exchange, queues): + source, destination = root / 'source', root / 'destination' + source.mkdir() + destination.mkdir() + expected = {} + for index in range(EXPECTED_COUNT): + name = 'message-{}.txt'.format(index) + payload = 'maintenance fixture {}\n'.format(index).encode('ascii') + (source / name).write_bytes(payload) + expected[name] = payload + + common = ('broker ' + BROKER + '\nexchange ' + exchange + '\ntopicPrefix v03\n' + 'durable True\nauto_delete False\nexpire 0\ninstances 1\nbatch 5\ntimeout 15\n' + 'messageAgeMax 0\nmessageCountMax 5\nsleep 0.1\nretry_ttl 0\n' + 'inflight None\naccelThreshold 0\npermCopy False\nmirror False\n' + 'directory ' + str(destination) + '\naccept .*\n') + configs = { + 'subscribe/maintenance_moth': common + 'queueName ' + queues[0] + '\ndownload False\nsubtopic #\n', + 'subscribe/maintenance_flow': common + 'queueName ' + queues[1] + '\ndownload True\nsubtopic #\n', + 'post/maintenance_post': ('post_broker ' + BROKER + '\npost_exchange ' + exchange + '\n' + 'post_baseUrl file:' + str(source) + '\npost_baseDir ' + str(source) + '\n' + 'post_topicPrefix v03\ndurable True\nauto_delete False\n' + 'messageAgeMax 0\ntimeout 15\n'), + } + for name, content in configs.items(): + path = root / 'fixtures' / (name + '.conf') + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content) + cli('add', str(path)) + installed = root / 'config' / 'sr3' / (name + '.conf') + assert installed.read_text() == content, 'sr3 add did not preserve ' + name + component, config_name = name.split('/') + configuration(root, component, config_name) + return list(configs), expected + + +def publish(root, count, expected): + import sarracenia + from sarracenia.moth import Moth + + cfg = configuration(root, 'post', 'maintenance_post') + props = cfg.dictify() + props.update(cfg.publishers[0]) + props['publisher_index'] = 0 + publisher = Moth.pubFactory(props) + try: + for name in list(expected)[:count]: + message = sarracenia.Message() + message.update(baseUrl='file:' + str(root / 'source'), relPath=name, + pubTime=sarracenia.nowstr(), size=len(expected[name]), + identity={'method': 'sha512', 'value': hashlib.sha512(expected[name]).hexdigest()}) + assert publisher.putNewMessage(message), 'publication failed for ' + name + finally: + publisher.close() + + +def consume(root, expected): + from sarracenia.moth import Moth + + props = configuration(root, 'subscribe', 'maintenance_moth').dictify() + props['subscription_index'] = 0 + consumer = Moth.subFactory(props) + received = [] + deadline = time.monotonic() + 10 + try: + while len(received) < EXPECTED_COUNT and time.monotonic() < deadline: + message = consumer.getNewMessage() + if message is None: + time.sleep(0.05) + continue + name = message['relPath'] + assert name in expected, 'unexpected product: ' + name + assert message['size'] == len(expected[name]), 'incorrect size for ' + name + assert message['identity']['value'] == hashlib.sha512(expected[name]).hexdigest() + assert consumer.ack(message), 'acknowledgement failed for ' + name + received.append(name) + assert sorted(received) == sorted(expected), 'expected five distinct deliveries, got ' + repr(received) + assert consumer.getNewMessage() is None, 'unexpected extra delivery' + finally: + consumer.close() + return received + + +def queue_state(connection, name): + from amqp.exceptions import NotFound + + channel = connection.channel() + try: + return channel.queue_declare(name, passive=True) + except NotFound: + return None + finally: + if channel.is_open: + channel.close() + + +def run(root, count, legacy_count): + import amqp + import sarracenia + + checkout = Path(__file__).resolve().parents[2] + module = Path(sarracenia.__file__).resolve() + assert checkout not in module.parents, 'run against the installed package, outside the checkout' + print(json.dumps({'module': str(module), 'python': sys.version}), flush=True) + prefix = 'sr-maintenance-' + uuid.uuid4().hex + queues = [prefix + '-moth', prefix + '-flow'] + connection = amqp.Connection('127.0.0.1', userid='guest', password='guest', + virtual_host='/', connect_timeout=15, read_timeout=15, write_timeout=15) + connection.connect() + configs = [] + try: + configs, expected = fixtures(root, prefix, queues) + cli('--dangerWillRobinson', str(len(configs)), 'declare', *configs) + for queue in queues: + state = queue_state(connection, queue) + assert state is not None and state.message_count == 0, 'declare failed for ' + queue + publish(root, count, expected) + received = consume(root, expected) + for queue in queues: + state = queue_state(connection, queue) + expected_count = 0 if queue == queues[0] else EXPECTED_COUNT + assert state.message_count == expected_count, 'unexpected queued count for ' + queue + worker = subprocess.run([sys.executable, str(Path(__file__).resolve()), '--flow-worker', str(root)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + universal_newlines=True, timeout=60) + print(worker.stdout, end='', flush=True) + assert worker.returncode == 0, 'Flow API worker failed' + actual = {path.name: path.read_bytes() for path in (root / 'destination').iterdir() if path.is_file()} + assert actual == expected, 'Flow did not download exactly the five complete fixture files' + assert queue_state(connection, queues[1]).message_count == 0, 'Flow left deliveries queued' + cli('stop', *configs) + status = cli('status', *configs) + selected = sum('stop' in line for line in status.splitlines()) if legacy_count else len(configs) + cli('--dangerWillRobinson', str(selected), 'cleanup', *configs) + for queue in queues: + assert queue_state(connection, queue) is None, 'cleanup left queue ' + queue + cli('--dangerWillRobinson', str(len(configs)), 'remove', *configs) + assert not list((root / 'config' / 'sr3').glob('*/*.conf')), 'remove left configurations behind' + print(json.dumps({'result': 'PASS', 'received': received, 'downloaded': sorted(actual), + 'configurations_removed': len(configs), 'queues_removed': len(queues)}), flush=True) + finally: + # Only fixture-owned resources are touched, including on assertion/timeout failure. + channel = connection.channel() + try: + for queue in queues: + channel.queue_delete(queue) + channel.exchange_delete(prefix) + finally: + channel.close() + connection.close() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--flow-worker', type=Path, help=argparse.SUPPRESS) + parser.add_argument('--publish-count', type=int, choices=range(0, EXPECTED_COUNT + 1), default=EXPECTED_COUNT, + help='publish fewer than five only to verify the missing-message failure control') + parser.add_argument('--legacy-cleanup-count', action='store_true', + help='reproduce the old stopped-row cleanup count; expected to fail') + args = parser.parse_args() + logging.basicConfig(level=logging.INFO) + if args.flow_worker: + isolate(args.flow_worker) + flow_worker(args.flow_worker) + return + with tempfile.TemporaryDirectory(prefix='sr-maintenance-') as temporary: + root = Path(temporary) + isolate(root) + run(root, args.publish_count, args.legacy_cleanup_count) + + +if __name__ == '__main__': + main() diff --git a/tests/sarracenia/flowcb/am_signal_test.py b/tests/sarracenia/flowcb/am_signal_test.py new file mode 100644 index 000000000..05daf3af9 --- /dev/null +++ b/tests/sarracenia/flowcb/am_signal_test.py @@ -0,0 +1,105 @@ +import copy +import multiprocessing +import os +import signal +import sys +import time + +import pytest + +import sarracenia.config +from sarracenia.flowcb.gather.am import Am as GatherAm +from sarracenia.flowcb.send.am import Am as SendAm +from sarracenia.instance import instance + + +class _RunningFlow: + def __init__(self): + self.stop_requested = False + + def stop_request(self): + self.stop_requested = True + + +def _make_options(name): + options = copy.deepcopy(sarracenia.config.default_config()) + options.component = 'flow' + options.config = name + options.sendTo = 'am://127.0.0.1:5005' + options.fileSizeMax = 0 + options.no = 1 + return options + + +def _signal_child(callback_name, connection): + controller = object.__new__(instance) + controller.o = _make_options(callback_name) + controller.running_instance = _RunningFlow() + signal.signal(signal.SIGTERM, controller.stop_signal) + + callback_class = SendAm if callback_name == 'send' else GatherAm + callback = callback_class(controller.o) + connection.send({ + 'ready': True, + 'parent_handler_preserved': signal.getsignal(signal.SIGTERM) == controller.stop_signal, + }) + + deadline = time.monotonic() + 2 + while not controller.running_instance.stop_requested and time.monotonic() < deadline: + time.sleep(0.01) + + if not controller.running_instance.stop_requested: + connection.send({'stop_requested': False}) + connection.close() + return + + system_exit = None + try: + callback.on_stop() + except SystemExit as ex: + system_exit = ex.code + connection.send({ + 'stop_requested': True, + 'socket_closed': callback.s.fileno() == -1, + 'system_exit': system_exit, + }) + connection.close() + + +def _run_signal_case(callback_name): + parent, child_connection = multiprocessing.Pipe() + child = multiprocessing.Process( + target=_signal_child, + args=(callback_name, child_connection), + ) + child.start() + child_connection.close() + + ready = parent.recv() if parent.poll(3) else {'ready': False} + if ready.get('ready'): + os.kill(child.pid, signal.SIGTERM) + + stopped = None + if parent.poll(3): + try: + stopped = parent.recv() + except EOFError: + pass + child.join(3) + if child.is_alive(): + child.terminate() + child.join(2) + parent.close() + return ready, stopped, child.exitcode + + +@pytest.mark.skipif(sys.platform == 'win32', reason='AM gather uses POSIX process handling') +@pytest.mark.parametrize('callback_name', ['gather', 'send']) +def test_am_callback_preserves_instance_sigterm_handler(callback_name): + ready, stopped, exitcode = _run_signal_case(callback_name) + + assert ready == {'ready': True, 'parent_handler_preserved': True} + assert stopped is not None + assert stopped['stop_requested'] is True + assert stopped['socket_closed'] is True + assert exitcode == 0