From 0cc2b304d3470ed33a33ba4881dba41f7cc19d6f Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Mon, 19 Jan 2026 22:53:11 +0000 Subject: [PATCH 01/10] testing with image --- pyproject.toml | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 pyproject.toml diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..14bdd94 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,35 @@ +[build-system] +requires = [ + "setuptools>=61", + "wheel", + "versioneer[toml]" +] +build-backend = "setuptools.build_meta" + +[project] +name = "fetch" +description = "Automatic data" +requires-python = ">=3.7" + +authors = [ + { name = "Jeremy Hooke", email = "jeremy.hooke@ga.gov.au" } +] + +dynamic = ["version"] + +dependencies = [ + "arrow", + "croniter", + "feedparser", + "lxml", + "pyyaml", + "requests>=2.21.0", + "setproctitle; sys_platform == 'linux'", +] + +[project.optional-dependencies] +ecmwf = ["ecmwf-api-client"] + +[project.scripts] +fetch-service = "fetch.scripts.service:main" +fetch-now = "fetch.scripts.now:main" From 473d9f91cba161a45e065f78bd4c75b2cfeaa814 Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Mon, 19 Jan 2026 23:21:05 +0000 Subject: [PATCH 02/10] hacky solution --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 14bdd94..2f7fd68 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,7 @@ dependencies = [ "lxml", "pyyaml", "requests>=2.21.0", + "fiona==1.10.1", "setproctitle; sys_platform == 'linux'", ] From b530606912dddae2c0dc29fdc3aed8e7e62982a9 Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Tue, 20 Jan 2026 19:57:30 +0000 Subject: [PATCH 03/10] write to S3 and some lcean up --- example-config.yaml | 4 +++- fetch/_core.py | 31 +++++++++++++------------------ fetch/s3.py | 41 +++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- 4 files changed, 58 insertions(+), 20 deletions(-) create mode 100644 fetch/s3.py diff --git a/example-config.yaml b/example-config.yaml index 3f54a63..94d5a5c 100755 --- a/example-config.yaml +++ b/example-config.yaml @@ -190,7 +190,9 @@ rules: # Convert files to tiff (from netCDF) process: !shell command: '/usr/local/bin/gdal_translate -a_srs "+proj=latlong +datum=WGS84" {parent_dir}/{filename} {parent_dir}/{file_stem}.tif' - expect_file: '{parent_dir}/{file_stem}.tif' + upload_dir: '/data/water_vapour' + bucket: 'ard-processing-data' + prefix: 'ancillary/water_vapour' # #------------------------------- # Examples using the ECMWF API | diff --git a/fetch/_core.py b/fetch/_core.py index e2c3d84..67a154f 100644 --- a/fetch/_core.py +++ b/fetch/_core.py @@ -14,6 +14,7 @@ import socket import subprocess import tempfile +import boto3 from email.mime.text import MIMEText from email.header import Header @@ -21,6 +22,7 @@ from typing import Callable from .util import rsync, Uri +from . import s3 _log = logging.getLogger(__name__) @@ -560,11 +562,12 @@ class ShellFileProcessor(FileProcessor): :type command: str """ - def __init__(self, command=None, expect_file=None, input_files=None): + def __init__(self, command=None, upload_dir=None, bucket=None, prefix=None): super(ShellFileProcessor, self).__init__() self.command = command - self.expect_file = expect_file - self.input_files = input_files + self.upload_dir = upload_dir + self.bucket = bucket + self.prefix = prefix def _apply_file_pattern(self, pattern, file_path, **keywords): """ @@ -609,20 +612,7 @@ def process(self, file_path): :raises: FileProcessError """ command = self.command - if self.input_files: - path_transform = RegexpOutputPathTransform(self.input_files[0]) - if not all([os.path.isfile(path_transform.transform_output_path(f, file_path)) - for f in self.input_files[1]]): - _log.info('Not all of the required_files are present.') - # This is used for reporting, so it is returning the file_path. - return file_path - else: - # format the path based on the group from - # transform output path - # command = path_transform.transform_output_path(command) - required_files_formating = path_transform.last_matched_groups - else: - required_files_formating = {} + required_files_formating = {} command = self._apply_file_pattern(command, file_path, **required_files_formating) _log.info('Running %r', command) @@ -632,10 +622,15 @@ def process(self, file_path): raise FileProcessError('Return code %r from command %r' % (returned, command)) # Check that output exists - expected_path = self._apply_file_pattern(self.expect_file, file_path, **required_files_formating) + # TODO expect_file = self.upload_dir + '/{file_stem}.h5' + expect_file = self.upload_dir + '/{filename}' + expected_path = self._apply_file_pattern(expect_file, file_path, **required_files_formating) if not os.path.exists(expected_path): raise FileProcessError('Expected output not found {!r} for command {!r}'.format(expected_path, command)) + s3.upload(expected_path, self.bucket, self.prefix) + _log.debug('File available %r', expected_path) + return expected_path diff --git a/fetch/s3.py b/fetch/s3.py new file mode 100644 index 0000000..f304bcd --- /dev/null +++ b/fetch/s3.py @@ -0,0 +1,41 @@ +import logging +import os +import boto3 + +from pathlib import Path +from typing import Callable + +from .util import rsync, Uri + +_log = logging.getLogger(__name__) + +class FileUploadError(Exception): + """ + An error in file processing. + """ + pass + +def upload(filepath: str, bucket: str, prefix: str): + """ + Upload all files in input_file_dir to S3. + + :raises: FileUploadError + """ + _log.info(f"Uploading file {filepath} to s3://{bucket}/{prefix}") + + s3 = boto3.client("s3") + + if not os.path.exists(filepath): + raise FileUploadError(f"File does not exist: {filepath}") + + directory, filename = os.path.split(filepath) + + s3_key = f"{prefix}/{filename}" + + try: + _log.debug(f"Uploading {filename} to s3://{bucket}/{s3_key}") + s3.upload_file(filepath, bucket, s3_key) + except Exception as e: + raise FileUploadError(f"Failed to upload {filename} to s3://{bucket}/{s3_key}") + + _log.info(f"Successfully Uploaded {filename} to s3://{bucket}/{s3_key}") diff --git a/pyproject.toml b/pyproject.toml index 2f7fd68..6e3ed41 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,7 @@ dependencies = [ "lxml", "pyyaml", "requests>=2.21.0", - "fiona==1.10.1", + "boto3", "setproctitle; sys_platform == 'linux'", ] From 7ce9a4223bdefd05aae59ba1c6d8b5d8d36f6743 Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Tue, 20 Jan 2026 19:59:35 +0000 Subject: [PATCH 04/10] missed testing edit --- fetch/_core.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/fetch/_core.py b/fetch/_core.py index 67a154f..33fea87 100644 --- a/fetch/_core.py +++ b/fetch/_core.py @@ -622,8 +622,7 @@ def process(self, file_path): raise FileProcessError('Return code %r from command %r' % (returned, command)) # Check that output exists - # TODO expect_file = self.upload_dir + '/{file_stem}.h5' - expect_file = self.upload_dir + '/{filename}' + expect_file = self.upload_dir + '/{file_stem}.h5' expected_path = self._apply_file_pattern(expect_file, file_path, **required_files_formating) if not os.path.exists(expected_path): From 51775557714f2417679b2d7ef8c79e4adc490bfe Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Wed, 21 Jan 2026 00:32:40 +0000 Subject: [PATCH 05/10] unit tests --- fetch/_core.py | 17 ++--- fetch/s3.py | 12 +--- pyproject.toml | 4 ++ test/test_load.py | 4 +- test/test_s3.py | 69 ++++++++++++++++++++ test/test_shellfileprocessor.py | 112 +++++++++++++++++++++++++------- 6 files changed, 176 insertions(+), 42 deletions(-) create mode 100644 test/test_s3.py diff --git a/fetch/_core.py b/fetch/_core.py index 33fea87..53a9967 100644 --- a/fetch/_core.py +++ b/fetch/_core.py @@ -562,21 +562,21 @@ class ShellFileProcessor(FileProcessor): :type command: str """ - def __init__(self, command=None, upload_dir=None, bucket=None, prefix=None): + def __init__(self, command, upload_dir, bucket, prefix): super(ShellFileProcessor, self).__init__() self.command = command self.upload_dir = upload_dir self.bucket = bucket self.prefix = prefix - def _apply_file_pattern(self, pattern, file_path, **keywords): + def _apply_file_pattern(self, pattern, file_path): """ Format the given pattern. :type file_path: str :rtype: str - >>> p = ShellFileProcessor() + >>> p = ShellFileProcessor('run command on file {filename}', '/data/upload', 's3-bucket', 's3-prefix') >>> p._apply_file_pattern('{file_stem} extension {file_suffix}', '/tmp/something.txt') 'something extension .txt' >>> p._apply_file_pattern('{filename} in {parent_dir}', '/tmp/something.txt') @@ -585,8 +585,7 @@ def _apply_file_pattern(self, pattern, file_path, **keywords): '/tmp' >>> p._apply_file_pattern('{parent_dirs[1]}', '/tmp/something.txt') '/' - >>> p._apply_file_pattern('{base}.hdf', '/tmp/something.hdf',**{'base':'/tmp/something'}) - '/tmp/something.hdf' + """ path = Path(file_path) return pattern.format( @@ -601,8 +600,7 @@ def _apply_file_pattern(self, pattern, file_path, **keywords): parent_dirs=[str(p) for p in path.parents], # A more flexible alternative to the above. - path=path, - **keywords + path=path ) def process(self, file_path): @@ -612,8 +610,7 @@ def process(self, file_path): :raises: FileProcessError """ command = self.command - required_files_formating = {} - command = self._apply_file_pattern(command, file_path, **required_files_formating) + command = self._apply_file_pattern(command, file_path) _log.info('Running %r', command) # Trigger command @@ -623,7 +620,7 @@ def process(self, file_path): # Check that output exists expect_file = self.upload_dir + '/{file_stem}.h5' - expected_path = self._apply_file_pattern(expect_file, file_path, **required_files_formating) + expected_path = self._apply_file_pattern(expect_file, file_path) if not os.path.exists(expected_path): raise FileProcessError('Expected output not found {!r} for command {!r}'.format(expected_path, command)) diff --git a/fetch/s3.py b/fetch/s3.py index f304bcd..54705c0 100644 --- a/fetch/s3.py +++ b/fetch/s3.py @@ -2,12 +2,8 @@ import os import boto3 -from pathlib import Path -from typing import Callable - -from .util import rsync, Uri - _log = logging.getLogger(__name__) +_s3 = boto3.client("s3") class FileUploadError(Exception): """ @@ -23,8 +19,6 @@ def upload(filepath: str, bucket: str, prefix: str): """ _log.info(f"Uploading file {filepath} to s3://{bucket}/{prefix}") - s3 = boto3.client("s3") - if not os.path.exists(filepath): raise FileUploadError(f"File does not exist: {filepath}") @@ -34,8 +28,8 @@ def upload(filepath: str, bucket: str, prefix: str): try: _log.debug(f"Uploading {filename} to s3://{bucket}/{s3_key}") - s3.upload_file(filepath, bucket, s3_key) + _s3.upload_file(filepath, bucket, s3_key) except Exception as e: - raise FileUploadError(f"Failed to upload {filename} to s3://{bucket}/{s3_key}") + raise FileUploadError(f"Failed to upload {filename} to s3://{bucket}/{s3_key}") from e _log.info(f"Successfully Uploaded {filename} to s3://{bucket}/{s3_key}") diff --git a/pyproject.toml b/pyproject.toml index 6e3ed41..d98ee87 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,10 @@ dependencies = [ "pyyaml", "requests>=2.21.0", "boto3", + "pytest", + "mock", + "pylint", + "pep8", "setproctitle; sys_platform == 'linux'", ] diff --git a/test/test_load.py b/test/test_load.py index a211e45..386b9f5 100644 --- a/test/test_load.py +++ b/test/test_load.py @@ -193,7 +193,9 @@ def _make_config(): 'process': ShellFileProcessor( command='/usr/local/bin/gdal_translate -a_srs "+proj=latlong +datum=WGS84" ' '{parent_dir}/{file_stem}.nc {parent_dir}/{file_stem}.tif', - expect_file='{parent_dir}/{file_stem}.tif' + upload_dir= '{parent_dir}', + bucket= 'ard-processing-data', + prefix= 'ancillary/water_vapour' ) }, 'NPP GDAS-forecast': { diff --git a/test/test_s3.py b/test/test_s3.py new file mode 100644 index 0000000..64d2a05 --- /dev/null +++ b/test/test_s3.py @@ -0,0 +1,69 @@ +""" +A package for testing the code in s3.py +""" +import os +from pathlib import Path +from fetch import s3 +from fetch.s3 import upload, FileUploadError + +import pytest +from unittest.mock import patch, MagicMock + +# Constants for tests +filepath = "/tmp/test.txt" +bucket = "my-bucket" +prefix = "my/prefix" + +def test_upload_success(): + # Set up mocks + with ( + patch("os.path.exists", return_value=True), + patch.object(s3, "_s3") as mock_s3, + patch.object(s3, "_log") as mock_log, + ): + # Call code + upload(filepath, bucket, prefix) + + # Expected logged messages + mock_log.info.assert_any_call("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix") + mock_log.debug.assert_any_call("Uploading test.txt to s3://my-bucket/my/prefix/test.txt") + mock_log.info.assert_any_call("Successfully Uploaded test.txt to s3://my-bucket/my/prefix/test.txt") + + # expected write to S3 + mock_s3.upload_file.assert_called_once_with( + filepath, + bucket, + "my/prefix/test.txt", + ) + +def test_given_file_not_exist(): + # Set up mocks + with ( + patch("os.path.exists", return_value=False), + patch.object(s3, "_s3") as mock_s3, + patch.object(s3, "_log") as mock_log, + ): + # Run code and confirm we get an expection back, checking type & message + with pytest.raises(FileUploadError, match="File does not exist: /tmp/test.txt"): + upload(filepath, bucket, prefix) + + # Expected logged messages + mock_log.info.assert_called_once_with("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix") + mock_s3.upload_file.assert_not_called() + +def test_write_to_s3_failed(): + # Set up mocks + with ( + patch("os.path.exists", return_value=True), + patch.object(s3, "_s3") as mock_s3, + patch.object(s3, "_log") as mock_log, + ): + mock_s3.upload_file.side_effect = Exception("poop") + + # Run code and confirm we get an expection back, checking type & message + with pytest.raises(FileUploadError, match="Failed to upload test.txt to s3://my-bucket/my/prefix/test.txt"): + upload(filepath, bucket, prefix) + + # Expected logged messages + mock_log.info.assert_called_once_with("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix") + mock_log.debug.assert_called_once_with("Uploading test.txt to s3://my-bucket/my/prefix/test.txt") diff --git a/test/test_shellfileprocessor.py b/test/test_shellfileprocessor.py index db0b0ce..2198cd2 100644 --- a/test/test_shellfileprocessor.py +++ b/test/test_shellfileprocessor.py @@ -1,31 +1,99 @@ - +""" +A package for testing the code in the ShellFileProcessor class, in _core.py +""" import os +import subprocess +import pytest from pathlib import Path +from unittest.mock import patch, MagicMock + +from fetch import s3 +from fetch import _core +from fetch._core import ShellFileProcessor, FileProcessError + +# Test Constants +command="run command on file {filename}" +upload_dir="/data/upload" +bucket="s3-bucket" +prefix="s3-prefix" +input_file="/data/staging/file.txt" + +def test_apply_file_pattern_basic(): + p = ShellFileProcessor( + command=command, + upload_dir=upload_dir, + bucket=bucket, + prefix=prefix, + ) + + result = p._apply_file_pattern(command, "/tmp/something.txt") + + assert result == "run command on file something.txt" + +def test_process_success(): + # Set up mocks + with ( + patch("subprocess.call", return_value=0), + patch("os.path.exists", return_value=True), + patch.object(s3, "_s3") as mock_s3, + patch.object(_core, "_log") as mock_log + ): + p = ShellFileProcessor( + command=command, + upload_dir=upload_dir, + bucket=bucket, + prefix=prefix, + ) + + # assert success using return value (and no raised errors) + assert p.process(input_file) == "/data/upload/file.h5" -from fetch._core import ShellFileProcessor + # confirm the correct messages are logged + mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt') + mock_log.debug.assert_called_once_with('File available %r', '/data/upload/file.h5') +def test_process_command_failed(): + # Set up mocks + with ( + patch("subprocess.call", return_value=1), + patch("os.path.exists", return_value=True), + patch.object(s3, "_s3") as mock_s3, + patch.object(_core, "_log") as mock_log + ): + p = ShellFileProcessor( + command=command, + upload_dir=upload_dir, + bucket=bucket, + prefix=prefix, + ) -def test_shellfilepro_required_files_there(): - command = 'ls {base}.py' - # required_files = (r'^(?P.*test.+)\.py$',['{base}.py','{base}.py']) - required_files = ('^(?P.*test.+)\\.py$', ['{base}.py', '{base}.py']) - file_path = os.path.abspath(__file__) - expect_file = '{base}.py' - sfp = ShellFileProcessor(command=command, expect_file=expect_file, input_files=required_files) - results = sfp.process(file_path) + # Run code and confirm we get an expection back, checking type & message + with pytest.raises(FileProcessError, match="Return code 1 from command 'run command on file file.txt'"): + p.process(input_file) - assert results == Path(__file__).absolute().as_posix() + # confirm the correct messages are logged + mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt') + mock_log.debug.assert_not_called() +def test_process_output_file_missing(): + # Set up mocks + with ( + patch("subprocess.call", return_value=0), + patch("os.path.exists", return_value=False), + patch.object(s3, "_s3") as mock_s3, + patch.object(_core, "_log") as mock_log + ): + p = ShellFileProcessor( + command=command, + upload_dir=upload_dir, + bucket=bucket, + prefix=prefix, + ) -def test_shellfilepro_required_files_not_there(): - command = 'ls {base}.py' - required_files = (r'^(?P.*test.+)\.py$', ['{base}.py', '/this/is/not/here/please.py']) - file_path = os.path.abspath(__file__) - sfp = ShellFileProcessor(command=command, expect_file=file_path, input_files=required_files) - results = sfp.process(file_path) - # A rather toothless assert given required_files_there would return the same result... - assert file_path == results - # Future option is to ask for a tmp_path, and make your command touch {tmp_path / 'test_file.txt'}. - # Then if the file exists, you know the command was run. It could work in both tests. + # Run code and confirm we get an expection back, checking type & message + with pytest.raises(FileProcessError, match="Expected output not found '/data/upload/file.h5' for command 'run command on file file.txt'"): + p.process(input_file) -# Add a test when no required_files parameter is supplied at all + # confirm the correct messages are logged + mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt') + mock_log.debug.assert_not_called() \ No newline at end of file From d4cbace846c40e6dc54220f9b561a2a25f76cf5c Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Thu, 22 Jan 2026 19:54:56 +0000 Subject: [PATCH 06/10] loging change checkpoint --- example-config.yaml | 4 --- fetch/_core.py | 78 +-------------------------------------------- fetch/auto.py | 37 ++------------------- fetch/ecmwf.py | 1 - fetch/load.py | 15 +-------- test/test_ecmwf.py | 3 -- test/test_load.py | 3 -- 7 files changed, 5 insertions(+), 136 deletions(-) diff --git a/example-config.yaml b/example-config.yaml index 94d5a5c..1c125e7 100755 --- a/example-config.yaml +++ b/example-config.yaml @@ -7,10 +7,6 @@ # The work directory (for log and lock files): directory: /data/fetch -# Notification settings (for errors): -notify: - email: ['jeremy.hooke@ga.gov.au'] - # Message bus config, for announcing arrivals. Optional. messaging: host: rhe-neo-dev01.dev.lan diff --git a/fetch/_core.py b/fetch/_core.py index 53a9967..9dcda3e 100644 --- a/fetch/_core.py +++ b/fetch/_core.py @@ -7,16 +7,10 @@ import datetime import errno import logging -import multiprocessing import os import re -import smtplib -import socket import subprocess import tempfile -import boto3 -from email.mime.text import MIMEText -from email.header import Header from pathlib import Path from typing import Callable @@ -470,76 +464,6 @@ def on_process_failure(self, process): pass -class TaskFailureEmailer(TaskFailureListener): - """ - Send failure information via email - """ - - def __init__(self, addresses): - """ - :type addresses: list of str - """ - self.addresses = addresses - - def on_file_failure(self, process_name, file_uri, summary, body_text): - """ - Send mail on a - :param process_name: - :param file_uri: - :param summary: - :param body_text: - :return: - """ - self._send_mail( - u'uri: {uri}\n{summary}\n\n{body}'.format( - uri=file_uri, - summary=summary, - body=body_text - ), - process_name - ) - - def on_process_failure(self, process): - """ - :type process: ScheduledProcess - """ - - # A negative exit code means it was killed via a signal. Probably by the user. - # Not worth emailing. - if process.exitcode < 0: - return - - with open(process.log_file, 'rt') as f: - msg = f.read() - - self._send_mail(msg, process.name) - - def _send_mail(self, body_text, process_name): - """ - :type body_text: str - :type process_name: str - """ - hostname = socket.getfqdn() - msg = MIMEText(body_text.encode('utf-8'), 'plain', 'utf-8') - msg['Subject'] = Header(u'{name} failure on {hostname}'.format( - name=process_name, - hostname=hostname - ).encode('utf-8'), 'utf-8') - from_address = 'fetch-{pid}@{hostname}'.format( - pid=multiprocessing.current_process().pid, - hostname=hostname - ) - msg['from'] = from_address - msg['to'] = ", ".join(self.addresses) - s = smtplib.SMTP('localhost') - s.sendmail( - from_address, - self.addresses, - msg.as_string() - ) - s.quit() - - class FileProcessor(SimpleObject): """ Any action that will process a file after retrieval. (base class) @@ -619,7 +543,7 @@ def process(self, file_path): raise FileProcessError('Return code %r from command %r' % (returned, command)) # Check that output exists - expect_file = self.upload_dir + '/{file_stem}.h5' + expect_file = self.upload_dir + '/{filename}' expected_path = self._apply_file_pattern(expect_file, file_path) if not os.path.exists(expected_path): diff --git a/fetch/auto.py b/fetch/auto.py index 7bdf393..c2859a2 100644 --- a/fetch/auto.py +++ b/fetch/auto.py @@ -25,7 +25,7 @@ from croniter import croniter from . import load -from ._core import ResultHandler, TaskFailureEmailer, RemoteFetchException, mkdirs +from ._core import ResultHandler, RemoteFetchException, mkdirs # setproctitle is only supported on some platforms (Linux). try: @@ -61,21 +61,6 @@ def _attempt_lock(lock_file): return True -def _redirect_output(log_file): - """ - Redirect all output to the given file. - - :type log_file: str - """ - output = open(log_file, 'w') - sys.stdout = output - sys.stderr = output - logging_clear() - handler = logging.StreamHandler(stream=output) - handler.setFormatter(_LOG_FORMATTER) - logging.getLogger().addHandler(handler) - - def _run_item(reporter, item, scheduled_time, log_directory, lock_directory): """ Run the given module in a subprocess @@ -143,7 +128,6 @@ def run(self): Configure the environment and run our module. """ _init_signals() - _redirect_output(self.log_file) if not _attempt_lock(self.lock_file): _log.debug('Lock is activated. Skipping run. %r', self.name) @@ -235,7 +219,7 @@ def _on_child_finish(child, notifiers): if exit_code != 0: _log.error( - 'Error return code %s from %r. Output logged to %r', + 'Error return code %s from %r.', exit_code, child.name, child.log_file ) @@ -270,16 +254,6 @@ def get_day_log_dir(log_directory, time_secs): :type time_secs: float :rtype: str - >>> get_day_log_dir('/tmp/day-dir-test', 1416285412.541422) - '/tmp/day-dir-test/2014/11-18' - """ - # We use localtime because the cron scheduling uses localtime. - t = time.localtime(time_secs) - - day_log_dir = os.path.join( - log_directory, - time.strftime('%Y', t), - time.strftime('%m-%d', t) ) if not os.path.exists(day_log_dir): mkdirs(day_log_dir) @@ -383,11 +357,6 @@ def load(self): _log.info('%s messaging configuration.', 'Loaded' if config.messaging_settings else 'No') - self.notifiers = [] - if config.notify_addresses: - self.notifiers.append(TaskFailureEmailer(config.notify_addresses)) - _log.info('%s addresses for error notification: %s', len(config.notify_addresses), config.notify_addresses) - if not os.path.exists(self.base_directory): raise ValueError('Configured base folder does not exist: %r' % self.base_directory) @@ -662,6 +631,6 @@ def _set_logging_levels(levels): _log.info('Set log level %s to %s', name, level) -_LOG_HANDLER = logging.StreamHandler(stream=sys.stderr) +_LOG_HANDLER = logging.StreamHandler(sys.stdout) _LOG_FORMATTER = logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s") _LOG_HANDLER.setFormatter(_LOG_FORMATTER) diff --git a/fetch/ecmwf.py b/fetch/ecmwf.py index dfa20a5..6a5a145 100644 --- a/fetch/ecmwf.py +++ b/fetch/ecmwf.py @@ -40,7 +40,6 @@ def retrieve(self, req): c = APIRequest( self.url, "datasets/%s" % (dataset, ), - self.email, self.key, self.trace, verbose=self.verbose diff --git a/fetch/load.py b/fetch/load.py index c270a16..8e0fa4d 100644 --- a/fetch/load.py +++ b/fetch/load.py @@ -10,7 +10,6 @@ import os import yaml -import yaml.resolver from croniter import croniter from . import ftp, http, ecmwf @@ -137,7 +136,7 @@ class Config(object): Configuration. """ - def __init__(self, directory, rules, notify_addresses, messaging_settings=None, log_levels=None): + def __init__(self, directory, rules, messaging_settings=None, log_levels=None): """ :type directory: str :type rules: list of ScheduledItem @@ -150,8 +149,6 @@ def __init__(self, directory, rules, notify_addresses, messaging_settings=None, # Empty list of rules is ok: they may be added after startup (a config reload/SIGHUP). self.rules = rules - self.notify_addresses = notify_addresses - if messaging_settings: # Optional library. #: pylint: disable=import-error @@ -174,12 +171,6 @@ def from_dict(cls, config): messaging_settings = config.get('messaging') log_levels = config.get('log') - notify_email_addresses = [] - if 'notify' in config: - notify_config = config['notify'] - if 'email' in notify_config: - notify_email_addresses = notify_config['email'] - rules = [] if 'rules' in config: for name, fields in config['rules'].items(): @@ -190,7 +181,6 @@ def from_dict(cls, config): return Config( directory, rules, - notify_email_addresses, messaging_settings=messaging_settings, log_levels=log_levels ) @@ -202,9 +192,6 @@ def to_dict(self): """ return remove_nones({ 'directory': self.directory, - 'notify': { - 'email': self.notify_addresses - }, 'log': self.log_levels, 'messaging': self.messaging_settings, 'rules': dict([ diff --git a/test/test_ecmwf.py b/test/test_ecmwf.py index 9485f4c..0be53b7 100644 --- a/test/test_ecmwf.py +++ b/test/test_ecmwf.py @@ -134,9 +134,6 @@ def _make_ecmwf_config(ancillary_data_root='/tmp/anc'): """ schedule = { 'directory': '/tmp/anc-fetch', - 'notify': { - 'email': ['test@ga.gov.au'] - }, 'log': { 'fetch': 'DEBUG' }, diff --git a/test/test_load.py b/test/test_load.py index 386b9f5..73b3d7d 100644 --- a/test/test_load.py +++ b/test/test_load.py @@ -116,9 +116,6 @@ def _make_config(): anc_data = '/tmp/anc' schedule = { 'directory': '/tmp/anc-fetch', - 'notify': { - 'email': ['jeremy.hooke@ga.gov.au'] - }, 'log': { 'fetch': 'DEBUG' }, From 9401fd9febc37bd4379ebbfb842d4227cf8ae04b Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Thu, 22 Jan 2026 20:17:11 +0000 Subject: [PATCH 07/10] logging removed --- fetch/auto.py | 52 ++++++++--------------------------------------- test/test_auto.py | 1 - 2 files changed, 9 insertions(+), 44 deletions(-) diff --git a/fetch/auto.py b/fetch/auto.py index c2859a2..702d06d 100644 --- a/fetch/auto.py +++ b/fetch/auto.py @@ -61,7 +61,7 @@ def _attempt_lock(lock_file): return True -def _run_item(reporter, item, scheduled_time, log_directory, lock_directory): +def _run_item(reporter, item, scheduled_time, lock_directory): """ Run the given module in a subprocess :type reporter: ResultHandler @@ -69,11 +69,11 @@ def _run_item(reporter, item, scheduled_time, log_directory, lock_directory): :rtype: ScheduledProcess """ p = ScheduledProcess( - reporter, item, scheduled_time, log_directory, lock_directory + reporter, item, scheduled_time, lock_directory ) _log.debug('Module info %r', item.module) - _log.info('Starting %r. Log %r, Lock %r', p.name, p.log_file, p.lock_file) + _log.info('Starting %r. Lock %r', p.name, p.lock_file) p.start() return p @@ -83,22 +83,21 @@ class ScheduledProcess(multiprocessing.Process): A subprocess to run a module. """ - def __init__(self, reporter, item, scheduled_time, log_directory, lock_directory, epoch_to_time=time.localtime): + def __init__(self, reporter, item, scheduled_time, lock_directory, epoch_to_time=time.localtime): """ :type reporter: fetch.ResultHandler :type item: fetch.load.ScheduledItem :type scheduled_time: float - :type log_directory: str :type lock_directory: str >>> from ._core import EmptySource >>> item = load.ScheduledItem('LS7 CPF', '* * * * *', EmptySource()) >>> # 04:36 UTC time >>> scheduled_time = 1416285412.541422 - >>> log, lock = '/tmp/test-log', '/tmp/test-lock' - >>> s = ScheduledProcess(None, item, scheduled_time, log, lock, epoch_to_time=time.gmtime) - >>> (s.name, s.log_file, s.lock_file) - ('fetch-0436-ls7-cpf', '/tmp/test-log/0436-ls7-cpf.log', '/tmp/test-lock/ls7-cpf.lck') + >>> lock = '/tmp/test-lock' + >>> s = ScheduledProcess(None, item, scheduled_time, lock, epoch_to_time=time.gmtime) + >>> (s.name, s.lock_file) + ('fetch-0436-ls7-cpf', '/tmp/test-lock/ls7-cpf.lck') """ super(ScheduledProcess, self).__init__() id_ = item.sanitized_name @@ -107,15 +106,7 @@ def __init__(self, reporter, item, scheduled_time, log_directory, lock_directory '{id}.lck'.format(id=id_) ) scheduled_time_st = time.strftime('%H%M', epoch_to_time(scheduled_time)) - log_file = os.path.join( - log_directory, - '{time}-{id}.log'.format( - id=id_, - time=scheduled_time_st - ) - ) - self.log_file = log_file self.lock_file = lock_file self.name = 'fetch-{}-{}'.format(scheduled_time_st, id_) self.scheduled_time = scheduled_time @@ -220,7 +211,7 @@ def _on_child_finish(child, notifiers): if exit_code != 0: _log.error( 'Error return code %s from %r.', - exit_code, child.name, child.log_file + exit_code, child.name, ) for n in notifiers: @@ -247,20 +238,6 @@ def _filter_finished_children(running_children, notifiers): return still_running -def get_day_log_dir(log_directory, time_secs): - """ - Get log directory for this day. - :type log_directory: str - :type time_secs: float - :rtype: str - - ) - if not os.path.exists(day_log_dir): - mkdirs(day_log_dir) - - return day_log_dir - - def _on_shutdown(running_children, notifiers): """ :type running_children: set of ScheduledProcess @@ -333,8 +310,6 @@ def __init__(self, config_path): self.schedule = None # : type: str self.base_directory = None - # : type: str - self.log_directory = None #: type: str self.lock_directory = None #: :type: list of fetch.TaskFailureListener @@ -367,11 +342,6 @@ def load(self): if not os.path.exists(self.lock_directory): mkdirs(self.lock_directory) - self.log_directory = os.path.join(self.base_directory, 'log') - _log.info('Using log directory %s', self.log_directory) - if not os.path.exists(self.log_directory): - mkdirs(self.log_directory) - if config.log_levels != self.log_levels: _set_logging_levels(config.log_levels) self.log_levels = config.log_levels @@ -510,8 +480,6 @@ def run_loop(o): reporter, scheduled_item, scheduled_time=scheduled_time, - # Use a unique log directory for each day - log_directory=get_day_log_dir(o.log_directory, scheduled_time), lock_directory=o.lock_directory ) running_children.add(p) @@ -571,8 +539,6 @@ def run_items(o, *item_names): NotifyResultHandler(o, chosen_item.sanitized_name), chosen_item, scheduled_time=scheduled_time, - # Use a unique log directory for each day - log_directory=get_day_log_dir(o.log_directory, scheduled_time), lock_directory=o.lock_directory ) running_children.add(p) diff --git a/test/test_auto.py b/test/test_auto.py index 46422db..9b67698 100644 --- a/test/test_auto.py +++ b/test/test_auto.py @@ -12,7 +12,6 @@ def __init__(self, name='p', exitcode=None, pid=None): self.exitcode = exitcode self.name = name self.pid = pid - self.log_file = '/tmp/test.log' running_proc = MockProcess(exitcode=None) failed_proc = MockProcess(exitcode=1) From b3ec9fb5d120c7523c195b0c04f47b0d1192f6af Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Thu, 22 Jan 2026 20:41:58 +0000 Subject: [PATCH 08/10] working and test passing --- check-code.sh | 3 +-- example-config.yaml | 9 --------- fetch/_core.py | 2 +- fetch/auto.py | 36 ++---------------------------------- fetch/load.py | 12 +----------- test/test_load.py | 16 ---------------- 6 files changed, 5 insertions(+), 73 deletions(-) diff --git a/check-code.sh b/check-code.sh index 3c9f5b3..9e05123 100755 --- a/check-code.sh +++ b/check-code.sh @@ -13,5 +13,4 @@ pylint --py3k --reports no ${py_files} pep8 ${py_files} --max-line-length 120 # Run tests -# -> But not those that require the neocommon library. -py.test fetch test -m 'not with_neocommon' +py.test fetch test diff --git a/example-config.yaml b/example-config.yaml index 1c125e7..1559ba5 100755 --- a/example-config.yaml +++ b/example-config.yaml @@ -7,18 +7,9 @@ # The work directory (for log and lock files): directory: /data/fetch -# Message bus config, for announcing arrivals. Optional. -messaging: - host: rhe-neo-dev01.dev.lan - # virtual_host: - username: fetch - password: fetch - # Logging level for modules (any python module can be added here) log: fetch: DEBUG - neocommon: DEBUG - neocommon.files: INFO # Download rules: rules: diff --git a/fetch/_core.py b/fetch/_core.py index 9dcda3e..b44725d 100644 --- a/fetch/_core.py +++ b/fetch/_core.py @@ -543,7 +543,7 @@ def process(self, file_path): raise FileProcessError('Return code %r from command %r' % (returned, command)) # Check that output exists - expect_file = self.upload_dir + '/{filename}' + expect_file = self.upload_dir + '/{file_stem}.h5' expected_path = self._apply_file_pattern(expect_file, file_path) if not os.path.exists(expected_path): diff --git a/fetch/auto.py b/fetch/auto.py index 702d06d..d44a719 100644 --- a/fetch/auto.py +++ b/fetch/auto.py @@ -314,8 +314,6 @@ def __init__(self, config_path): self.lock_directory = None #: :type: list of fetch.TaskFailureListener self.notifiers = [] - #: :type: dict of (str, str) - self.messaging_settings = None # Key-values are log names and levels. #: :type: dict of (str, str) self.log_levels = None @@ -328,9 +326,6 @@ def load(self): self.schedule = Schedule(config.rules) self.base_directory = config.directory - self.messaging_settings = config.messaging_settings - - _log.info('%s messaging configuration.', 'Loaded' if config.messaging_settings else 'No') if not os.path.exists(self.base_directory): raise ValueError('Configured base folder does not exist: %r' % self.base_directory) @@ -360,33 +355,6 @@ def __init__(self, config, job_id): self.config = config self.job_id = job_id - def _announce_files_complete(self, source_uri, paths, msg_metadata=None): - """ - Announce on the message bus that files are complete. - - No-op if there is no messaging configuration. - :type source_uri: str - :type paths: list of str - """ - md = msg_metadata or {} - md.update({ - 'source-uri': source_uri - }) - - _log.info('Completed %r -> %r', source_uri, paths) - if self.config.messaging_settings: - # Optional library. - #: pylint: disable=import-error - from neocommon import message, Uri as NeoUri - uris = [NeoUri.parse(path) for path in paths] - with message.NeoMessenger(message.MessengerConnection(**self.config.messaging_settings)) as msg: - msg.announce_ancillary( - message.AncillaryUpdate( - ancillary_type=self.job_id, - uris=uris, - properties=md - ) - ) def files_complete(self, source_uri, paths, msg_metadata=None): """ @@ -398,7 +366,7 @@ def files_complete(self, source_uri, paths, msg_metadata=None): :type msg_metadata: dict of (str, str) :return: """ - self._announce_files_complete(source_uri, paths, msg_metadata=msg_metadata) + _log.info('Completed %r -> %r', source_uri, paths) def file_complete(self, source_uri, path, msg_metadata=None): """ @@ -407,7 +375,7 @@ def file_complete(self, source_uri, path, msg_metadata=None): :type msg_metadata: dict of (str, str) :type path: str """ - self._announce_files_complete(source_uri, [path], msg_metadata=msg_metadata) + _log.info('Completed %r -> %r', source_uri, path) def file_error(self, uri, summary, body): """ diff --git a/fetch/load.py b/fetch/load.py index 8e0fa4d..2c6d69d 100644 --- a/fetch/load.py +++ b/fetch/load.py @@ -136,7 +136,7 @@ class Config(object): Configuration. """ - def __init__(self, directory, rules, messaging_settings=None, log_levels=None): + def __init__(self, directory, rules, log_levels=None): """ :type directory: str :type rules: list of ScheduledItem @@ -149,13 +149,6 @@ def __init__(self, directory, rules, messaging_settings=None, log_levels=None): # Empty list of rules is ok: they may be added after startup (a config reload/SIGHUP). self.rules = rules - if messaging_settings: - # Optional library. - #: pylint: disable=import-error - from neocommon.message import MessengerConnection - verify_can_construct(MessengerConnection, messaging_settings, identifier='messaging settings') - - self.messaging_settings = messaging_settings self.log_levels = log_levels @classmethod @@ -168,7 +161,6 @@ def from_dict(cls, config): """ directory = config.get('directory') - messaging_settings = config.get('messaging') log_levels = config.get('log') rules = [] @@ -181,7 +173,6 @@ def from_dict(cls, config): return Config( directory, rules, - messaging_settings=messaging_settings, log_levels=log_levels ) @@ -193,7 +184,6 @@ def to_dict(self): return remove_nones({ 'directory': self.directory, 'log': self.log_levels, - 'messaging': self.messaging_settings, 'rules': dict([ ( r.name, remove_nones({ diff --git a/test/test_load.py b/test/test_load.py index 73b3d7d..78f32f7 100644 --- a/test/test_load.py +++ b/test/test_load.py @@ -8,8 +8,6 @@ from fetch import load -with_neocommon = pytest.mark.with_neocommon - def _fail_with_diff(reparsed_config, source_config): print('-' * 20) @@ -44,20 +42,6 @@ def test_dump_load_obj_full(): _check_load_dump_config(_make_config) -@with_neocommon -def test_dump_load_obj_with_messaging(): - def make_config_no_messaging(): - c = _make_config() - c['messaging'] = { - 'host': 'rhe-pma-test08.test.lan', - 'username': 'fetch', - 'password': 'fetch' - } - return c - - _check_load_dump_config(make_config_no_messaging) - - def print_simple_obj_diff(dict1, dict2): if type(dict2) in (int, float, str, text): print('- {!r}'.format(dict1)) From 64be6ccc502435dea95892a43a67cf15bef6d4cf Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Fri, 23 Jan 2026 00:36:54 +0000 Subject: [PATCH 09/10] add logging for debug --- fetch/_core.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/fetch/_core.py b/fetch/_core.py index b44725d..d53dc03 100644 --- a/fetch/_core.py +++ b/fetch/_core.py @@ -541,6 +541,8 @@ def process(self, file_path): returned = subprocess.call(command, shell=True) if returned != 0: raise FileProcessError('Return code %r from command %r' % (returned, command)) + else: + _log.debug("Command completed successfully.") # Check that output exists expect_file = self.upload_dir + '/{file_stem}.h5' From ba992f51d3246ee610571ad9168009e5be79e2ad Mon Sep 17 00:00:00 2001 From: Brad Greer Date: Wed, 28 Jan 2026 22:24:05 +0000 Subject: [PATCH 10/10] DOFS-31 code style fixes --- fetch/auto.py | 1 - fetch/s3.py | 2 ++ test/test_load.py | 7 +++--- test/test_s3.py | 15 ++++++------ test/test_shellfileprocessor.py | 41 ++++++++++++++++++--------------- 5 files changed, 36 insertions(+), 30 deletions(-) diff --git a/fetch/auto.py b/fetch/auto.py index d44a719..6bf5b1c 100644 --- a/fetch/auto.py +++ b/fetch/auto.py @@ -355,7 +355,6 @@ def __init__(self, config, job_id): self.config = config self.job_id = job_id - def files_complete(self, source_uri, paths, msg_metadata=None): """ Call on completion of multiple files. diff --git a/fetch/s3.py b/fetch/s3.py index 54705c0..d128b6a 100644 --- a/fetch/s3.py +++ b/fetch/s3.py @@ -5,12 +5,14 @@ _log = logging.getLogger(__name__) _s3 = boto3.client("s3") + class FileUploadError(Exception): """ An error in file processing. """ pass + def upload(filepath: str, bucket: str, prefix: str): """ Upload all files in input_file_dir to S3. diff --git a/test/test_load.py b/test/test_load.py index 78f32f7..19e3b29 100644 --- a/test/test_load.py +++ b/test/test_load.py @@ -3,7 +3,6 @@ import tempfile -import pytest from pathlib import Path from fetch import load @@ -174,9 +173,9 @@ def _make_config(): 'process': ShellFileProcessor( command='/usr/local/bin/gdal_translate -a_srs "+proj=latlong +datum=WGS84" ' '{parent_dir}/{file_stem}.nc {parent_dir}/{file_stem}.tif', - upload_dir= '{parent_dir}', - bucket= 'ard-processing-data', - prefix= 'ancillary/water_vapour' + upload_dir='{parent_dir}', + bucket='ard-processing-data', + prefix='ancillary/water_vapour' ) }, 'NPP GDAS-forecast': { diff --git a/test/test_s3.py b/test/test_s3.py index 64d2a05..512979a 100644 --- a/test/test_s3.py +++ b/test/test_s3.py @@ -1,19 +1,18 @@ """ -A package for testing the code in s3.py +A package for testing the code in s3.py """ -import os -from pathlib import Path from fetch import s3 from fetch.s3 import upload, FileUploadError import pytest -from unittest.mock import patch, MagicMock +from unittest.mock import patch # Constants for tests filepath = "/tmp/test.txt" bucket = "my-bucket" prefix = "my/prefix" + def test_upload_success(): # Set up mocks with ( @@ -36,6 +35,7 @@ def test_upload_success(): "my/prefix/test.txt", ) + def test_given_file_not_exist(): # Set up mocks with ( @@ -45,12 +45,13 @@ def test_given_file_not_exist(): ): # Run code and confirm we get an expection back, checking type & message with pytest.raises(FileUploadError, match="File does not exist: /tmp/test.txt"): - upload(filepath, bucket, prefix) + upload(filepath, bucket, prefix) # Expected logged messages mock_log.info.assert_called_once_with("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix") mock_s3.upload_file.assert_not_called() + def test_write_to_s3_failed(): # Set up mocks with ( @@ -59,10 +60,10 @@ def test_write_to_s3_failed(): patch.object(s3, "_log") as mock_log, ): mock_s3.upload_file.side_effect = Exception("poop") - + # Run code and confirm we get an expection back, checking type & message with pytest.raises(FileUploadError, match="Failed to upload test.txt to s3://my-bucket/my/prefix/test.txt"): - upload(filepath, bucket, prefix) + upload(filepath, bucket, prefix) # Expected logged messages mock_log.info.assert_called_once_with("Uploading file /tmp/test.txt to s3://my-bucket/my/prefix") diff --git a/test/test_shellfileprocessor.py b/test/test_shellfileprocessor.py index 2198cd2..2335782 100644 --- a/test/test_shellfileprocessor.py +++ b/test/test_shellfileprocessor.py @@ -1,22 +1,20 @@ """ A package for testing the code in the ShellFileProcessor class, in _core.py """ -import os -import subprocess import pytest -from pathlib import Path -from unittest.mock import patch, MagicMock +from unittest.mock import patch, call from fetch import s3 from fetch import _core from fetch._core import ShellFileProcessor, FileProcessError # Test Constants -command="run command on file {filename}" -upload_dir="/data/upload" -bucket="s3-bucket" -prefix="s3-prefix" -input_file="/data/staging/file.txt" +command = "run command on file {filename}" +upload_dir = "/data/upload" +bucket = "s3-bucket" +prefix = "s3-prefix" +input_file = "/data/staging/file.txt" + def test_apply_file_pattern_basic(): p = ShellFileProcessor( @@ -30,14 +28,15 @@ def test_apply_file_pattern_basic(): assert result == "run command on file something.txt" + def test_process_success(): # Set up mocks with ( patch("subprocess.call", return_value=0), patch("os.path.exists", return_value=True), - patch.object(s3, "_s3") as mock_s3, + patch.object(s3, "_s3"), patch.object(_core, "_log") as mock_log - ): + ): p = ShellFileProcessor( command=command, upload_dir=upload_dir, @@ -50,16 +49,20 @@ def test_process_success(): # confirm the correct messages are logged mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt') - mock_log.debug.assert_called_once_with('File available %r', '/data/upload/file.h5') + mock_log.debug.assert_has_calls([ + call("Command completed successfully."), + call('File available %r', '/data/upload/file.h5') + ]) + def test_process_command_failed(): # Set up mocks with ( patch("subprocess.call", return_value=1), patch("os.path.exists", return_value=True), - patch.object(s3, "_s3") as mock_s3, + patch.object(s3, "_s3"), patch.object(_core, "_log") as mock_log - ): + ): p = ShellFileProcessor( command=command, upload_dir=upload_dir, @@ -75,14 +78,15 @@ def test_process_command_failed(): mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt') mock_log.debug.assert_not_called() + def test_process_output_file_missing(): # Set up mocks with ( patch("subprocess.call", return_value=0), patch("os.path.exists", return_value=False), - patch.object(s3, "_s3") as mock_s3, + patch.object(s3, "_s3"), patch.object(_core, "_log") as mock_log - ): + ): p = ShellFileProcessor( command=command, upload_dir=upload_dir, @@ -91,9 +95,10 @@ def test_process_output_file_missing(): ) # Run code and confirm we get an expection back, checking type & message - with pytest.raises(FileProcessError, match="Expected output not found '/data/upload/file.h5' for command 'run command on file file.txt'"): + message = "Expected output not found '/data/upload/file.h5' for command 'run command on file file.txt'" + with pytest.raises(FileProcessError, match=message): p.process(input_file) # confirm the correct messages are logged mock_log.info.assert_called_once_with('Running %r', 'run command on file file.txt') - mock_log.debug.assert_not_called() \ No newline at end of file + mock_log.debug.assert_called_once_with("Command completed successfully.")