From e790c7342aad60c7e62a27deecbfc1d861c3d766 Mon Sep 17 00:00:00 2001 From: bjk7119 Date: Thu, 27 Aug 2026 17:23:53 +0900 Subject: [PATCH 1/5] fix(binary): exclude excel files from binary analysis results --- src/fosslight_binary/binary_analysis.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/fosslight_binary/binary_analysis.py b/src/fosslight_binary/binary_analysis.py index b883ad1..2c196e7 100755 --- a/src/fosslight_binary/binary_analysis.py +++ b/src/fosslight_binary/binary_analysis.py @@ -32,7 +32,7 @@ PKG_NAME = "fosslight_binary" logger = logging.getLogger(constant.LOGGER_NAME) -_REMOVE_FILE_EXTENSION = ['json', 'js'] +_REMOVE_FILE_EXTENSION = ['json', 'js', 'xlsx', 'xls', 'xlsm'] _REMOVE_FILE_COMMAND_RESULT = ['timezone data', 'apple binary property list'] INCLUDE_FILE_COMMAND_RESULT = ['current ar archive'] From 70ff181bcabca30423741b4c246f93870b6a992a Mon Sep 17 00:00:00 2001 From: bjk7119 Date: Thu, 27 Aug 2026 17:24:15 +0900 Subject: [PATCH 2/5] fix(binary): always remove the temp dir when analysis is interrupted --- src/fosslight_binary/binary_analysis.py | 67 +++++++++++++++++++++++-- 1 file changed, 63 insertions(+), 4 deletions(-) diff --git a/src/fosslight_binary/binary_analysis.py b/src/fosslight_binary/binary_analysis.py index 2c196e7..4dc4f25 100755 --- a/src/fosslight_binary/binary_analysis.py +++ b/src/fosslight_binary/binary_analysis.py @@ -35,8 +35,11 @@ _REMOVE_FILE_EXTENSION = ['json', 'js', 'xlsx', 'xls', 'xlsm'] _REMOVE_FILE_COMMAND_RESULT = ['timezone data', 'apple binary property list'] INCLUDE_FILE_COMMAND_RESULT = ['current ar archive'] +_TEMP_DIR_PREFIX = '.fosslight_temp_' +_LOG_FILE_PREFIX = 'fosslight_log_bin_' _error_logs = [] +_temp_output_path = "" _root_path = "" start_time = "" finish_time = "" @@ -71,6 +74,45 @@ def get_checksum_and_tlsh(bin_with_path): return checksum_value, tlsh_value, error_msg +def _prepare_temp_dir(temp_path): + """Create the temp directory that holds intermediate output. + + A directory left behind by a previous run killed with SIGKILL or a power + loss is removed first. Kept around, the copytree at the end of the run + would copy that run's result files into the output directory as well. + """ + global _temp_output_path + + if os.path.isdir(temp_path): + shutil.rmtree(temp_path, ignore_errors=True) + os.makedirs(temp_path, exist_ok=True) + _temp_output_path = temp_path + + +def _cleanup_temp_dir(): + """Remove the temp directory if it is still there. + + Called from the finally block of find_binaries so that no temp directory + survives, however the analysis ends (interrupt, exception, sys.exit). + """ + global _temp_output_path + + temp_path, _temp_output_path = _temp_output_path, "" + if not temp_path or not os.path.isdir(temp_path): + return + + temp_path = os.path.abspath(temp_path) + logging_logger = logging.getLogger(constant.LOGGER_NAME) + for handler in logging_logger.handlers[:]: + if (isinstance(handler, logging.FileHandler) + and os.path.dirname(os.path.abspath(handler.baseFilename)) == temp_path): + handler.flush() + handler.close() + logging_logger.removeHandler(handler) + + shutil.rmtree(temp_path, ignore_errors=True) + + def init(path_to_find_bin, output_file_name, formats, path_to_exclude=[]): global logger, _result_log @@ -85,7 +127,8 @@ def init(path_to_find_bin, output_file_name, formats, path_to_exclude=[]): output_path = os.path.abspath(output_path) original_output_path = output_path - output_path = os.path.join(output_path, '.fosslight_temp') + output_path = os.path.join(output_path, f"{_TEMP_DIR_PREFIX}{file_time}") + _prepare_temp_dir(output_path) while len(output_files) < len(output_extensions): output_files.append(None) @@ -125,7 +168,7 @@ def init(path_to_find_bin, output_file_name, formats, path_to_exclude=[]): logger.error(f"Format error - {msg}") sys.exit(1) - log_file = os.path.join(output_path, f"fosslight_log_bin_{file_time}.txt") + log_file = os.path.join(output_path, f"{_LOG_FILE_PREFIX}{file_time}.txt") logger, _result_log = init_log(log_file, True, logging.INFO, logging.DEBUG, PKG_NAME, path_to_find_bin, path_to_exclude) @@ -172,6 +215,22 @@ def get_file_list(path_to_find, excluded_files): def find_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token="", simple_mode=False, correct_mode=True, correct_filepath="", path_to_exclude=[], all_exclude_mode=()): + """Analyze binaries, leaving no temp directory behind however the run ends. + + Ctrl+C (KeyboardInterrupt), an unexpected exception and the sys.exit raised + by error_occured all pass through the finally block. + """ + try: + return _analyze_binaries(path_to_find_bin, output_dir, formats, kb_url, kb_token, + simple_mode, correct_mode, correct_filepath, path_to_exclude, + all_exclude_mode) + finally: + _cleanup_temp_dir() + + +def _analyze_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token="", simple_mode=False, + correct_mode=True, correct_filepath="", path_to_exclude=[], + all_exclude_mode=()): global start_time, finish_time, _root_path, _result_log mode = "Normal Mode" @@ -291,7 +350,8 @@ def find_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token="", try: if os.path.isfile(log_file): - move_log_file(log_file, os.path.join(original_output_path, f"fosslight_log_bin_{timestamp_for_filename(start_time)}.txt")) + move_log_file(log_file, os.path.join(original_output_path, + f"{_LOG_FILE_PREFIX}{timestamp_for_filename(start_time)}.txt")) else: logger.debug("Moving binary analysis log file is skipped") except Exception as ex: @@ -300,7 +360,6 @@ def find_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token="", try: if os.path.isdir(output_path): shutil.copytree(output_path, original_output_path, dirs_exist_ok=True) - shutil.rmtree(output_path) else: logger.debug(f"Temp directory not found, skip moving: {output_path}") except Exception as ex: From 21f036570cf83c304fd84e0f01229b234c0c1417 Mon Sep 17 00:00:00 2001 From: bjk7119 Date: Thu, 17 Sep 2026 15:12:50 +0900 Subject: [PATCH 3/5] fix(binary): consolidate temp output lifecycle handling --- src/fosslight_binary/binary_analysis.py | 97 +++++++++++++++---------- 1 file changed, 57 insertions(+), 40 deletions(-) diff --git a/src/fosslight_binary/binary_analysis.py b/src/fosslight_binary/binary_analysis.py index 4dc4f25..2a50033 100755 --- a/src/fosslight_binary/binary_analysis.py +++ b/src/fosslight_binary/binary_analysis.py @@ -39,7 +39,6 @@ _LOG_FILE_PREFIX = 'fosslight_log_bin_' _error_logs = [] -_temp_output_path = "" _root_path = "" start_time = "" finish_time = "" @@ -74,30 +73,26 @@ def get_checksum_and_tlsh(bin_with_path): return checksum_value, tlsh_value, error_msg -def _prepare_temp_dir(temp_path): +def _prepare_temp_dir(output_dir, file_time): """Create the temp directory that holds intermediate output. - A directory left behind by a previous run killed with SIGKILL or a power - loss is removed first. Kept around, the copytree at the end of the run - would copy that run's result files into the output directory as well. + The timestamp-only name matches the scanner's single-run-per-second + execution model. Remove a stale directory before reusing its name. """ - global _temp_output_path - + os.makedirs(output_dir, exist_ok=True) + temp_path = os.path.join(output_dir, f'{_TEMP_DIR_PREFIX}{file_time}') if os.path.isdir(temp_path): - shutil.rmtree(temp_path, ignore_errors=True) - os.makedirs(temp_path, exist_ok=True) - _temp_output_path = temp_path + shutil.rmtree(temp_path) + os.makedirs(temp_path) + return temp_path -def _cleanup_temp_dir(): +def _cleanup_temp_dir(temp_path): """Remove the temp directory if it is still there. - Called from the finally block of find_binaries so that no temp directory - survives, however the analysis ends (interrupt, exception, sys.exit). + Called from the finally block of find_binaries so that only this + invocation's directory is removed, however the analysis ends. """ - global _temp_output_path - - temp_path, _temp_output_path = _temp_output_path, "" if not temp_path or not os.path.isdir(temp_path): return @@ -113,7 +108,37 @@ def _cleanup_temp_dir(): shutil.rmtree(temp_path, ignore_errors=True) -def init(path_to_find_bin, output_file_name, formats, path_to_exclude=[]): +def _finalize_temp_output(temp_output_path, final_output_path, log_file, file_time, log=None): + """Publish analysis artifacts and always remove the temp directory.""" + if not temp_output_path or not os.path.isdir(temp_output_path): + return True + + publish_ok = True + try: + if os.path.isfile(log_file): + move_log_file(log_file, os.path.join( + final_output_path, f'{_LOG_FILE_PREFIX}{file_time}.txt')) + else: + if log: + log.debug("Moving binary analysis log file is skipped") + except Exception as ex: + publish_ok = False + if log: + log.error(f"Failed to move log file: {ex}") + + try: + shutil.copytree(temp_output_path, final_output_path, dirs_exist_ok=True) + except Exception as ex: + publish_ok = False + if log: + log.error(f"Failed to publish scan artifacts: {ex}") + finally: + _cleanup_temp_dir(temp_output_path) + + return publish_ok + + +def init(path_to_find_bin, output_file_name, formats, path_to_exclude=[], temp_path_holder=None): global logger, _result_log _json_ext = ".json" @@ -127,8 +152,9 @@ def init(path_to_find_bin, output_file_name, formats, path_to_exclude=[]): output_path = os.path.abspath(output_path) original_output_path = output_path - output_path = os.path.join(output_path, f"{_TEMP_DIR_PREFIX}{file_time}") - _prepare_temp_dir(output_path) + output_path = _prepare_temp_dir(output_path, file_time) + if temp_path_holder is not None: + temp_path_holder["path"] = output_path while len(output_files) < len(output_extensions): output_files.append(None) @@ -220,17 +246,18 @@ def find_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token="", Ctrl+C (KeyboardInterrupt), an unexpected exception and the sys.exit raised by error_occured all pass through the finally block. """ + temp_path_holder = {} try: return _analyze_binaries(path_to_find_bin, output_dir, formats, kb_url, kb_token, simple_mode, correct_mode, correct_filepath, path_to_exclude, - all_exclude_mode) + all_exclude_mode, temp_path_holder=temp_path_holder) finally: - _cleanup_temp_dir() + _cleanup_temp_dir(temp_path_holder.get("path")) def _analyze_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token="", simple_mode=False, correct_mode=True, correct_filepath="", path_to_exclude=[], - all_exclude_mode=()): + all_exclude_mode=(), temp_path_holder=None): global start_time, finish_time, _root_path, _result_log mode = "Normal Mode" @@ -245,7 +272,7 @@ def _analyze_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token _result_log, binary_yaml_file, compressed_yaml_file = init_simple(output_dir, PKG_NAME, start_time) else: _result_log, result_reports, output_extensions, formats, output_path, original_output_path, log_file = init( - path_to_find_bin, output_dir, formats, path_to_exclude) + path_to_find_bin, output_dir, formats, path_to_exclude, temp_path_holder) total_bin_cnt = 0 db_loaded_cnt = 0 @@ -348,23 +375,6 @@ def _analyze_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token else: logger.error(f"Fail to generate result file.:{writing_msg}") - try: - if os.path.isfile(log_file): - move_log_file(log_file, os.path.join(original_output_path, - f"{_LOG_FILE_PREFIX}{timestamp_for_filename(start_time)}.txt")) - else: - logger.debug("Moving binary analysis log file is skipped") - except Exception as ex: - logger.debug(f"Failed to move log file: {ex}") - - try: - if os.path.isdir(output_path): - shutil.copytree(output_path, original_output_path, dirs_exist_ok=True) - else: - logger.debug(f"Temp directory not found, skip moving: {output_path}") - except Exception as ex: - logger.debug(f"Failed to move temp files: {ex}") - try: print_result_log(mode=mode, success=True, result_log=_result_log, file_cnt=str(cnt_file_except_skipped), @@ -373,6 +383,13 @@ def _analyze_binaries(path_to_find_bin, output_dir, formats, kb_url="", kb_token except Exception as ex: error_occured(error_msg=f"Print log : {ex}", exit=False) + if not simple_mode: + publish_ok = _finalize_temp_output( + output_path, original_output_path, log_file, + timestamp_for_filename(start_time), logger) + if not publish_ok: + success_to_write = False + return success_to_write, scan_item From 5be9ba7ed1ede11b56efb4643f087eabb15ef52d Mon Sep 17 00:00:00 2001 From: bjk7119 Date: Tue, 22 Sep 2026 08:43:24 +0900 Subject: [PATCH 4/5] fix(binary): create unique temp dirs per scan --- src/fosslight_binary/binary_analysis.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/src/fosslight_binary/binary_analysis.py b/src/fosslight_binary/binary_analysis.py index 2a50033..960731b 100755 --- a/src/fosslight_binary/binary_analysis.py +++ b/src/fosslight_binary/binary_analysis.py @@ -28,6 +28,7 @@ import subprocess import re import shutil +import tempfile PKG_NAME = "fosslight_binary" logger = logging.getLogger(constant.LOGGER_NAME) @@ -76,15 +77,12 @@ def get_checksum_and_tlsh(bin_with_path): def _prepare_temp_dir(output_dir, file_time): """Create the temp directory that holds intermediate output. - The timestamp-only name matches the scanner's single-run-per-second - execution model. Remove a stale directory before reusing its name. + Keep the timestamp in the directory name for traceability, but let the + OS add a unique suffix so concurrent scans targeting the same output path + never clobber each other's intermediate files. """ os.makedirs(output_dir, exist_ok=True) - temp_path = os.path.join(output_dir, f'{_TEMP_DIR_PREFIX}{file_time}') - if os.path.isdir(temp_path): - shutil.rmtree(temp_path) - os.makedirs(temp_path) - return temp_path + return tempfile.mkdtemp(prefix=f'{_TEMP_DIR_PREFIX}{file_time}_', dir=output_dir) def _cleanup_temp_dir(temp_path): From 6f5750ac681509ebaac37700f2e23d28c694829c Mon Sep 17 00:00:00 2001 From: bjk7119 Date: Tue, 22 Sep 2026 09:51:17 +0900 Subject: [PATCH 5/5] fix(binary): clean up temp directories after analysis --- src/fosslight_binary/binary_analysis.py | 4 +- tests/test_temp_dir_cleanup.py | 308 ++++++++++++++++++++++++ tests/test_temp_dir_concurrency.py | 65 +++++ 3 files changed, 374 insertions(+), 3 deletions(-) create mode 100644 tests/test_temp_dir_cleanup.py create mode 100644 tests/test_temp_dir_concurrency.py diff --git a/src/fosslight_binary/binary_analysis.py b/src/fosslight_binary/binary_analysis.py index 960731b..dc0f927 100755 --- a/src/fosslight_binary/binary_analysis.py +++ b/src/fosslight_binary/binary_analysis.py @@ -107,7 +107,7 @@ def _cleanup_temp_dir(temp_path): def _finalize_temp_output(temp_output_path, final_output_path, log_file, file_time, log=None): - """Publish analysis artifacts and always remove the temp directory.""" + """Publish analysis artifacts from the temp directory.""" if not temp_output_path or not os.path.isdir(temp_output_path): return True @@ -130,8 +130,6 @@ def _finalize_temp_output(temp_output_path, final_output_path, log_file, file_ti publish_ok = False if log: log.error(f"Failed to publish scan artifacts: {ex}") - finally: - _cleanup_temp_dir(temp_output_path) return publish_ok diff --git a/tests/test_temp_dir_cleanup.py b/tests/test_temp_dir_cleanup.py new file mode 100644 index 0000000..f2fb8d7 --- /dev/null +++ b/tests/test_temp_dir_cleanup.py @@ -0,0 +1,308 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# Copyright (c) 2026 LG Electronics Inc. +# SPDX-License-Identifier: Apache-2.0 + +"""Regression tests for temp directory cleanup and excel exclusion. + +Interrupting an analysis used to leave .fosslight_temp behind, and a following +run then showed the excel inside that directory in the Binary result. Design: +docs/superpowers/specs/2026-08-27-binary-temp-dir-cleanup-design.md +""" + +import glob +import logging +import os +import zipfile + +import pytest + +import fosslight_util.constant as constant +from fosslight_util.exclude import get_excluded_paths +from fosslight_util.time import timestamp_for_filename +from fosslight_binary import binary_analysis as ba + +TEMP_DIR_PREFIX = '.fosslight_temp_' + + +@pytest.fixture(autouse=True) +def real_log_init(monkeypatch): + """Let init_log initialize logging as it does under the real CLI. + + init_log skips both the log directory creation and the FileHandler setup + when logger.hasHandlers() is True. The pytest logging plugin attaches a + capture handler for every test phase (directly onto a logger whose + propagate is False), so left alone neither the temp directory nor the log + file is created and the cleanup cannot be observed. Clearing the handlers + right before the call reproduces the real condition; the initialization + itself is delegated to the real init_log. + """ + logger = logging.getLogger(constant.LOGGER_NAME) + original_propagate = logger.propagate + real_init_log = ba.init_log + + def _clear_handlers(): + for handler in logger.handlers[:]: + handler.close() + logger.removeHandler(handler) + + def _init_log_without_pytest_handlers(*args, **kwargs): + _clear_handlers() + logger.propagate = False + return real_init_log(*args, **kwargs) + + monkeypatch.setattr(ba, 'init_log', _init_log_without_pytest_handlers) + ba._error_logs.clear() + yield + _clear_handlers() + logger.propagate = original_propagate + + +@pytest.fixture(autouse=True) +def offline_binary_db(monkeypatch): + """Pass through the Binary DB lookup, the only step that hits the network.""" + monkeypatch.setattr( + ba, 'get_oss_info_from_db', + lambda bin_list, kb_url="", kb_token="": (bin_list, 0, "")) + + +def _write_binary(path): + """Write a file that binaryornot classifies as a binary.""" + with open(path, 'wb') as f: + f.write(b'\x7fELF\x02\x01\x01\x00' + b'\x00' * 512) + + +def _write_xlsx(path): + """Write a real xlsx-shaped zip, which content sniffing reports as binary.""" + with zipfile.ZipFile(path, 'w') as archive: + archive.writestr('xl/workbook.xml', '') + + +def _scan_dir(tmp_path): + """A scan target directory holding exactly one binary.""" + scan_dir = tmp_path / 'scan' + scan_dir.mkdir() + _write_binary(scan_dir / 'mybin') + return scan_dir + + +def _out_dir(tmp_path): + out_dir = tmp_path / 'out' + out_dir.mkdir(exist_ok=True) + return out_dir + + +def _temp_dirs(out_dir): + return list(out_dir.glob(f'{TEMP_DIR_PREFIX}*')) + + +def _reported_paths(scan_item): + return {item.source_name_or_path + for items in scan_item.file_items.values() + for item in items} + + +def test_excel_is_not_reported_as_binary(tmp_path): + """An excel file never shows up as a row in the Binary result. + + Reproduces the fosslight_scanner path: the exclude list is computed up + front and the excel appears afterwards. get_file_list only checks + membership in that pre-computed excluded_files, so a file created later + slips past the extension policy and lands in the result. + """ + scan_dir = _scan_dir(tmp_path) + all_exclude_mode = get_excluded_paths(str(scan_dir), []) + _write_xlsx(scan_dir / 'fosslight_report_src_20260827.xlsx') + + _, scan_item = ba.find_binaries(str(scan_dir), str(_out_dir(tmp_path)), ['excel'], + all_exclude_mode=all_exclude_mode) + + assert _reported_paths(scan_item) == {'mybin'} + + +def test_temp_dir_name_includes_timestamp(tmp_path, monkeypatch): + """The temp directory name includes the run timestamp.""" + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + fixed_start_time = '20260827_000000' + created_paths = [] + + real_prepare_temp_dir = ba._prepare_temp_dir + + def _capture_temp_path(output_dir, file_time): + temp_path = real_prepare_temp_dir(output_dir, file_time) + created_paths.append(temp_path) + return temp_path + + monkeypatch.setattr(ba, 'current_timestamp_utc', lambda: fixed_start_time) + monkeypatch.setattr(ba, '_prepare_temp_dir', _capture_temp_path) + + ba.find_binaries(str(scan_dir), str(out_dir), ['excel']) + + assert os.path.dirname(created_paths[0]) == str(out_dir) + assert os.path.basename(created_paths[0]).startswith( + f'{TEMP_DIR_PREFIX}{timestamp_for_filename(fixed_start_time)}_') + + +def test_finalize_temp_output_copies_and_removes_temp_dir(tmp_path): + """Finalization publishes artifacts""" + temp_dir = tmp_path / 'temp' + final_dir = tmp_path / 'final' + temp_dir.mkdir() + (temp_dir / 'report.xlsx').write_bytes(b'report') + + publish_ok = ba._finalize_temp_output( + str(temp_dir), str(final_dir), str(temp_dir / 'missing.log'), + '20260917_120000') + + assert publish_ok is True + assert (final_dir / 'report.xlsx').read_bytes() == b'report' + assert temp_dir.exists() + + +def test_finalize_temp_output_reports_copy_failure_and_removes_temp_dir(tmp_path, monkeypatch): + """A publication failure is reported without forcing cleanup here.""" + temp_dir = tmp_path / 'temp' + final_dir = tmp_path / 'final' + temp_dir.mkdir() + + def fail_copytree(*args, **kwargs): + raise OSError('copy failed') + + monkeypatch.setattr(ba.shutil, 'copytree', fail_copytree) + + publish_ok = ba._finalize_temp_output( + str(temp_dir), str(final_dir), str(temp_dir / 'missing.log'), + '20260917_120000') + + assert publish_ok is False + assert temp_dir.exists() + + +def test_copy_failure_marks_analysis_unsuccessful(tmp_path, monkeypatch): + """A failed result copy changes the analysis result to unsuccessful.""" + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + + def fail_copytree(*args, **kwargs): + raise OSError('copy failed') + + monkeypatch.setattr(ba.shutil, 'copytree', fail_copytree) + + success, _ = ba.find_binaries(str(scan_dir), str(out_dir), ['excel']) + + assert success is False + + +def test_success_path_cleans_temp_dir_once(tmp_path, monkeypatch): + """The success path delegates temp cleanup to one place only.""" + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + real_cleanup = ba._cleanup_temp_dir + cleanup_calls = [] + + def capture_cleanup(path): + cleanup_calls.append(path) + return real_cleanup(path) + + monkeypatch.setattr(ba, '_cleanup_temp_dir', capture_cleanup) + + ba.find_binaries(str(scan_dir), str(out_dir), ['excel']) + + assert len(cleanup_calls) == 1 + + +def test_stale_temp_dir_is_not_copied_into_output(tmp_path, monkeypatch): + """A file left in the temp directory never leaks into the output directory. + + The copytree at the end of the run copies the whole temp directory into + the output directory, so an excel left there by a previous run killed with + SIGKILL or a power loss would show up among the results. + """ + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + fixed_start_time = '20260827_000000' + monkeypatch.setattr(ba, 'current_timestamp_utc', lambda: fixed_start_time) + stale_temp = out_dir / ( + f'{TEMP_DIR_PREFIX}{timestamp_for_filename(fixed_start_time)}') + stale_temp.mkdir() + _write_xlsx(stale_temp / 'fosslight_report_bin_leftover.xlsx') + + ba.find_binaries(str(scan_dir), str(out_dir), ['excel']) + + assert not (out_dir / 'fosslight_report_bin_leftover.xlsx').exists() + + +def test_temp_dir_removed_on_keyboard_interrupt(tmp_path, monkeypatch): + """Interrupting with Ctrl+C leaves no temp directory behind.""" + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + + def _interrupt(bin_with_path): + raise KeyboardInterrupt + + monkeypatch.setattr(ba, 'get_checksum_and_tlsh', _interrupt) + + with pytest.raises(KeyboardInterrupt): + ba.find_binaries(str(scan_dir), str(out_dir), ['excel']) + + assert not _temp_dirs(out_dir) + + +def test_log_file_is_removed_with_temp_dir_on_interrupt(tmp_path, monkeypatch): + """The interrupted run's log is removed with its temp directory.""" + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + + def _interrupt(bin_with_path): + raise KeyboardInterrupt + + monkeypatch.setattr(ba, 'get_checksum_and_tlsh', _interrupt) + + with pytest.raises(KeyboardInterrupt): + ba.find_binaries(str(scan_dir), str(out_dir), ['excel']) + + assert not glob.glob(os.path.join(str(out_dir), 'fosslight_log_bin_*.txt')) + + +def test_temp_dir_removed_on_unexpected_exception(tmp_path, monkeypatch): + """An unexpected exception leaves no temp directory behind.""" + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + + def _boom(*args, **kwargs): + raise RuntimeError("unexpected") + + monkeypatch.setattr(ba, 'get_excluded_paths', _boom) + + with pytest.raises(RuntimeError): + ba.find_binaries(str(scan_dir), str(out_dir), ['excel']) + + assert not _temp_dirs(out_dir) + + +def test_temp_dir_removed_when_init_exits_before_returning(tmp_path, monkeypatch): + """init() exiting before it returns still cleans up its temp directory. + + On Windows the spdx and cyclonedx formats are dropped, and init() calls + sys.exit(0) once nothing is left. That happens after the temp directory has + been created, so the cleanup cannot rely on init()'s return value. + """ + scan_dir = _scan_dir(tmp_path) + out_dir = _out_dir(tmp_path) + monkeypatch.setattr(ba.platform, 'system', lambda: 'Windows') + + with pytest.raises(SystemExit): + ba.find_binaries(str(scan_dir), str(out_dir), ['spdx-json']) + + assert not _temp_dirs(out_dir) + + +def test_temp_dir_removed_when_scan_path_is_missing(tmp_path): + """The sys.exit raised by error_occured(exit=True) also cleans up.""" + out_dir = _out_dir(tmp_path) + + with pytest.raises(SystemExit): + ba.find_binaries(str(tmp_path / 'no_such_dir'), str(out_dir), ['excel']) + + assert not _temp_dirs(out_dir) diff --git a/tests/test_temp_dir_concurrency.py b/tests/test_temp_dir_concurrency.py new file mode 100644 index 0000000..f9dcd08 --- /dev/null +++ b/tests/test_temp_dir_concurrency.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# Copyright (c) 2026 LG Electronics Inc. +# SPDX-License-Identifier: Apache-2.0 + +import os + +import fosslight_binary.binary_analysis as ba + + +def test_same_timestamp_uses_unique_temp_directory(tmp_path): + output_dir = tmp_path / 'out' + + first = ba._prepare_temp_dir(str(output_dir), "20260917_120000") + second = ba._prepare_temp_dir(str(output_dir), "20260917_120000") + + assert os.path.basename(first).startswith('.fosslight_temp_20260917_120000') + assert os.path.basename(second).startswith('.fosslight_temp_20260917_120000') + assert first != second + assert os.path.isdir(first) + assert os.path.isdir(second) + + +def test_same_timestamp_replaces_stale_temp_directory(tmp_path): + output_dir = tmp_path / 'out' + first = ba._prepare_temp_dir(str(output_dir), '20260917_120000') + stale_file = os.path.join(first, 'stale.txt') + with open(stale_file, 'w', encoding='utf-8') as file: + file.write('stale') + + second = ba._prepare_temp_dir(str(output_dir), '20260917_120000') + + assert second != first + assert os.path.exists(stale_file) + assert os.path.isdir(second) + + +def test_cleanup_temp_dir_only_removes_requested_path(tmp_path): + first = tmp_path / "first" + second = tmp_path / "second" + first.mkdir() + second.mkdir() + + ba._cleanup_temp_dir(str(first)) + + assert not first.exists() + assert second.exists() + + +def test_find_binaries_passes_invocation_path_to_cleanup(tmp_path, monkeypatch): + temp_path = tmp_path / "invocation-temp" + observed = {} + + def fake_analyze(*args, temp_path_holder=None, **kwargs): + temp_path.mkdir() + temp_path_holder["path"] = str(temp_path) + + def capture_cleanup(path): + observed["path"] = path + + monkeypatch.setattr(ba, "_analyze_binaries", fake_analyze) + monkeypatch.setattr(ba, "_cleanup_temp_dir", capture_cleanup) + ba.find_binaries("scan", "out", ["excel"]) + + assert observed["path"] == str(temp_path)