diff --git a/pyproject.toml b/pyproject.toml index 7984e23..9e80fe0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,6 +2,7 @@ name = "SigMF" description = "Easily interact with Signal Metadata Format (SigMF) recordings." keywords = ["gnuradio", "radio"] +license = { file = "COPYING-LGPL" } classifiers = [ "Development Status :: 5 - Production/Stable", "License :: OSI Approved :: GNU Lesser General Public License v3 or later (LGPLv3+)", @@ -45,6 +46,7 @@ dependencies = [ [tool.setuptools] packages = ["sigmf"] +license-files = [] [tool.setuptools.dynamic] version = {attr = "sigmf.__version__"} readme = {file = ["README.md"], content-type = "text/markdown"} @@ -100,7 +102,7 @@ profile = "black" legacy_tox_ini = ''' [tox] skip_missing_interpreters = True - envlist = py{37,38,39,310,311,312,313} + envlist = py{37,38,39,310,311,312,313,314} [testenv] usedevelop = True diff --git a/sigmf/__init__.py b/sigmf/__init__.py index 7383046..edcbb62 100644 --- a/sigmf/__init__.py +++ b/sigmf/__init__.py @@ -5,7 +5,7 @@ # SPDX-License-Identifier: LGPL-3.0-or-later # version of this python module -__version__ = "1.6.2" +__version__ = "1.7.0" # matching version of the SigMF specification __specification__ = "1.2.6" diff --git a/sigmf/archive.py b/sigmf/archive.py index e3abec9..f0bef9a 100644 --- a/sigmf/archive.py +++ b/sigmf/archive.py @@ -12,7 +12,7 @@ import tempfile from pathlib import Path -from .error import SigMFFileError +from .error import SigMFFileError, SigMFFileExistsError SIGMF_ARCHIVE_EXT = ".sigmf" SIGMF_METADATA_EXT = ".sigmf-meta" @@ -22,11 +22,7 @@ class SigMFArchive: """ - Archive a SigMFFile - - A `.sigmf` file must include both valid metadata and data. - If `self.data_file` is not set or the requested output file - is not writable, raises `SigMFFileError`. + Archive a SigMFFile into a tar file. Parameters ---------- @@ -35,7 +31,7 @@ class SigMFArchive: A SigMFFile object with valid metadata and data_file. name : PathLike | str | bytes - Path to archive file to create. If file exists, overwrite. + Path to archive file to create. If `name` doesn't end in .sigmf, it will be appended. For example: if `name` == "/tmp/archive1", then the following archive will be created: @@ -56,12 +52,21 @@ class SigMFArchive: - archive1/ - archive1.sigmf-meta - archive1.sigmf-data + + overwrite : bool, default False + If False, raise exception if archive file already exists. + + Raises + ------ + SigMFFileError + If `sigmffile` has no data_file set, or if `name` is not writable. + """ - def __init__(self, sigmffile, name=None, fileobj=None): + def __init__(self, sigmffile, name=None, fileobj=None, overwrite=False): is_buffer = fileobj is not None self.sigmffile = sigmffile - self.path, arcname, fileobj = self._resolve(name, fileobj) + self.path, arcname, fileobj = self._resolve(name, fileobj, overwrite) self._ensure_data_file_set() self._validate() @@ -106,13 +111,22 @@ def _ensure_data_file_set(self): def _validate(self): self.sigmffile.validate() - def _resolve(self, name, fileobj): + def _resolve(self, name, fileobj, overwrite=False): """ Resolve both (name, fileobj) into (path, arcname, fileobj) given either or both. + Parameters + ---------- + name : PathLike | str | bytes | None + Path to archive file to create. + fileobj : BufferedWriter | None + Open file handle object. + overwrite : bool, default False + If False, raise exception if archive file already exists. + Returns ------- - path : PathLike + path : Path Path of the archive file. arcname : str Name of the sigmf object within the archive. @@ -144,6 +158,10 @@ def _resolve(self, name, fileobj): raise SigMFFileError(f"Invalid extension ({path.suffix} != {SIGMF_ARCHIVE_EXT}).") arcname = path.stem + # check if file exists and overwrite is disabled + if not overwrite and path.exists(): + raise SigMFFileExistsError(path, "Archive file") + try: fileobj = open(path, "wb") except (OSError, IOError) as exc: diff --git a/sigmf/convert/__main__.py b/sigmf/convert/__main__.py index 937c2b3..de7d6c1 100644 --- a/sigmf/convert/__main__.py +++ b/sigmf/convert/__main__.py @@ -60,6 +60,7 @@ def main() -> None: exclusive_group.add_argument( "--ncd", action="store_true", help="Output .sigmf-meta only and process as a Non-Conforming Dataset (NCD)" ) + parser.add_argument("--overwrite", action="store_true", help="Overwrite existing output files") parser.add_argument("--version", action="version", version=f"%(prog)s v{toolversion}") args = parser.parse_args() @@ -89,11 +90,23 @@ def main() -> None: if magic_bytes == b"RIFF": # WAV file - _ = wav_to_sigmf(wav_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd) + _ = wav_to_sigmf( + wav_path=input_path, + out_path=output_path, + create_archive=args.archive, + create_ncd=args.ncd, + overwrite=args.overwrite, + ) elif magic_bytes == b"BLUE": # BLUE file - _ = blue_to_sigmf(blue_path=input_path, out_path=output_path, create_archive=args.archive, create_ncd=args.ncd) + _ = blue_to_sigmf( + blue_path=input_path, + out_path=output_path, + create_archive=args.archive, + create_ncd=args.ncd, + overwrite=args.overwrite, + ) else: raise SigMFConversionError( diff --git a/sigmf/convert/blue.py b/sigmf/convert/blue.py index 6349f4f..46cd6c8 100644 --- a/sigmf/convert/blue.py +++ b/sigmf/convert/blue.py @@ -498,8 +498,9 @@ def _build_common_metadata( tuple[dict, dict] (global_info, capture_info) dictionaries. """ - # helper to look up extended header values by tag + def get_tag(tag): + """helper to look up extended header values by tag""" for entry in h_extended: if entry["tag"] == tag: return entry["value"] @@ -670,6 +671,7 @@ def construct_sigmf( h_extended: list, is_metadata_only: bool = False, create_archive: bool = False, + overwrite: bool = False, ) -> SigMFFile: """ Built & write a SigMF object from BLUE metadata. @@ -688,6 +690,8 @@ def construct_sigmf( If True, creates a metadata-only SigMF file. create_archive : bool, optional When True, package output as SigMF archive instead of a meta/data pair. + overwrite : bool, optional + If False, raise exception if output files already exist. Returns ------- @@ -723,12 +727,12 @@ def construct_sigmf( meta.add_capture(0, metadata=capture_info) if create_archive: - meta.tofile(filenames["archive_fn"], toarchive=True) + meta.tofile(filenames["archive_fn"], toarchive=True, overwrite=overwrite) log.info("wrote SigMF archive to %s", filenames["archive_fn"]) # metadata returned should be for this archive meta = fromfile(filenames["archive_fn"]) else: - meta.tofile(filenames["meta_fn"], toarchive=False) + meta.tofile(filenames["meta_fn"], toarchive=False, overwrite=overwrite) log.info("wrote SigMF metadata to %s", filenames["meta_fn"]) log.debug("created %r", meta) @@ -790,6 +794,7 @@ def blue_to_sigmf( out_path: Optional[str] = None, create_archive: bool = False, create_ncd: bool = False, + overwrite: bool = False, ) -> SigMFFile: """ Read a MIDAS Bluefile, optionally write SigMF, return associated SigMF object. @@ -804,6 +809,8 @@ def blue_to_sigmf( When True, package output as a .sigmf archive. create_ncd : bool, optional When True, create Non-Conforming Dataset with header_bytes and trailing_bytes. + overwrite : bool, optional + If False, raise exception if output files already exist. Returns ------- @@ -846,7 +853,7 @@ def blue_to_sigmf( # write NCD metadata to specified output path if provided if out_path is not None: - ncd_meta.tofile(filenames["meta_fn"]) + ncd_meta.tofile(filenames["meta_fn"], overwrite=overwrite) log.info("wrote SigMF non-conforming metadata to %s", filenames["meta_fn"]) return ncd_meta @@ -872,6 +879,7 @@ def blue_to_sigmf( h_extended=h_extended, is_metadata_only=metadata_only, create_archive=create_archive, + overwrite=overwrite, ) log.debug(">>>>>>>>> Fixed Header") diff --git a/sigmf/convert/wav.py b/sigmf/convert/wav.py index 4fc226d..c298b0a 100644 --- a/sigmf/convert/wav.py +++ b/sigmf/convert/wav.py @@ -19,6 +19,7 @@ from .. import SigMFFile from .. import __version__ as toolversion from .. import fromfile +from ..error import SigMFFileExistsError from ..sigmffile import get_sigmf_filenames from ..utils import SIGMF_DATETIME_ISO8601_FMT, get_data_type_str @@ -78,6 +79,7 @@ def wav_to_sigmf( out_path: Optional[str] = None, create_archive: bool = False, create_ncd: bool = False, + overwrite: bool = False, ) -> SigMFFile: """ Read a wav, optionally write sigmf, return associated SigMF object. @@ -92,6 +94,8 @@ def wav_to_sigmf( When True, package output as a .sigmf archive. create_ncd : bool, optional When True, create Non-Conforming Dataset with header_bytes and trailing_bytes. + overwrite : bool, optional + If False, raise exception if output files already exist. Returns ------- @@ -172,7 +176,7 @@ def wav_to_sigmf( filenames = get_sigmf_filenames(out_path) output_dir = filenames["meta_fn"].parent output_dir.mkdir(parents=True, exist_ok=True) - meta.tofile(filenames["meta_fn"], toarchive=False) + meta.tofile(filenames["meta_fn"], toarchive=False, overwrite=overwrite) log.info("wrote SigMF non-conforming metadata to %s", filenames["meta_fn"]) log.debug("created %r", meta) @@ -197,20 +201,25 @@ def wav_to_sigmf( meta = SigMFFile(data_file=data_path, global_info=global_info) meta.add_capture(0, metadata=capture_info) - meta.tofile(filenames["archive_fn"], toarchive=True) + meta.tofile(filenames["archive_fn"], toarchive=True, overwrite=overwrite) log.info("wrote SigMF archive to %s", filenames["archive_fn"]) # metadata returned should be for this archive meta = fromfile(filenames["archive_fn"]) else: # write separate meta and data files data_path = filenames["data_fn"] + + # check if data file exists when overwrite is disabled + if not overwrite and data_path.exists(): + raise SigMFFileExistsError(data_path, "Data file") + wav_data.tofile(data_path) log.info("wrote SigMF dataset to %s", data_path) meta = SigMFFile(data_file=data_path, global_info=global_info) meta.add_capture(0, metadata=capture_info) - meta.tofile(filenames["meta_fn"], toarchive=False) + meta.tofile(filenames["meta_fn"], toarchive=False, overwrite=overwrite) log.info("wrote SigMF metadata to %s", filenames["meta_fn"]) log.debug("created %r", meta) diff --git a/sigmf/error.py b/sigmf/error.py index 9a1ca5f..1551177 100644 --- a/sigmf/error.py +++ b/sigmf/error.py @@ -24,5 +24,14 @@ class SigMFFileError(SigMFError): """Exceptions related to reading or writing SigMF files or archives.""" +class SigMFFileExistsError(SigMFFileError): + """Exception raised when a file already exists and overwrite is disabled.""" + + def __init__(self, file_path, file_type="File"): + super().__init__(f"{file_type} {file_path} already exists. Use overwrite=True to overwrite.") + self.file_path = file_path + self.file_type = file_type + + class SigMFConversionError(SigMFError): """Exceptions related to converting to SigMF format.""" diff --git a/sigmf/sigmffile.py b/sigmf/sigmffile.py index 79a5186..5ae4291 100644 --- a/sigmf/sigmffile.py +++ b/sigmf/sigmffile.py @@ -23,7 +23,13 @@ SIGMF_METADATA_EXT, SigMFArchive, ) -from .error import SigMFAccessError, SigMFConversionError, SigMFError, SigMFFileError +from .error import ( + SigMFAccessError, + SigMFConversionError, + SigMFError, + SigMFFileError, + SigMFFileExistsError, +) from .utils import dict_merge, get_magic_bytes @@ -790,16 +796,24 @@ def validate(self): """ validate.validate(self._metadata, self.get_schema()) - def archive(self, name=None, fileobj=None): + def archive(self, name=None, fileobj=None, overwrite=False): """Dump contents to SigMF archive format. `name` and `fileobj` are passed to SigMFArchive and are defined there. - """ - archive = SigMFArchive(self, name, fileobj) + Parameters + ---------- + name : str, optional + Name of the archive file to create. If None, a temporary file will be created. + fileobj : file-like object, optional + A file-like object to write the archive to. If None, a file will be created at `name`. + overwrite : bool, default False + If False, raise exception if archive file already exists. + """ + archive = SigMFArchive(self, name, fileobj, overwrite=overwrite) return archive.path - def tofile(self, file_path, pretty=True, toarchive=False, skip_validate=False): + def tofile(self, file_path, pretty=True, toarchive=False, skip_validate=False, overwrite=False): """ Write metadata file or full archive containing metadata & dataset. @@ -812,13 +826,21 @@ def tofile(self, file_path, pretty=True, toarchive=False, skip_validate=False): toarchive : bool, default False If True will write both dataset & metadata into SigMF archive format as a single `tar` file. If False will only write metadata to `sigmf-meta`. + skip_validate : bool, default False + Skip validation of metadata before writing. + overwrite : bool, default False + If False, raise exception if output file already exists. """ if not skip_validate: self.validate() fns = get_sigmf_filenames(file_path) + if toarchive: - self.archive(fns["archive_fn"]) + self.archive(fns["archive_fn"], overwrite=overwrite) else: + # check if metadata file exists + if not overwrite and fns["meta_fn"].exists(): + raise SigMFFileExistsError(fns["meta_fn"], "Metadata file") with open(fns["meta_fn"], "w") as fp: self.dump(fp, pretty=pretty) fp.write("\n") # text files should end in carriage return @@ -1076,7 +1098,7 @@ def get_collection_field(self, key: str, default=None): """ return self._metadata[self.COLLECTION_KEY].get(key, default) - def tofile(self, file_path, pretty: bool = True) -> None: + def tofile(self, file_path, pretty: bool = True, overwrite: bool = False) -> None: """ Write metadata file @@ -1086,8 +1108,15 @@ def tofile(self, file_path, pretty: bool = True) -> None: Location to save. pretty : bool, default True When True will write more human-readable output, otherwise will be flat JSON. + overwrite : bool, default False + If False, raise exception if collection file already exists. """ filenames = get_sigmf_filenames(file_path) + + # check if collection file exists + if not overwrite and filenames["collection_fn"].exists(): + raise SigMFFileExistsError(filenames["collection_fn"], "Collection file") + with open(filenames["collection_fn"], "w") as handle: self.dump(handle, pretty=pretty) handle.write("\n") # text files should end in carriage return diff --git a/tests/test_archive.py b/tests/test_archive.py index cd500b6..36abfa8 100644 --- a/tests/test_archive.py +++ b/tests/test_archive.py @@ -47,13 +47,13 @@ def test_archive_creation_requires_data_file(self): """Test that archiving without data file raises error""" self.sigmf_object.data_file = None with self.assertRaises(error.SigMFFileError): - self.sigmf_object.archive(name=self.temp_path_archive) + self.sigmf_object.archive(name=self.temp_path_archive, overwrite=True) def test_archive_creation_validates_metadata(self): """Test that invalid metadata raises error""" del self.sigmf_object._metadata["global"]["core:datatype"] # required field with self.assertRaises(jsonschema.exceptions.ValidationError): - self.sigmf_object.archive(name=self.temp_path_archive) + self.sigmf_object.archive(name=self.temp_path_archive, overwrite=True) def test_archive_creation_validates_extension(self): """Test that wrong extension raises error""" diff --git a/tests/test_archivereader.py b/tests/test_archivereader.py index e93e24d..80552b3 100644 --- a/tests/test_archivereader.py +++ b/tests/test_archivereader.py @@ -52,7 +52,7 @@ def test_access_data_without_untar(self): SigMFFile.NUM_CHANNELS_KEY: num_channels, }, ) - temp_meta.tofile(temp_archive.name, toarchive=True) + temp_meta.tofile(temp_archive.name, toarchive=True, overwrite=True) readback = SigMFArchiveReader(temp_archive.name) readback_samples = readback[:] @@ -85,7 +85,7 @@ def test_access_data_without_untar(self): def test_archiveread_data_file_unchanged(test_sigmffile): with NamedTemporaryFile(suffix=".sigmf") as temp_file: input_samples = test_sigmffile.read_samples() - test_sigmffile.archive(temp_file.name) + test_sigmffile.archive(temp_file.name, overwrite=True) arc = sigmf.fromfile(temp_file.name) output_samples = arc.read_samples() diff --git a/tests/test_collection.py b/tests/test_collection.py index 0f80660..72b4e5a 100644 --- a/tests/test_collection.py +++ b/tests/test_collection.py @@ -57,15 +57,15 @@ def test_load_collection(self, subdir: str) -> None: metadata = copy.deepcopy(TEST_METADATA) meta1 = SigMFFile(metadata=metadata, data_file=data_path1) meta2 = SigMFFile(metadata=metadata, data_file=data_path2) - meta1.tofile(meta_path1) - meta2.tofile(meta_path2) + meta1.tofile(meta_path1, overwrite=True) + meta2.tofile(meta_path2, overwrite=True) # create collection collection = SigMFCollection( metafiles=[meta_name1, meta_name2], base_path=str(self.temp_dir / subdir), ) - collection.tofile(collection_path) + collection.tofile(collection_path, overwrite=True) # load collection collection_loopback = fromfile(collection_path) diff --git a/tests/test_convert_blue.py b/tests/test_convert_blue.py index 0465963..e81f850 100644 --- a/tests/test_convert_blue.py +++ b/tests/test_convert_blue.py @@ -130,9 +130,9 @@ def check_data_and_metadata(self, meta, form, atol): def test_blue_to_sigmf_pair(self) -> None: """test standard blue to sigmf conversion with file pairs""" for form, atol in self.format_tolerance: - sigmf_path = self.tmp_path / f"bar{format}" + sigmf_path = self.tmp_path / f"bar{form}" self.write_minimal(form.encode()) - meta = blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path) + meta = blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, overwrite=True) filenames = sigmf.sigmffile.get_sigmf_filenames(sigmf_path) self.assertTrue(filenames["data_fn"].exists(), "dataset path missing") self.assertTrue(filenames["meta_fn"].exists(), "metadata path missing") @@ -142,8 +142,8 @@ def test_blue_to_sigmf_archive(self) -> None: """test blue to sigmf conversion with archive output""" for form, atol in self.format_tolerance: self.write_minimal(form.encode()) - sigmf_path = self.tmp_path / f"baz{format}" - meta = blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_archive=True) + sigmf_path = self.tmp_path / f"baz{form}" + meta = blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_archive=True, overwrite=True) filenames = sigmf.sigmffile.get_sigmf_filenames(sigmf_path) self.assertTrue(filenames["archive_fn"].exists(), "archive path missing") self.check_data_and_metadata(meta, form, atol) @@ -156,6 +156,45 @@ def test_blue_to_sigmf_ncd(self) -> None: _validate_ncd(self, meta, self.blue_path) self.check_data_and_metadata(meta, form, atol) + def test_pair_overwrite_protection(self) -> None: + """test overwrite protection for pair output""" + self.write_minimal(b"CF") + sigmf_path = self.tmp_path / "overwrite_test" + blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, overwrite=True) + with self.assertRaises(sigmf.error.SigMFFileError) as context: + blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + # test overwrite works + meta2 = blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, overwrite=True) + self.assertIsInstance(meta2, sigmf.SigMFFile) + + def test_archive_overwrite_protection(self) -> None: + """test overwrite protection for archive output""" + self.write_minimal(b"CI") + sigmf_path = self.tmp_path / "archive_overwrite_test" + blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_archive=True, overwrite=True) + with self.assertRaises(sigmf.error.SigMFFileError) as context: + blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_archive=True, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + # test overwrite works + meta2 = blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_archive=True, overwrite=True) + self.assertIsInstance(meta2, sigmf.SigMFFile) + + def test_ncd_overwrite_protection(self) -> None: + """test overwrite protection for NCD output""" + self.write_minimal(b"SU") + sigmf_path = self.tmp_path / "ncd_overwrite_test" + blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_ncd=True, overwrite=True) + with self.assertRaises(sigmf.error.SigMFFileError) as context: + blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_ncd=True, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + # test overwrite works + meta2 = blue_to_sigmf(blue_path=self.blue_path, out_path=sigmf_path, create_ncd=True, overwrite=True) + self.assertIsInstance(meta2, sigmf.SigMFFile) + class TestBlueWithNonSigMFRepo(unittest.TestCase): """BLUE converter tests using external files""" diff --git a/tests/test_convert_wav.py b/tests/test_convert_wav.py index 3f61c77..e2861f4 100644 --- a/tests/test_convert_wav.py +++ b/tests/test_convert_wav.py @@ -66,7 +66,7 @@ def tearDown(self) -> None: def test_wav_to_sigmf_pair(self) -> None: """test standard wav to sigmf conversion with file pairs""" sigmf_path = self.tmp_path / "bar" - meta = wav_to_sigmf(wav_path=str(self.wav_path), out_path=str(sigmf_path)) + meta = wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path) filenames = sigmf.sigmffile.get_sigmf_filenames(sigmf_path) self.assertTrue(filenames["data_fn"].exists(), "dataset path missing") self.assertTrue(filenames["meta_fn"].exists(), "metadata path missing") @@ -76,10 +76,19 @@ def test_wav_to_sigmf_pair(self) -> None: # allow numerical differences due to PCM quantization self.assertTrue(np.allclose(self.audio_data, data, atol=1e-4)) + # test overwrite protection + with self.assertRaises(sigmf.error.SigMFFileError) as context: + wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + # test overwrite works + meta2 = wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, overwrite=True) + self.assertIsInstance(meta2, sigmf.SigMFFile) + def test_wav_to_sigmf_archive(self) -> None: """test wav to sigmf conversion with archive output""" sigmf_path = self.tmp_path / "baz.ext" - meta = wav_to_sigmf(wav_path=str(self.wav_path), out_path=str(sigmf_path), create_archive=True) + meta = wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, create_archive=True) filenames = sigmf.sigmffile.get_sigmf_filenames(sigmf_path) self.assertTrue(filenames["archive_fn"].exists(), "archive path missing") # verify data @@ -88,9 +97,18 @@ def test_wav_to_sigmf_archive(self) -> None: # allow numerical differences due to PCM quantization self.assertTrue(np.allclose(self.audio_data, data, atol=1e-4)) + # test overwrite protection + with self.assertRaises(sigmf.error.SigMFFileError) as context: + wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, create_archive=True, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + # test overwrite works + meta2 = wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, create_archive=True, overwrite=True) + self.assertIsInstance(meta2, sigmf.SigMFFile) + def test_wav_to_sigmf_ncd(self) -> None: """test wav to sigmf conversion as Non-Conforming Dataset""" - meta = wav_to_sigmf(wav_path=str(self.wav_path), create_ncd=True) + meta = wav_to_sigmf(wav_path=self.wav_path, create_ncd=True) _validate_ncd(self, meta, self.wav_path) # verify data @@ -99,6 +117,17 @@ def test_wav_to_sigmf_ncd(self) -> None: self.assertGreater(len(data), 0, "Should read some samples") self.assertTrue(np.allclose(self.audio_data, data, atol=1e-4)) + # test overwrite protection when creating NCD with output path + sigmf_path = self.tmp_path / "ncd_test" + wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, create_ncd=True, overwrite=True) + with self.assertRaises(sigmf.error.SigMFFileError) as context: + wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, create_ncd=True, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + # test overwrite works + meta2 = wav_to_sigmf(wav_path=self.wav_path, out_path=sigmf_path, create_ncd=True, overwrite=True) + self.assertIsInstance(meta2, sigmf.SigMFFile) + class TestWAVWithNonSigMFRepo(unittest.TestCase): """Test WAV converter with real example files if available""" diff --git a/tests/test_ncd.py b/tests/test_ncd.py index a3e2ba7..b90f188 100644 --- a/tests/test_ncd.py +++ b/tests/test_ncd.py @@ -46,7 +46,7 @@ def test_load_ncd(self, subdir: str) -> None: # create metadata file ncd_metadata = copy.deepcopy(TEST_METADATA) meta = SigMFFile(metadata=ncd_metadata, data_file=data_path) - meta.tofile(meta_path) + meta.tofile(meta_path, overwrite=True) # load dataset & validate we can read all the data meta_loopback = fromfile(meta_path) diff --git a/tests/test_sigmffile.py b/tests/test_sigmffile.py index c809e0b..d101bf9 100644 --- a/tests/test_sigmffile.py +++ b/tests/test_sigmffile.py @@ -375,7 +375,7 @@ def test_add_annotation(): def test_fromarchive(test_sigmffile): with tempfile.NamedTemporaryFile(suffix=".sigmf") as temp_file: - archive_path = test_sigmffile.archive(name=temp_file.name) + archive_path = test_sigmffile.archive(name=temp_file.name, overwrite=True) result = sigmf.fromarchive(archive_path=archive_path) assert result._metadata == test_sigmffile._metadata == TEST_METADATA @@ -384,3 +384,106 @@ def test_add_multiple_captures_and_annotations(): sigf = SigMFFile() for idx in range(3): simulate_capture(sigf, idx, 1024) + + +class TestOverwrite(unittest.TestCase): + """test file overwrite protection""" + + def setUp(self): + """create temporary directory and test files""" + self.temp_dir = Path(tempfile.mkdtemp()) + self.test_data_path = self.temp_dir / "test.sigmf-data" + self.test_meta_path = self.temp_dir / "test.sigmf-meta" + self.test_archive_path = self.temp_dir / "test.sigmf" + self.test_collection_path = self.temp_dir / "test.sigmf-collection" + + # write test data file + TEST_FLOAT32_DATA.tofile(self.test_data_path) + + # create test sigmf object + self.sigmf_obj = SigMFFile(TEST_METADATA, data_file=self.test_data_path) + + # create alternate test data for overwrite testing + self.alt_data = np.arange(16, 32, dtype=np.float32) # different data for checksum verification + self.alt_data_path = self.temp_dir / "alt.sigmf-data" + self.alt_data.tofile(self.alt_data_path) + + def tearDown(self): + """clean up temporary directory""" + shutil.rmtree(self.temp_dir) + + def test_prevent_metadata_overwrite(self): + """tofile raises exception when metadata file exists and overwrite=False""" + # create existing metadata file + self.sigmf_obj.tofile(self.test_meta_path) + with self.assertRaises(error.SigMFFileError) as context: + self.sigmf_obj.tofile(self.test_meta_path, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + def test_metadata_overwrite_works(self): + """tofile succeeds when metadata file exists and overwrite=True""" + # create existing metadata file + self.sigmf_obj.tofile(self.test_meta_path) + self.assertTrue(self.test_meta_path.exists()) + original_content = self.test_meta_path.read_text() + original_checksum = self.sigmf_obj.get_global_field("core:sha512") + + # create sigmf object with different data and metadata + alt_sigmf = SigMFFile() + alt_sigmf.set_global_field(SigMFFile.DATATYPE_KEY, "rf32_le") + alt_sigmf.set_global_field("core:description", "overwritten file") + alt_sigmf.set_data_file(self.alt_data_path) + + # should succeed with overwrite=True and content should change + alt_sigmf.tofile(self.test_meta_path, overwrite=True) + self.assertTrue(self.test_meta_path.exists()) + new_content = self.test_meta_path.read_text() + new_checksum = alt_sigmf.get_global_field("core:sha512") + + self.assertNotEqual(original_content, new_content, "file content should change when overwritten") + self.assertNotEqual(original_checksum, new_checksum, "SHA512 checksum should change when overwritten") + + def test_prevent_archive_overwrite(self): + """tofile archive raises exception when archive exists and overwrite=False""" + # create existing archive + self.sigmf_obj.tofile(self.test_archive_path, toarchive=True) + with self.assertRaises(error.SigMFFileError) as context: + self.sigmf_obj.tofile(self.test_archive_path, toarchive=True, overwrite=False) + self.assertIn("already exists", str(context.exception)) + + def test_archive_overwrite_works(self): + """tofile archive succeeds when archive exists and overwrite=True""" + # create existing archive + self.sigmf_obj.tofile(self.test_archive_path, toarchive=True) + self.assertTrue(self.test_archive_path.exists()) + original_checksum = self.sigmf_obj.get_global_field("core:sha512") + + # create sigmf object with different data + alt_sigmf = SigMFFile() + alt_sigmf.set_global_field(SigMFFile.DATATYPE_KEY, "rf32_le") + alt_sigmf.set_global_field("core:description", "overwritten archive") + alt_sigmf.set_data_file(self.alt_data_path) + + # should succeed with overwrite=True and content should change + alt_sigmf.tofile(self.test_archive_path, toarchive=True, overwrite=True) + self.assertTrue(self.test_archive_path.exists()) + + # verify by reading the archive content back + readback_sigmf = sigmf.fromarchive(self.test_archive_path) + new_checksum = readback_sigmf.get_global_field("core:sha512") + + self.assertEqual(readback_sigmf.get_global_field("core:description"), "overwritten archive") + self.assertNotEqual(original_checksum, new_checksum, "SHA512 checksum should change when overwritten") + + def test_default_behavior(self): + """overwrite defaults to False for safety""" + # create existing files + self.sigmf_obj.tofile(self.test_meta_path) + self.sigmf_obj.tofile(self.test_archive_path, toarchive=True) + + # should raise exceptions with default overwrite=False + with self.assertRaises(error.SigMFFileError): + self.sigmf_obj.tofile(self.test_meta_path) + + with self.assertRaises(error.SigMFFileError): + self.sigmf_obj.tofile(self.test_archive_path, toarchive=True)