Check for encoding mismatch with send_c_store() (#893)

docs/changelog/v2.1.0.rst

@@ -13,6 +13,7 @@ Fixes
 * Sanitise filenames for received datasets for non-conformant SOP Instance
  UIDs (:issue:`823`)
 
+
 Enhancements
 ............
 
@@ -27,6 +28,10 @@ Enhancements
  <pynetdicom.service_class.StorageManagementServiceClass>` (:issue:`880`)
 * Added :meth:`~pynetdicom.events.Event.encoded_dataset` to simplify accessing
  the encoded dataset without first decoding it
+* Added a check to :meth:`~pynetdicom.association.Association.send_c_store` to
+ ensure that the *Transfer Syntax UID* matches the encoding of the dataset
+ (:issue:`891`)
+
 
 Changes
 .......

pynetdicom/ae.py

@@ -45,8 +45,8 @@
 
 
 _T = TypeVar("_T")
-ListCXType = List[PresentationContext]
-TSyntaxType = Optional[Union[str, UID, Sequence[Union[str, UID]]]]
+ListCXType = list[PresentationContext]
+TSyntaxType = None | str | UID | Sequence[str] | Sequence[UID]
 
 
 class ApplicationEntity:

pynetdicom/association.py

@@ -18,7 +18,7 @@
 from pydicom import dcmread
 from pydicom.dataset import Dataset
 from pydicom.tag import BaseTag
-from pydicom.uid import UID
+from pydicom.uid import UID, ImplicitVRLittleEndian, ExplicitVRBigEndian
 
 # pylint: disable=no-name-in-module
 from pynetdicom.acse import ACSE
@@ -1881,6 +1881,34 @@ def send_c_store(
  "UID' file meta information element"
  )
 
+ ts_encoding: tuple[bool, bool] = (
+ tsyntax.is_implicit_VR,
+ tsyntax.is_little_endian,
+ )
+ # `dataset` might also be created from scratch
+ ds_encoding = getattr(
+ dataset,
+ "original_encoding",
+ (dataset.is_implicit_VR, dataset.is_little_endian),
+ )
+ if None not in ds_encoding and ts_encoding != ds_encoding:
+ s = ("explicit VR", "implicit VR")[cast(bool, ds_encoding[0])]
+ s += (" big endian", " little endian")[cast(bool, ds_encoding[1])]
+ msg = (
+ f"'dataset' is encoded as {s} but the file meta has a "
+ f"(0002,0010) Transfer Syntax UID of '{tsyntax.name}'"
+ )
+ if ds_encoding == (True, True):
+ LOGGER.warning(f"{msg} - using 'Implicit VR Little Endian' instead")
+ tsyntax = ImplicitVRLittleEndian
+ elif ds_encoding == (False, False):
+ LOGGER.warning(f"{msg} - using 'Explicit VR Big Endian' instead")
+ tsyntax = ExplicitVRBigEndian
+ else:
+ raise AttributeError(
+ f"{msg} - please set an appropriate Transfer Syntax"
+ )
+
  # Get a Presentation Context to use for sending the message
  context = self._get_valid_context(
  sop_class, tsyntax, "scu", allow_conversion=allow_conversion

pynetdicom/tests/test_assoc.py

@@ -1846,6 +1846,69 @@ def handle_store(event):
 
  scp.shutdown()
 
+ def test_dataset_encoding_mismatch(self, caplog):
+ """Tests for when transfer syntax doesn't match dataset encoding."""
+
+ def handle_store(event):
+ return 0x0000
+
+ handlers = [(evt.EVT_C_STORE, handle_store)]
+
+ self.ae = ae = AE()
+ ae.acse_timeout = 5
+ ae.dimse_timeout = 5
+ ae.network_timeout = 5
+ ae.add_supported_context(
+ CTImageStorage,
+ [ExplicitVRBigEndian, ImplicitVRLittleEndian],
+ )
+ scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=handlers)
+
+ ae.add_requested_context(CTImageStorage, ImplicitVRLittleEndian)
+ ae.add_requested_context(CTImageStorage, ExplicitVRBigEndian)
+ assoc = ae.associate("localhost", 11112)
+
+ assert assoc.is_established
+ ds = dcmread(DATASET_PATH)
+ assert ds.is_little_endian
+ assert not ds.is_implicit_VR
+ assert ds.file_meta.TransferSyntaxUID == ExplicitVRLittleEndian
+ ds.is_implicit_VR = True
+ with caplog.at_level(logging.WARNING, logger="pynetdicom"):
+ status = assoc.send_c_store(ds)
+ assert status.Status == 0x0000
+
+ ds.is_implicit_VR = False
+ ds.is_little_endian = False
+ status = assoc.send_c_store(ds)
+ assert status.Status == 0x0000
+
+ ds.is_implicit_VR = False
+ ds.is_little_endian = True
+ ds.file_meta.TransferSyntaxUID = ImplicitVRLittleEndian
+ msg = (
+ "'dataset' is encoded as explicit VR little endian but the file "
+ r"meta has a \(0002,0010\) Transfer Syntax UID of 'Implicit VR "
+ "Little Endian' - please set an appropriate Transfer Syntax"
+ )
+ with pytest.raises(AttributeError, match=msg):
+ status = assoc.send_c_store(ds)
+
+ assoc.release()
+ assert assoc.is_released
+ scp.shutdown()
+
+ assert (
+ "'dataset' is encoded as implicit VR little endian but the file "
+ "meta has a (0002,0010) Transfer Syntax UID of 'Explicit VR "
+ "Little Endian' - using 'Implicit VR Little Endian' instead"
+ ) in caplog.text
+ assert (
+ "'dataset' is encoded as explicit VR big endian but the file "
+ "meta has a (0002,0010) Transfer Syntax UID of 'Explicit VR "
+ "Little Endian' - using 'Explicit VR Big Endian' instead"
+ ) in caplog.text
+
  # Regression tests
  def test_no_send_mismatch(self):
  """Test sending a dataset with mismatched transfer syntax (206)."""