diff --git a/dmeta/functions.py b/dmeta/functions.py index c7e008a..e5b5783 100644 --- a/dmeta/functions.py +++ b/dmeta/functions.py @@ -1,27 +1,55 @@ # -*- coding: utf-8 -*- """DMeta functions.""" + import os import shutil import zipfile from PIL import Image -from art import tprint -import defusedxml.lxml as lxml -from .errors import DMetaBaseError +import defusedxml.ElementTree as ET from .util import get_file_format, extract, read_json -from .params import CORE_XML_MAP, APP_XML_MAP, OVERVIEW, DMETA_VERSION, \ - UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR, \ - SUPPORTED_MICROSOFT_FORMATS, SUPPORTED_FORMATS, \ - JPEG_MARKER_PREFIX, JPEG_SOI, JPEG_EOI, JPEG_SOS, JPEG_COM, \ - JPEG_APP_FIRST, JPEG_APP_LAST, JPEG_STANDALONE_MARKERS, \ - GIF_TRAILER, GIF_EXTENSION_INTRODUCER, GIF_IMAGE_DESCRIPTOR, \ - GIF_EXT_GRAPHIC_CONTROL, GIF_EXT_APPLICATION, \ - GIF_APP_EXT_NETSCAPE_IDENTIFIER - - -def overwrite_metadata( - xml_path, - metadata=None, - is_core=True): +from .params import ( + CORE_XML_MAP, + APP_XML_MAP, + SUPPORTED_MICROSOFT_FORMATS, + SUPPORTED_FORMATS, + JPEG_MARKER_PREFIX, + JPEG_SOI, + JPEG_EOI, + JPEG_SOS, + JPEG_COM, + JPEG_APP_FIRST, + JPEG_APP_LAST, + JPEG_STANDALONE_MARKERS, + GIF_TRAILER, + GIF_EXTENSION_INTRODUCER, + GIF_IMAGE_DESCRIPTOR, + GIF_EXT_GRAPHIC_CONTROL, + GIF_EXT_APPLICATION, + GIF_APP_EXT_NETSCAPE_IDENTIFIER, +) + + +def _output_path(source_file_path, in_place, suffix): + """ + Determine the output path for a processed file. + + :param source_file_path: path to the source file + :type source_file_path: str + :param in_place: whether to modify the file in place + :type in_place: bool + :param suffix: suffix to add to the filename when not modifying in place + :type suffix: str + :return: path for the output file + :rtype: str + """ + if in_place: + return source_file_path + else: + base, ext = os.path.splitext(source_file_path) + return base + suffix + ext + + +def overwrite_metadata(xml_path, metadata=None, is_core=True): """ Overwrite metadata in an XML file based on a predefined mapping. @@ -36,74 +64,241 @@ def overwrite_metadata( """ xml_map = CORE_XML_MAP if is_core else APP_XML_MAP if os.path.exists(xml_path): - e_core = lxml.parse(xml_path) - for xml_element in e_core.iter(): + # Parse XML file, make changes, and write back + tree = ET.parse(xml_path) + for xml_element in tree.iter(): for personal_field in xml_map if metadata is None else metadata: associated_xml_tag = xml_map[personal_field] - if (associated_xml_tag in xml_element.tag): - xml_element.text = "" if metadata is None else metadata[personal_field] - e_core.write(xml_path) - - -def clear(microsoft_file_name, in_place=False, verbose=False): + if associated_xml_tag in xml_element.tag: + xml_element.text = ( + "" if metadata is None else metadata[personal_field] + ) + tree.write(xml_path) + + +def _cleanup_extracted(unzipped_dir, source_file, verbose): + """Clean up extracted directory and close file handle.""" + # Ensure the source_file is closed before removing directory + source_file.close() + # On Windows, we may need to force garbage collection or add a small delay + # to ensure file handles are fully released before removing the directory + import gc + + gc.collect() # Force garbage collection to release any remaining references + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If we can't remove the directory immediately on Windows, + # try again after a short delay + import time + + time.sleep(0.1) + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If it still fails, log the error but don't crash + if verbose: + print( + f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later" + ) + + +def _extract_and_prepare(source_file_path, verbose): """ - Clear all the editable metadata in the given Microsoft file. - - :param microsoft_file_name: name of Microsoft file - :type microsoft_file_name: str - :param in_place: the `in_place` flag applies the changes directly to the original file - :type in_place: bool - :param verbose: the `verbose` flag enables detailed output + Extract the file and prepare the processing environment. + + :param source_file_path: path to the source file to process + :type source_file_path: str + :param verbose: whether to enable verbose output :type verbose: bool - :return: None + :return: tuple of (unzipped_dir, source_file, xml_paths_dict, microsoft_format) + :rtype: tuple """ - microsoft_format = get_file_format(microsoft_file_name) + # Validate format + microsoft_format = get_file_format(source_file_path) if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: - return - unzipped_dir, source_file = extract(microsoft_file_name) + return None, None, None, None + + # Extract the file + unzipped_dir, source_file = extract(source_file_path) + doc_props_dir = os.path.join(unzipped_dir, "docProps") core_xml_path = os.path.join(doc_props_dir, "core.xml") app_xml_path = os.path.join(doc_props_dir, "app.xml") + + xml_paths = { + 'core_xml_path': core_xml_path, + 'app_xml_path': app_xml_path + } + + return unzipped_dir, source_file, xml_paths, microsoft_format - def is_metadata_cleared(xml_path, is_core=True): - if not os.path.exists(xml_path): - return True - tree = lxml.parse(xml_path) - xml_map = CORE_XML_MAP if is_core else APP_XML_MAP - for xml_element in tree.iter(): - for personal_field in xml_map: - associated_xml_tag = xml_map[personal_field] - if (associated_xml_tag in xml_element.tag): - if xml_element.text and xml_element.text.strip(): - return False - return True - core_cleared = is_metadata_cleared(core_xml_path) - app_cleared = is_metadata_cleared(app_xml_path, is_core=False) +def _repack_and_cleanup(unzipped_dir, source_file, modified_path, verbose, process_type): + """ + Repack the file into ZIP and perform cleanup. + + :param unzipped_dir: path to the unzipped directory + :type unzipped_dir: str + :param source_file: the opened source file object + :type source_file: zipfile.ZipFile + :param modified_path: path to the modified file to create + :type modified_path: str + :param verbose: whether to enable verbose output + :type verbose: bool + :param process_type: type of processing ('clear' or 'update') + :type process_type: str + :return: path to the modified file + :rtype: str + """ + # Re-zip the file + with zipfile.ZipFile(modified_path, "w", compression=zipfile.ZIP_DEFLATED) as zip_file: + for file_name in source_file.namelist(): + zip_file.write(os.path.join(unzipped_dir, file_name), file_name) - if core_cleared and app_cleared: - if verbose: - print(f"Metadata is already cleared for: {microsoft_file_name}") - shutil.rmtree(unzipped_dir) - return + if verbose: + action = "cleared" if process_type == 'clear' else "updated" + print(f"{action.capitalize()} metadata for: {modified_path}") - # Clear metadata if not already cleared - overwrite_metadata(core_xml_path) - overwrite_metadata(app_xml_path, is_core=False) + return modified_path - modified = microsoft_file_name - if not in_place: - modified = microsoft_file_name[:microsoft_file_name.rfind('.')] + "_cleared" + "." + microsoft_format - with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: - for file_name in source_file.namelist(): - file.write(os.path.join(unzipped_dir, file_name), file_name) - file.close() - shutil.rmtree(unzipped_dir) - if verbose: - print(f"Cleared metadata for: {microsoft_file_name}") +def _process_metadata(source_file_path, in_place, verbose, process_type, **kwargs): + """ + Shared function to process metadata in Microsoft files (clear/update) + + :param source_file_path: path to the source file to process + :type source_file_path: str + :param in_place: whether to modify the file in place + :type in_place: bool + :param verbose: whether to enable verbose output + :type verbose: bool + :param process_type: type of processing ('clear' or 'update') + :type process_type: str + :param kwargs: additional arguments for processing (e.g., new_metadata_dict for update) + :return: path to the processed file or None if unsuccessful + :rtype: str or None + """ + # Extract and prepare + unzipped_dir, source_file, xml_paths, microsoft_format = _extract_and_prepare(source_file_path, verbose) + if xml_paths is None: + return None + + core_xml_path = xml_paths['core_xml_path'] + app_xml_path = xml_paths['app_xml_path'] + + try: + # Define processing logic based on type + if process_type == 'clear': + # Define the clear-specific check function + def is_metadata_cleared(xml_path, is_core=True): + if not os.path.exists(xml_path): + return True + tree = ET.parse(xml_path) + xml_map = CORE_XML_MAP if is_core else APP_XML_MAP + for xml_element in tree.iter(): + for personal_field in xml_map: + associated_xml_tag = xml_map[personal_field] + if associated_xml_tag in xml_element.tag: + if xml_element.text and xml_element.text.strip(): + return False + return True + + # Check if metadata is already cleared + core_cleared = is_metadata_cleared(core_xml_path) + app_cleared = is_metadata_cleared(app_xml_path, is_core=False) + + if core_cleared and app_cleared: + if verbose: + print(f"Metadata is already cleared for: {source_file_path}") + return source_file_path # Return the original file path when already cleared + + # Clear metadata if not already cleared + overwrite_metadata(core_xml_path) + overwrite_metadata(app_xml_path, is_core=False) + + # Determine output filename + modified = _output_path(source_file_path, in_place, "_cleared") + + elif process_type == 'update': + # Get the metadata from kwargs + new_metadata_dict = kwargs.get('new_metadata_dict', {}) + personal_fields_core_xml = { + k: new_metadata_dict[k] for k in CORE_XML_MAP.keys() if k in new_metadata_dict + } + personal_fields_app_xml = {k: new_metadata_dict[k] for k in APP_XML_MAP.keys() if k in new_metadata_dict} + + has_core_tags = len(personal_fields_core_xml) > 0 + has_app_tags = len(personal_fields_app_xml) > 0 + + if not (has_core_tags or has_app_tags): + print("There isn't any chosen personal field to remove.") + return None + + # Define the update-specific check function + def is_metadata_up_to_date(xml_path, metadata, is_core=True): + if not os.path.exists(xml_path): + return False + tree = ET.parse(xml_path) + xml_map = CORE_XML_MAP if is_core else APP_XML_MAP + for xml_element in tree.iter(): + for personal_field in xml_map if metadata is None else metadata: + associated_xml_tag = xml_map[personal_field] + if associated_xml_tag in xml_element.tag: + if xml_element.text != metadata[personal_field]: + return False + return True + + core_up_to_date = ( + is_metadata_up_to_date(core_xml_path, personal_fields_core_xml) + if has_core_tags + else True + ) + app_up_to_date = ( + is_metadata_up_to_date(app_xml_path, personal_fields_app_xml) + if has_app_tags + else True + ) + + if core_up_to_date and app_up_to_date: + if verbose: + print(f"Metadata is already up to date for: {source_file_path}") + return source_file_path + + # Update metadata if not already up to date + if has_core_tags: + overwrite_metadata(core_xml_path, personal_fields_core_xml) + if has_app_tags: + overwrite_metadata(app_xml_path, personal_fields_app_xml, is_core=False) + + # Determine output filename + modified = _output_path(source_file_path, in_place, "_updated") + else: + raise ValueError(f"Unknown process type: {process_type}") + + # Repack and cleanup + result = _repack_and_cleanup(unzipped_dir, source_file, modified, verbose, process_type) + return result + + finally: + _cleanup_extracted(unzipped_dir, source_file, verbose) + + +def clear(microsoft_file_name, in_place=False, verbose=False): + """ + Clear all the editable metadata in the given Microsoft file. - return modified + :param microsoft_file_name: name of Microsoft file + :type microsoft_file_name: str + :param in_place: the `in_place` flag applies the changes directly to the original file + :type in_place: bool + :param verbose: the `verbose` flag enables detailed output + :type verbose: bool + :return: path to the cleared file if successful, None if format is unsupported, + or original file path if metadata is already cleared + :rtype: str or None + """ + return _process_metadata(microsoft_file_name, in_place, verbose, 'clear') def clear_all(in_place=False, verbose=False): @@ -117,9 +312,7 @@ def clear_all(in_place=False, verbose=False): :return: None """ path = os.getcwd() - counter = { - fmt: 0 for fmt in SUPPORTED_FORMATS - } + counter = {fmt: 0 for fmt in SUPPORTED_FORMATS} for root, _, files in os.walk(path): for file in files: @@ -131,7 +324,11 @@ def clear_all(in_place=False, verbose=False): if verbose: for fmt in counter.keys(): - print("Metadata of {} files with the format of {} has been cleared.".format(counter[fmt], fmt)) + print( + "Metadata of {} files with the format of {} has been cleared.".format( + counter[fmt], fmt + ) + ) def update(config_file_name, microsoft_file_name, in_place=False, verbose=False): @@ -146,70 +343,11 @@ def update(config_file_name, microsoft_file_name, in_place=False, verbose=False) :type in_place: bool :param verbose: the `verbose` flag enables detailed output :type verbose: bool - :return: None + :return: path to the updated file if successful, None if format is unsupported + :rtype: str or None """ config = read_json(config_file_name) - personal_fields_core_xml = {k: config[k] for k in CORE_XML_MAP.keys() if k in config} - personal_fields_app_xml = {k: config[k] for k in APP_XML_MAP.keys() if k in config} - - has_core_tags = len(personal_fields_core_xml) > 0 - has_app_tags = len(personal_fields_app_xml) > 0 - - if not (has_core_tags or has_app_tags): - print("There isn't any chosen personal field to remove.") - return - - microsoft_format = get_file_format(microsoft_file_name) - if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: - return - - unzipped_dir, source_file = extract(microsoft_file_name) - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - - # Check if metadata is already up to date - def is_metadata_up_to_date(xml_path, metadata, is_core=True): - if not os.path.exists(xml_path): - return False - tree = lxml.parse(xml_path) - xml_map = CORE_XML_MAP if is_core else APP_XML_MAP - for xml_element in tree.iter(): - for personal_field in xml_map if metadata is None else metadata: - associated_xml_tag = xml_map[personal_field] - if (associated_xml_tag in xml_element.tag): - if xml_element.text != metadata[personal_field]: - return False - return True - - core_up_to_date = is_metadata_up_to_date(core_xml_path, personal_fields_core_xml) if has_core_tags else True - app_up_to_date = is_metadata_up_to_date(app_xml_path, personal_fields_app_xml) if has_app_tags else True - - if core_up_to_date and app_up_to_date: - if verbose: - print(f"Metadata is already up to date for: {microsoft_file_name}") - shutil.rmtree(unzipped_dir) - return - - # Update metadata if not already up to date - if has_core_tags: - overwrite_metadata(core_xml_path, personal_fields_core_xml) - if has_app_tags: - overwrite_metadata(app_xml_path, personal_fields_app_xml, is_core=False) - - modified = microsoft_file_name - if not in_place: - modified = microsoft_file_name[:microsoft_file_name.rfind('.')] + "_updated" + "." + microsoft_format - with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: - for file_name in source_file.namelist(): - file.write(os.path.join(unzipped_dir, file_name), file_name) - file.close() - shutil.rmtree(unzipped_dir) - - if verbose: - print(f"Updated metadata for: {microsoft_file_name}") - - return modified + return _process_metadata(microsoft_file_name, in_place, verbose, 'update', new_metadata_dict=config) def update_all(config_file_name, in_place=False, verbose=False): @@ -225,9 +363,7 @@ def update_all(config_file_name, in_place=False, verbose=False): :return: None """ path = os.getcwd() - counter = { - format: 0 for format in SUPPORTED_MICROSOFT_FORMATS - } + counter = {format: 0 for format in SUPPORTED_MICROSOFT_FORMATS} for root, _, files in os.walk(path): for file in files: @@ -239,7 +375,11 @@ def update_all(config_file_name, in_place=False, verbose=False): if verbose: for format in counter.keys(): - print("Metadata of {} files with the format of {} has been updated.".format(counter[format], format)) + print( + "Metadata of {} files with the format of {} has been updated.".format( + counter[format], format + ) + ) def clear_png_metadata(png_file_name, in_place=False, verbose=False): @@ -265,8 +405,9 @@ def clear_png_metadata(png_file_name, in_place=False, verbose=False): # Remove metadata with Image.open(png_file_name) as img: - clean_img = Image.new(img.mode, img.size) - clean_img.putdata(list(img.getdata())) + # Copy the image to strip all metadata while preserving pixel data, + # palette (for "P" mode), and alpha channel (for "RGBA"/"LA" etc.). + clean_img = img.copy() clean_img.save(output_path, format="PNG") if verbose: @@ -288,7 +429,9 @@ def clear_jpeg_metadata(jpeg_file_name, in_place=False, verbose=False): :type verbose: bool :return: path to cleaned JPEG file """ - if not os.path.exists(jpeg_file_name) or not jpeg_file_name.lower().endswith((".jpg", ".jpeg")): + if not os.path.exists(jpeg_file_name) or not jpeg_file_name.lower().endswith( + (".jpg", ".jpeg") + ): return with open(jpeg_file_name, "rb") as f: @@ -313,7 +456,7 @@ def clear_jpeg_metadata(jpeg_file_name, in_place=False, verbose=False): break continue length = (data[i] << 8) | data[i + 1] - payload = data[i:i + length] + payload = data[i : i + length] i += length if JPEG_APP_FIRST <= marker <= JPEG_APP_LAST or marker == JPEG_COM: continue @@ -379,7 +522,7 @@ def skip_sub_blocks(start): i = 13 if packed & 0x80: gct = 3 * (1 << ((packed & 0x07) + 1)) - out += data[i:i + gct] + out += data[i : i + gct] i += gct # Walk data stream per GIF89a; drop metadata-bearing extensions only. @@ -391,11 +534,11 @@ def skip_sub_blocks(start): if b == GIF_EXTENSION_INTRODUCER and i + 1 < n: label = data[i + 1] if label == GIF_EXT_GRAPHIC_CONTROL: - out += data[i:i + 8] + out += data[i : i + 8] i += 8 elif label == GIF_EXT_APPLICATION and i + 2 < n: ident_len = data[i + 2] - ident = data[i + 3:i + 3 + ident_len] + ident = data[i + 3 : i + 3 + ident_len] j = skip_sub_blocks(i + 3 + ident_len) if ident == GIF_APP_EXT_NETSCAPE_IDENTIFIER: out += data[i:j] @@ -431,6 +574,48 @@ def skip_sub_blocks(start): return output_path +def extract_metadata(microsoft_file_name): + """ + Extract metadata from the given Microsoft file. + + :param microsoft_file_name: name of Microsoft file + :type microsoft_file_name: str + :return: dictionary containing the metadata fields + :rtype: dict + """ + microsoft_format = get_file_format(microsoft_file_name) + if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: + return {} + + unzipped_dir, source_file = extract(microsoft_file_name) + + try: + doc_props_dir = os.path.join(unzipped_dir, "docProps") + core_xml_path = os.path.join(doc_props_dir, "core.xml") + app_xml_path = os.path.join(doc_props_dir, "app.xml") + + metadata = {} + + # Extract from core.xml + if os.path.exists(core_xml_path): + tree = ET.parse(core_xml_path) + for xml_element in tree.iter(): + for personal_field, associated_xml_tag in CORE_XML_MAP.items(): + if associated_xml_tag in xml_element.tag: + metadata[personal_field] = xml_element.text or "" + + # Extract from app.xml + if os.path.exists(app_xml_path): + tree = ET.parse(app_xml_path) + for xml_element in tree.iter(): + for personal_field, associated_xml_tag in APP_XML_MAP.items(): + if associated_xml_tag in xml_element.tag: + metadata[personal_field] = xml_element.text or "" + + return metadata + finally: + _cleanup_extracted(unzipped_dir, source_file, False) # Using the existing cleanup function + CLEAR_HANDLERS = { "docx": clear, "pptx": clear, @@ -461,76 +646,4 @@ def clear_file(file_name, in_place=False, verbose=False): handler = CLEAR_HANDLERS.get(fmt) if handler is None: return None - return handler(file_name, in_place, verbose) - - -def extract_metadata(microsoft_file_name): - """ - Extract all the editable metadata from the given Microsoft file. - - :param microsoft_file_name: name of Microsoft file - :type microsoft_file_name: str - :return: dict containing the extracted metadata - """ - unzipped_dir, _ = extract(microsoft_file_name) - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - - extracted_metadata = {} - - def _extract_metadata_from_xml(xml_path, xml_map): - if os.path.exists(xml_path): - tree = lxml.parse(xml_path) - for xml_element in tree.iter(): - for personal_field, xml_tag in xml_map.items(): - if xml_tag in xml_element.tag: - value = xml_element.text if xml_element.text else "" - extracted_metadata[personal_field] = value.strip() - - _extract_metadata_from_xml(core_xml_path, CORE_XML_MAP) - _extract_metadata_from_xml(app_xml_path, APP_XML_MAP) - - # Clean up - shutil.rmtree(unzipped_dir) - return extracted_metadata - - -def dmeta_help(): - """ - Print DMeta details. - - :return: None - """ - print(OVERVIEW) - print("Repo : https://github.com/openscilab/dmeta") - print("Webpage : https://openscilab.com") - - -def run_dmeta(args): - """ - Run DMeta. - - :param args: input arguments - :type args: argparse.Namespace - :return: None - """ - verbose = args.verbose - if args.clear: - clear_file(args.clear[0], args.inplace, verbose) - elif args.clear_all: - clear_all(args.inplace, verbose) - elif args.update: - if not args.config: - raise DMetaBaseError(UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR) - else: - update(args.config[0], args.update[0], args.inplace, verbose) - elif args.update_all: - if not args.config: - raise DMetaBaseError(UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR) - else: - update_all(args.config[0], args.inplace, verbose) - else: - tprint("DMeta") - tprint("V:" + DMETA_VERSION) - dmeta_help() + return handler(file_name, in_place, verbose) \ No newline at end of file diff --git a/tests/test_dmeta.py b/tests/test_dmeta.py index a88b65a..0f6a782 100644 --- a/tests/test_dmeta.py +++ b/tests/test_dmeta.py @@ -7,18 +7,28 @@ from dmeta.functions import extract_metadata +def _safe_remove(file_path: str) -> None: + """Remove a file if it exists and differs from the original.""" + if file_path and os.path.exists(file_path): + os.remove(file_path) + + TESTS_DIR_PATH = os.path.join(os.getcwd(), "tests") + def test1(): # clear a single .docx file [not inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.docx") output_path = clear(microsoft_file_name) + assert output_path is not None, "clear() returned None" for value in extract_metadata(output_path).values(): assert value == "" + # Clean up created file + _safe_remove(output_path) def test2(): - # clear a single .docx file [inplace] + # clear a single .pptx file [inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.pptx") _ = clear(microsoft_file_name, in_place=True) for value in extract_metadata(microsoft_file_name).values(): @@ -27,42 +37,79 @@ def test2(): def test3(): # clear all existing .docx files [not inplace] - os.chdir(TESTS_DIR_PATH) - clear_all() + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + # Get list of files before clearing + original_files = set(os.listdir(".")) + clear_all() + # Clean up newly created files + new_files = set(os.listdir(".")) - original_files + for new_file in new_files: + if os.path.isfile(new_file): + os.remove(new_file) + finally: + os.chdir(original_dir) def test4(): - # clear all existing .docx files [inplace] - os.chdir(TESTS_DIR_PATH) - clear_all(in_place=True) + # clear all existing files [inplace] + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + clear_all(in_place=True) + finally: + os.chdir(original_dir) def test5(): # update a single .docx file [not inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.docx") _author = extract_metadata(microsoft_file_name)["authors"] - output_path = update(os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=False) + output_path = update( + os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=False + ) + assert output_path is not None, "update() returned None" assert extract_metadata(microsoft_file_name)["authors"] == _author assert extract_metadata(output_path)["authors"] == "UPDATED-AUTHOR" + # Clean up created file + _safe_remove(output_path) def test6(): # update a single .docx file [inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.docx") - _ = update(os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=True) + _ = update( + os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=True + ) assert extract_metadata(microsoft_file_name)["authors"] == "UPDATED-AUTHOR" def test7(): # update all existing .docx files [not inplace] - os.chdir(TESTS_DIR_PATH) - update_all(os.path.join(TESTS_DIR_PATH, "config.json")) + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + # Get list of files before update + original_files = set(os.listdir(".")) + update_all(os.path.join(TESTS_DIR_PATH, "config.json")) + # Clean up newly created files + new_files = set(os.listdir(".")) - original_files + for new_file in new_files: + if os.path.isfile(new_file): + os.remove(new_file) + finally: + os.chdir(original_dir) def test8(): - # update all existing .docx files [inplace] - os.chdir(TESTS_DIR_PATH) - update_all(os.path.join(TESTS_DIR_PATH, "config.json"), in_place=True) + # update all existing files [inplace] + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + update_all(os.path.join(TESTS_DIR_PATH, "config.json"), in_place=True) + finally: + os.chdir(original_dir) def test9(): @@ -72,20 +119,27 @@ def test9(): with Image.open(png_file) as img: assert img.info == {} + def test10(): # clear the metadata of the .png file [not inplace] png_file = os.path.join(TESTS_DIR_PATH, "test.png") output_path = clear_png_metadata(png_file, in_place=False, verbose=False) + assert output_path is not None, "clear_png_metadata() returned None" with Image.open(output_path) as img: assert img.info == {} + # Clean up created file + _safe_remove(output_path) def test11(): # clear the metadata of the .jpg file [not inplace] jpeg_file = os.path.join(TESTS_DIR_PATH, "test.jpg") output_path = clear_jpeg_metadata(jpeg_file, in_place=False, verbose=False) + assert output_path is not None, "clear_jpeg_metadata() returned None" with Image.open(output_path) as img: assert img.info == {} + # Clean up created file + _safe_remove(output_path) def test12(): @@ -100,8 +154,11 @@ def test13(): # clear the metadata of the .gif file [not inplace] gif_file = os.path.join(TESTS_DIR_PATH, "test.gif") output_path = clear_gif_metadata(gif_file, in_place=False, verbose=False) + assert output_path is not None, "clear_gif_metadata() returned None" with Image.open(output_path) as img: assert "comment" not in img.info + # Clean up created file + _safe_remove(output_path) def test14():