DMSC Integration Testing

Last updated: September 15, 2025 23:37:02

Test: ingestor|nmx|file_found_is_not_old|manual

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

None

instrument = 'nmx', coda_proposal_id = '977695'
datefinder =

def test_file_found_is_not_old(
instrument: str,
coda_proposal_id: str,
datefinder: Callable[..., datetime],
) -> None:
> date = datefinder(instrument=instrument, proposal_id=coda_proposal_id)

tests/ingestor/ingestor_test.py:45:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/ingestor/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:98: in get_latest_nexus_time
path = get_latest_nexus_paths(instrument=instrument, proposal_id=proposal_id)[0]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'NMX', proposal_id = '977695', n = 1, inspect = 20

def get_latest_nexus_paths(
instrument: str, proposal_id: str | None, n: int = 1, inspect: int = 20
) -> list[Path]:
"""
Get the latest file paths manually by sorting the files in the directory.

Parameters
----------
instrument:
The instrument name.
proposal_id:
SciCat proposal ID for CODA.
If ``None``, the environment variable ``OVERRIDE_RAW_DATA_DIR`` must be set.
n:
The number of files to return.
inspect:
The number of files to inspect before returning.

Returns
-------
:
Paths to the n latest files for the given instrument and proposal.
"""
instrument = NEXUS_INSTRUMENT_NAME[instrument]
path = _coda_raw_data_path(proposal_id)
files_found = []
logging.info(
"Scanning the %d latest files in %s for instrument %s",
inspect,
path,
instrument,
)
recent_files = sorted(path.glob("*.hdf"))[-inspect::][::-1]
logging.info("Recent files: %s", recent_files)
for file in recent_files:
logging.info("Inspecting file: %s", file)
try:
with snx.File(file, "r", locking=False) as f:
instrument_in_file = f["/entry/instrument/name"][()]
if instrument_in_file == instrument:
logging.info(
"File %s matches requested instrument %s. Adding to list. "
"Last modified: %s",
file,
instrument,
datetime.fromtimestamp(
os.stat(file).st_mtime, tz=local_timezone()
),
)
files_found.append(file)
if len(files_found) == n:
logging.info("Found %d files. Returning.", n)
return files_found
else:
logging.info(
"File instrument is %s, but %s was requested. Skipping.",
instrument_in_file,
instrument,
)
except OSError as e:
logging.warning(
"File %s could not be opened. Reason: %s",
file,
e,
)
continue
logging.error("No files found for instrument %s", instrument)
> raise RuntimeError(f"No files found for instrument {instrument}")
E RuntimeError: No files found for instrument NMX

.tox/ingestor/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

instrument = 'nmx', coda_proposal_id = '977695'
datefinder =

def test_file_found_is_not_old(
instrument: str,
coda_proposal_id: str,
datefinder: Callable[..., datetime],
) -> None:
date = datefinder(instrument=instrument, proposal_id=coda_proposal_id)
> assert date > datetime.now(date.tzinfo) - timedelta(hours=24)
E AssertionError: assert datetime.datetime(2025, 8, 30, 8, 8, 20, 908820, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')) > (datetime.datetime(2025, 9, 1, 10, 26, 45, 532642, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')) - datetime.timedelta(days=1))
E + where datetime.datetime(2025, 9, 1, 10, 26, 45, 532642, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')) = (datetime.timezone(datetime.timedelta(seconds=7200), 'CEST'))
E + where = datetime.now
E + and datetime.timezone(datetime.timedelta(seconds=7200), 'CEST') = datetime.datetime(2025, 8, 30, 8, 8, 20, 908820, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')).tzinfo
E + and datetime.timedelta(days=1) = timedelta(hours=24)

tests/ingestor/ingestor_test.py:46: AssertionError

instrument = 'nmx', coda_proposal_id = '977695'
datefinder =

def test_file_found_is_not_old(
instrument: str,
coda_proposal_id: str,
datefinder: Callable[..., datetime],
) -> None:
date = datefinder(instrument=instrument, proposal_id=coda_proposal_id)
> assert date > datetime.now(date.tzinfo) - timedelta(hours=24)
E AssertionError: assert datetime.datetime(2025, 8, 30, 8, 8, 20, 908820, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')) > (datetime.datetime(2025, 9, 1, 0, 25, 40, 439046, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')) - datetime.timedelta(days=1))
E + where datetime.datetime(2025, 9, 1, 0, 25, 40, 439046, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')) = (datetime.timezone(datetime.timedelta(seconds=7200), 'CEST'))
E + where = datetime.now
E + and datetime.timezone(datetime.timedelta(seconds=7200), 'CEST') = datetime.datetime(2025, 8, 30, 8, 8, 20, 908820, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), 'CEST')).tzinfo
E + and datetime.timedelta(days=1) = timedelta(hours=24)

tests/ingestor/ingestor_test.py:46: AssertionError

None

None

None

None

instrument = 'nmx', coda_proposal_id = '977695'
datefinder =

def test_file_found_is_not_old(
instrument: str,
coda_proposal_id: str,
datefinder: Callable[..., datetime],
) -> None:
> date = datefinder(instrument=instrument, proposal_id=coda_proposal_id)

tests/ingestor/ingestor_test.py:45:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/ingestor/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:98: in get_latest_nexus_time
path = get_latest_nexus_paths(instrument=instrument, proposal_id=proposal_id)[0]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

instrument = 'NMX', proposal_id = '977695', n = 1, inspect = 20

def get_latest_nexus_paths(
instrument: str, proposal_id: str | None, n: int = 1, inspect: int = 20
) -> list[Path]:
"""
Get the latest file paths manually by sorting the files in the directory.

Parameters
----------
instrument:
The instrument name.
proposal_id:
SciCat proposal ID for CODA.
If ``None``, the environment variable ``OVERRIDE_RAW_DATA_DIR`` must be set.
n:
The number of files to return.
inspect:
The number of files to inspect before returning.

Returns
-------
:
Paths to the n latest files for the given instrument and proposal.
"""
instrument = NEXUS_INSTRUMENT_NAME[instrument]
path = _coda_raw_data_path(proposal_id)
files_found = []
logging.info(
"Scanning the %d latest files in %s for instrument %s",
inspect,
path,
instrument,
)
recent_files = sorted(path.glob("*.hdf"))[-inspect::][::-1]
logging.info("Recent files: %s", recent_files)
for file in recent_files:
logging.info("Inspecting file: %s", file)
try:
with snx.File(file, "r") as f:
instrument_in_file = f["/entry/instrument/name"][()]
if instrument_in_file == instrument:
logging.info(
"File %s matches requested instrument %s. Adding to list. "
"Last modified: %s",
file,
instrument,
datetime.fromtimestamp(
os.stat(file).st_mtime, tz=local_timezone()
),
)
files_found.append(file)
if len(files_found) == n:
logging.info("Found %d files. Returning.", n)
return files_found
else:
logging.info(
"File instrument is %s, but %s was requested. Skipping.",
instrument_in_file,
instrument,
)
except OSError as e:
logging.warning(
"File %s could not be opened. Reason: %s",
file,
e,
)
continue
logging.error("No files found for instrument %s", instrument)
> raise RuntimeError(f"No files found for instrument {instrument}")
E RuntimeError: No files found for instrument NMX

.tox/ingestor/lib/python3.12/site-packages/dmsc_nightly/nexusfiles/coda.py:94: RuntimeError

None

None