Repository: pydicom/pynetdicom
Pull request #659 (2021-07-06 06:44:59)

use socket.settimeout instead of SO_RCVTIMEO
<!--
Please prefix your PR title with [WIP] for PRs that are in
progress and [MRG] when you consider them ready for review.
-->
Values for SO_RCVTIMEO is in milli seconds for windows while it's in seconds for mac and linux. This is the reason for early time outs in #653. Thanks @arunkant for the fix!

#### Reference issue
#653 

#### Tasks
- [x] Unit tests added that reproduce issue or prove feature is working
- [x] Fix or feature added
- [ ] Documentation and examples updated (if relevant)
- [ ] Unit tests passing and coverage at 100% after adding fix/feature
- [x] Type annotations updated and passing with mypy
- [ ] Apps updated and tested (if relevant)

Need some suggestions on how do I write the tests.

codecov[bot]: # [Codecov](https://codecov.io/gh/pydicom/pynetdicom/pull/659?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pydicom) Report
> Merging [#659](https://codecov.io/gh/pydicom/pynetdicom/pull/659?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pydicom) (a411c7d) into [master](https://codecov.io/gh/pydicom/pynetdicom/commit/9e8a86d70149e2a23ba2492b705165bf24fc7bdb?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pydicom) (9e8a86d) will **decrease** coverage by `0.18%`.
> The diff coverage is `100.00%`.

```diff
@@             Coverage Diff             @@
##            master     #659      +/-   ##
===========================================
- Coverage   100.00%   99.81%   -0.19%     
===========================================
  Files           28       28              
  Lines         8553     8549       -4     
===========================================
- Hits          8553     8533      -20     
- Misses           0       16      +16     
```

| [Impacted Files](https://codecov.io/gh/pydicom/pynetdicom/pull/659?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pydicom) | Coverage Δ | |
|---|---|---|
| [pynetdicom/transport.py](https://codecov.io/gh/pydicom/pynetdicom/pull/659/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pydicom#diff-cHluZXRkaWNvbS90cmFuc3BvcnQucHk=) | `100.00% <100.00%> (ø)` | |
| [pynetdicom/utils.py](https://codecov.io/gh/pydicom/pynetdicom/pull/659/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pydicom#diff-cHluZXRkaWNvbS91dGlscy5weQ==) | `89.25% <0.00%> (-10.75%)` | :arrow_down: |
| [pynetdicom/dul.py](https://codecov.io/gh/pydicom/pynetdicom/pull/659/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pydicom#diff-cHluZXRkaWNvbS9kdWwucHk=) | `98.34% <0.00%> (-1.66%)` | :arrow_down: |

scaramallion: The problem with using `settimeout()` is the same timeout is used for sends which isn't appropriate. It'd be better to do:

```python
import sys

# In AssociationSocket._create_socket
timeout = self.assoc.network_timeout
if timeout is not None:
    if sys.platform == "win32":
        to = pack('l', int(timeout) * 1000)
    else:
        to = pack('ll', int(timeout), int(timeout % 1 * 1000))

    sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVTIMEO, to)
```
And the corresponding change in `AssociationServer.server_bind`

chsasank: As expected #660 passes on everything but windows. Merging it to this branch. Tests should pass fine now.
arunkant: Hi @scaramallion
> The problem with using settimeout() is the same timeout is used for sends which isn't appropriate.

Can you please elaborate more on why timeout for sends is not appropriate
chsasank: DCMTK sets both send and recv timeouts: https://github.com/DCMTK/dcmtk/blob/5c6b398e5e7f594b590133ce30757391d14085ca/dcmnet/libsrc/dcmtrans.cc#L97. In fact they have a warning

> Warning: The macro DISABLE_SEND_TIMEOUT is not supported anymore

Patch:
diff --git a/docs/changelog/v2.0.0.rst b/docs/changelog/v2.0.0.rst
index 9f90f9d14..d08c23ac4 100644
--- a/docs/changelog/v2.0.0.rst
+++ b/docs/changelog/v2.0.0.rst
@@ -41,6 +41,7 @@ Fixes
 * Fixed an exception in the state machine during association negotiation as the acceptor
   when an A-ABORT-RQ PDU is received before the initial A-ASSOCIATE-RQ has been
   processed (:issue:`721`)
+* Fixed network timeouts not working as expected on Windows (:issue:`653`)
 
 
 Enhancements
diff --git a/docs/examples/print.rst b/docs/examples/print.rst
index 6e2590ebe..e05f9e3c0 100644
--- a/docs/examples/print.rst
+++ b/docs/examples/print.rst
@@ -217,7 +217,7 @@ notifications over the same association (and ignore them).
         attr_list.FilmSizeID = 'A4'
 
         # Can only contain a single item, is a reference to the *Film Session*
-        attr_list.ReferencedFilmSessionSequence = [Dataset]
+        attr_list.ReferencedFilmSessionSequence = [Dataset()]
         item = attr_list.ReferencedFilmSessionSequence[0]
         item.ReferencedSOPClassUID = session.SOPClassUID
         item.ReferencedSOPInstanceUID = session.SOPInstanceUID
diff --git a/pynetdicom/dul.py b/pynetdicom/dul.py
index df493181b..1dd0a3963 100644
--- a/pynetdicom/dul.py
+++ b/pynetdicom/dul.py
@@ -267,7 +267,10 @@ class DULServiceProvider(Thread):
         # Try and read the PDU type and length from the socket
         try:
             bytestream.extend(self.socket.recv(6))
-        except (socket.error, socket.timeout):
+        except (socket.error, socket.timeout) as exc:
+            # READ_PDU_EXC_A
+            LOGGER.error("Connection closed before the entire PDU was received")
+            LOGGER.exception(exc)
             # Evt17: Transport connection closed
             self.event_queue.put("Evt17")
             return
@@ -277,14 +280,17 @@ class DULServiceProvider(Thread):
             # Byte 2 is always reserved
             # Bytes 3-6 are always the PDU length
             pdu_type, _, pdu_length = struct.unpack(">BBL", bytestream)
-        except struct.error:
-            # Raised if there's not enough data
+        except struct.error as exc:
+            # READ_PDU_EXC_B
+            LOGGER.error("Insufficient data received to decode the PDU")
             # Evt17: Transport connection closed
             self.event_queue.put("Evt17")
             return
 
         # If the `pdu_type` is unrecognised
         if pdu_type not in (0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07):
+            # READ_PDU_EXC_C
+            LOGGER.error(f"Unknown PDU type received '0x{pdu_type:02X}'")
             # Evt19: Unrecognised or invalid PDU received
             self.event_queue.put("Evt19")
             return
@@ -292,14 +298,22 @@ class DULServiceProvider(Thread):
         # Try and read the rest of the PDU
         try:
             bytestream += self.socket.recv(pdu_length)
-        except (socket.error, socket.timeout):
+        except (socket.error, socket.timeout) as exc:
+            # READ_PDU_EXC_D
+            LOGGER.error("Connection closed before the entire PDU was received")
+            LOGGER.exception(exc)
             # Evt17: Transport connection closed
             self.event_queue.put("Evt17")
             return
 
         # Check that the PDU data was completely read
         if len(bytestream) != 6 + pdu_length:
+            # READ_PDU_EXC_E
             # Evt17: Transport connection closed
+            LOGGER.error(
+                f"The received PDU is shorter than expected ({len(bytestream)} of "
+                f"{6 + pdu_length} bytes received)"
+            )
             self.event_queue.put("Evt17")
             return
 
@@ -308,6 +322,7 @@ class DULServiceProvider(Thread):
             pdu, event = self._decode_pdu(bytestream)
             self.event_queue.put(event)
         except Exception as exc:
+            # READ_PDU_EXC_F
             LOGGER.error("Unable to decode the received PDU data")
             LOGGER.exception(exc)
             # Evt19: Unrecognised or invalid PDU received
diff --git a/pynetdicom/transport.py b/pynetdicom/transport.py
index a110bcf4b..91697eef7 100644
--- a/pynetdicom/transport.py
+++ b/pynetdicom/transport.py
@@ -15,7 +15,6 @@ try:
     _HAS_SSL = True
 except ImportError:
     _HAS_SSL = False
-from struct import pack
 import threading
 from typing import (
     TYPE_CHECKING,
@@ -258,9 +257,9 @@ class AssociationSocket:
 
         *Socket Options*
 
-        - ``SO_REUSEADDR`` is 1
-        - ``SO_RCVTIMEO`` is set to the Association's ``network_timeout``
-          value.
+        - :attr:`socket.SO_REUSEADDR` is 1
+        - :meth:`socket.settimeout` sets the Association's
+          :attr:`~pynetdicom.association.Associaton.network_timeout` value.
 
         Parameters
         ----------
@@ -287,17 +286,8 @@ class AssociationSocket:
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         # If no timeout is set then recv() will block forever if
         #   the connection is kept alive with no data sent
-        # SO_RCVTIMEO: the timeout on receive calls in seconds
-        #   set using a packed binary string containing two uint32s as
-        #   (seconds, microseconds)
         if self.assoc.network_timeout is not None:
-            timeout_seconds = int(self.assoc.network_timeout)
-            timeout_microsec = int(self.assoc.network_timeout % 1 * 1000)
-            sock.setsockopt(
-                socket.SOL_SOCKET,
-                socket.SO_RCVTIMEO,
-                pack("ll", timeout_seconds, timeout_microsec),
-            )
+            sock.settimeout(self.assoc.network_timeout)
 
         sock.bind(address)
 
@@ -787,7 +777,7 @@ class AssociationServer(TCPServer):
         """Bind the socket and set the socket options.
 
         - ``socket.SO_REUSEADDR`` is set to ``1``
-        - ``socket.SO_RCVTIMEO`` is set to
+        - socket.settimeout is used to set to
           :attr:`AE.network_timeout
           <pynetdicom.ae.ApplicationEntity.network_timeout>` unless the
           value is ``None`` in which case it will be left unset.
@@ -805,17 +795,8 @@ class AssociationServer(TCPServer):
         self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         # If no timeout is set then recv() will block forever if
         #   the connection is kept alive with no data sent
-        # SO_RCVTIMEO: the timeout on receive calls in seconds
-        #   set using a packed binary string containing two uint32s as
-        #   (seconds, microseconds)
         if self.ae.network_timeout is not None:
-            timeout_seconds = int(self.ae.network_timeout)
-            timeout_microsec = int(self.ae.network_timeout % 1 * 1000)
-            self.socket.setsockopt(
-                socket.SOL_SOCKET,
-                socket.SO_RCVTIMEO,
-                pack("ll", timeout_seconds, timeout_microsec),
-            )
+            self.socket.settimeout(self.ae.network_timeout)
 
         # Bind the socket to an (address, port)
         #   If address is '' then the socket is reachable by any


Test patch:
diff --git a/pynetdicom/tests/test_assoc.py b/pynetdicom/tests/test_assoc.py
index c3175b8e5..4037cd5ad 100644
--- a/pynetdicom/tests/test_assoc.py
+++ b/pynetdicom/tests/test_assoc.py
@@ -117,9 +117,8 @@ class TestAssociation:
         if self.ae:
             self.ae.shutdown()
 
-    def test_bad_connection(self):
+    def test_bad_connection(self, caplog):
         """Test connect to non-AE"""
-        # sometimes causes hangs in Travis
         ae = AE()
         ae.add_requested_context(Verification)
         ae.acse_timeout = 5
@@ -583,32 +582,35 @@ class TestAssociation:
 
         scp.shutdown()
 
-    def test_unknown_abort_source(self):
+    def test_unknown_abort_source(self, caplog):
         """Test an unknown abort source handled correctly #561"""
+        with caplog.at_level(logging.ERROR, logger="pynetdicom"):
 
-        def handle_req(event):
-            pdu = b"\x07\x00\x00\x00\x00\x04\x00\x00\x01\x00"
-            event.assoc.dul.socket.send(pdu)
-            # Give the requestor time to process the message before killing
-            #   the connection
-            time.sleep(0.1)
+            def handle_req(event):
+                pdu = b"\x07\x00\x00\x00\x00\x04\x00\x00\x01\x00"
+                event.assoc.dul.socket.send(pdu)
+                # Give the requestor time to process the message before killing
+                #   the connection
+                time.sleep(0.1)
 
-        self.ae = ae = AE()
-        ae.acse_timeout = 5
-        ae.dimse_timeout = 5
-        ae.network_timeout = 5
-        ae.add_supported_context(Verification)
+            self.ae = ae = AE()
+            ae.acse_timeout = 5
+            ae.dimse_timeout = 5
+            ae.network_timeout = 5
+            ae.add_supported_context(Verification)
 
-        hh = [(evt.EVT_REQUESTED, handle_req)]
+            hh = [(evt.EVT_REQUESTED, handle_req)]
 
-        scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=hh)
+            scp = ae.start_server(("localhost", 11112), block=False, evt_handlers=hh)
 
-        ae.add_requested_context(Verification)
-        assoc = ae.associate("localhost", 11112)
-        assert not assoc.is_established
-        assert assoc.is_aborted
+            ae.add_requested_context(Verification)
+            assoc = ae.associate("localhost", 11112)
+            assert not assoc.is_established
+            assert assoc.is_aborted
 
-        scp.shutdown()
+            scp.shutdown()
+
+            assert "Insufficient data received to decode the PDU" in caplog.text
 
 
 class TestCStoreSCP:
@@ -1192,23 +1194,26 @@ class TestAssociationSendCEcho:
 
         scp.shutdown()
 
-    def test_network_times_out_requestor(self):
+    def test_network_times_out_requestor(self, caplog):
         """Regression test for #286."""
-        self.ae = ae = AE()
-        ae.add_requested_context(Verification)
-        ae.add_supported_context(Verification)
-        scp = ae.start_server(("localhost", 11112), block=False)
+        with caplog.at_level(logging.ERROR, logger="pynetdicom"):
+            self.ae = ae = AE()
+            ae.add_requested_context(Verification)
+            ae.add_supported_context(Verification)
+            scp = ae.start_server(("localhost", 11112), block=False)
 
-        assoc = ae.associate("localhost", 11112)
-        assert assoc.is_established
-        assert assoc.network_timeout == 60
-        assoc.network_timeout = 0.5
-        assert assoc.network_timeout == 0.5
+            assoc = ae.associate("localhost", 11112)
+            assert assoc.is_established
+            assert assoc.network_timeout == 60
+            assoc.network_timeout = 0.5
+            assert assoc.network_timeout == 0.5
 
-        time.sleep(1.0)
-        assert assoc.is_aborted
+            while not assoc.is_aborted:
+                time.sleep(0.01)
 
-        scp.shutdown()
+            scp.shutdown()
+
+            assert "Network timeout reached" in caplog.text
 
     def test_network_times_out_acceptor(self):
         """Regression test for #286."""
diff --git a/pynetdicom/tests/test_dul.py b/pynetdicom/tests/test_dul.py
index 1535aaeea..dec3058fb 100644
--- a/pynetdicom/tests/test_dul.py
+++ b/pynetdicom/tests/test_dul.py
@@ -67,7 +67,16 @@ class DummyAssociation:
 
 
 class TestDUL:
-    """Run tests on DUL service provider."""
+    """Run tests on DUL service provider.
+
+    DULServiceProvider._read_pdu_data() tests
+    READ_PDU_EXC_A - test_transport.py::TestTLS::test_tls_yes_server_not_client
+    READ_PDU_EXC_B - test_assoc.py::TestAssociation::test_unknown_abort_source
+    READ_PDU_EXC_C - test_assoc.py::TestAssociation::test_bad_connection
+    READ_PDU_EXC_D - test_dul.py::TestDUL::test_recv_short_aborts
+    READ_PDU_EXC_E - test_dul.py::TestDUL::test_recv_missing_data
+    READ_PDU_EXC_F - test_dul.py::TestDUL::test_recv_bad_pdu_aborts
+    """
 
     def setup(self):
         self.scp = None
@@ -135,121 +144,164 @@ class TestDUL:
             dul.to_provider_queue.put("TEST")
             dul._process_recv_primitive()
 
-    def test_recv_failure_aborts(self):
+    def test_recv_failure_aborts(self, caplog):
         """Test connection close during PDU recv causes abort."""
-        commands = [
-            ("recv", None),  # recv a-associate-rq
-            ("send", a_associate_ac),
-            ("send", b"\x07\x00\x00\x00\x00\x04"),
-            ("exit", None),
-        ]
-        self.scp = scp = start_server(commands)
-
-        def handle(event):
-            scp.step()
-            scp.step()
-
-        hh = [(evt.EVT_REQUESTED, handle)]
-
-        ae = AE()
-        ae.acse_timeout = 5
-        ae.dimse_timeout = 5
-        ae.network_timeout = 0.2
-        ae.add_requested_context("1.2.840.10008.1.1")
-        assoc = ae.associate("localhost", 11112, evt_handlers=hh)
-        assert assoc.is_established
-
-        scp.step()  # send short pdu
-        scp.step()  # close connection
-        scp.shutdown()
-
-        # Need to wait for network timeout to expire
-        timeout = 0
-        while not assoc.is_aborted and timeout < 1:
-            time.sleep(0.05)
-            timeout += 0.05
-        assert assoc.is_aborted
-
-    def test_recv_short_aborts(self):
+        with caplog.at_level(logging.ERROR, logger="pynetdicom"):
+            commands = [
+                ("recv", None),  # recv a-associate-rq
+                ("send", a_associate_ac),
+                ("send", b"\x07\x00\x00\x00\x00\x04"),
+                ("exit", None),
+            ]
+            self.scp = scp = start_server(commands)
+
+            def handle(event):
+                scp.step()
+                scp.step()
+
+            hh = [(evt.EVT_REQUESTED, handle)]
+
+            ae = AE()
+            ae.acse_timeout = 5
+            ae.dimse_timeout = 5
+            ae.network_timeout = 0.2
+            ae.add_requested_context("1.2.840.10008.1.1")
+            assoc = ae.associate("localhost", 11112, evt_handlers=hh)
+            assert assoc.is_established
+
+            scp.step()  # send short pdu
+            scp.step()  # close connection
+            scp.shutdown()
+
+            # Need to wait for network timeout to expire
+            timeout = 0
+            while not assoc.is_aborted and timeout < 1:
+                time.sleep(0.05)
+                timeout += 0.05
+            assert assoc.is_aborted
+            assert (
+                "The received PDU is shorter than expected (6 of 10 bytes received)"
+            ) in caplog.text
+
+    def test_recv_short_aborts(self, caplog):
         """Test receiving short PDU causes abort."""
-        commands = [
-            ("recv", None),  # recv a-associate-rq
-            ("send", a_associate_ac),
-            ("send", b"\x07\x00\x00\x00\x00\x04\x00\x00"),  # Send short PDU
-            ("exit", None),
-        ]
-        self.scp = scp = start_server(commands)
-
-        def handle(event):
-            scp.step()
+        with caplog.at_level(logging.ERROR, logger="pynetdicom"):
+            commands = [
+                ("recv", None),  # recv a-associate-rq
+                ("send", a_associate_ac),
+                ("send", b"\x07\x00\x00\x00\x00\x04"),  # Send first 6
+                ("send", b"\x00\x00"),  # Send short remainder
+                ("exit", None),
+            ]
+            self.scp = scp = start_server(commands)
+
+            def handle(event):
+                scp.step()  # recv A-ASSOCIATE-RQ
+                scp.step()  # send A-ASSOCIATE-AC
+
+            hh = [(evt.EVT_REQUESTED, handle)]
+
+            ae = AE()
+            ae.acse_timeout = 5
+            ae.dimse_timeout = 5
+            ae.network_timeout = 0.5
+            ae.add_requested_context("1.2.840.10008.1.1")
+            assoc = ae.associate("localhost", 11112, evt_handlers=hh)
+            assert assoc.is_established
+
+            scp.step()  # send short pdu
+            time.sleep(0.1)
+            assoc.dul.socket.socket.close()
+            # Need to wait for network timeout to expire
+            timeout = 0
+            while not assoc.is_aborted and timeout < 1:
+                time.sleep(0.05)
+                timeout += 0.05
             scp.step()
+            scp.step()  # exit
+            assert assoc.is_aborted
+            scp.shutdown()
 
-        hh = [(evt.EVT_REQUESTED, handle)]
-
-        ae = AE()
-        ae.acse_timeout = 5
-        ae.dimse_timeout = 5
-        ae.network_timeout = 0.2
-        ae.add_requested_context("1.2.840.10008.1.1")
-        assoc = ae.associate("localhost", 11112, evt_handlers=hh)
-        assert assoc.is_established
-
-        scp.step()  # send short pdu
-        # Need to wait for network timeout to expire
-        timeout = 0
-        while not assoc.is_aborted and timeout < 1:
-            time.sleep(0.05)
-            timeout += 0.05
-        assert assoc.is_aborted
-
-        scp.step()
-        scp.shutdown()
+            assert "Connection closed before the entire PDU was received" in caplog.text
 
-    def test_recv_missing_data(self):
+    def test_recv_missing_data(self, caplog):
         """Test missing data when receiving."""
-        commands = [
-            ("recv", None),  # recv a-associate-rq
-            ("send", a_associate_ac),
-            ("send", b"\x07\x00\x00\x00\x00\x02\x00"),  # Send short PDU
-            ("exit", None),
-        ]
-        self.scp = scp = start_server(commands)
+        with caplog.at_level(logging.ERROR, logger="pynetdicom"):
+            commands = [
+                ("recv", None),  # recv a-associate-rq
+                ("send", a_associate_ac),
+                ("send", b"\x07\x00\x00\x00\x00\x02\x00"),  # Send short PDU
+                ("exit", None),
+            ]
+            self.scp = scp = start_server(commands)
+
+            def handle(event):
+                scp.step()
+                scp.step()
+
+            hh = [(evt.EVT_REQUESTED, handle)]
+
+            ae = AE()
+            ae.acse_timeout = 5
+            ae.dimse_timeout = 5
+            # ae.network_timeout = 0.5
+            ae.add_requested_context("1.2.840.10008.1.1")
+            assoc = ae.associate("localhost", 11112, evt_handlers=hh)
+            assert assoc.is_established
+
+            scp.step()  # send short pdu
 
-        def handle(event):
             scp.step()
-            scp.step()
-
-        hh = [(evt.EVT_REQUESTED, handle)]
+            scp.shutdown()
+            assert assoc.is_aborted
+            assert (
+                "The received PDU is shorter than expected (7 of 8 bytes received)"
+            ) in caplog.text
 
-        def recv(nr_bytes):
-            return assoc.dul.socket.socket.recv(6)
-
-        ae = AE()
-        ae.acse_timeout = 5
-        ae.dimse_timeout = 5
-        ae.network_timeout = 0.5
-        ae.add_requested_context("1.2.840.10008.1.1")
-        assoc = ae.associate("localhost", 11112, evt_handlers=hh)
-        assert assoc.is_established
-        assoc.dul.socket.recv = recv
-
-        scp.step()  # send short pdu
-        # Need to wait for network timeout to expire
-        timeout = 0
-        while not assoc.is_aborted and timeout < 1:
-            time.sleep(0.05)
-            timeout += 0.05
-        assert assoc.is_aborted
+    def test_recv_bad_pdu_aborts(self, caplog):
+        """Test receiving undecodable PDU causes abort."""
+        with caplog.at_level(logging.ERROR, logger="pynetdicom"):
+            commands = [
+                ("recv", None),  # recv a-associate-rq
+                ("send", a_associate_ac),
+                ("send", b"\x07\x00\x00\x00\x00\x02\x00\x00"),
+                ("recv", None),
+                ("exit", None),
+            ]
+            self.scp = scp = start_server(commands)
+
+            def handle(event):
+                scp.step()
+                scp.step()
+
+            hh = [(evt.EVT_REQUESTED, handle)]
+
+            ae = AE()
+            ae.acse_timeout = 5
+            ae.dimse_timeout = 5
+            ae.network_timeout = 5
+            ae.add_requested_context("1.2.840.10008.1.1")
+            assoc = ae.associate("localhost", 11112, evt_handlers=hh)
+            assert assoc.is_established
+
+            scp.step()  # send bad PDU
+
+            while assoc.dul.is_alive():
+                time.sleep(0.001)
+
+            scp.step()  # receive abort
+            scp.step()
+            scp.shutdown()
 
-        scp.step()
-        scp.shutdown()
+            assert assoc.is_aborted
+            assert "Unable to decode the received PDU data" in caplog.text
 
-    def test_recv_bad_pdu_aborts(self):
-        """Test receiving undecodable PDU causes abort."""
+    def test_unknown_pdu_aborts(self, caplog):
+        # with caplog.at_level(logging.ERROR, logger="pynetdicom"):
         commands = [
             ("recv", None),  # recv a-associate-rq
             ("send", a_associate_ac),
-            ("send", b"\x07\x00\x00\x00\x00\x02\x00\x00"),
+            ("send", b"\x53\x00\x00\x00\x00\x02"),
             ("recv", None),
             ("exit", None),
         ]
@@ -271,14 +323,14 @@ class TestDUL:
 
         scp.step()  # send bad PDU
 
-        while assoc.dul.is_alive():
-            time.sleep(0.001)
+        time.sleep(0.1)
 
         scp.step()  # receive abort
         scp.step()
         scp.shutdown()
 
         assert assoc.is_aborted
+        # assert "Unknown PDU type received '0x53'" in caplog.text
 
     def test_exception_in_reactor(self):
         """Test that an exception being raised in the DUL reactor kills the
diff --git a/pynetdicom/tests/test_fsm.py b/pynetdicom/tests/test_fsm.py
index 0db0e67d3..80f6e1766 100644
--- a/pynetdicom/tests/test_fsm.py
+++ b/pynetdicom/tests/test_fsm.py
@@ -1310,13 +1310,14 @@ class TestState03(TestStateBase):
             pass
 
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
-
         self.move_to_state(assoc, scp)
 
         scp.step()
+
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
+
         scp.step()
         scp.step()
-
         scp.shutdown()
 
         # Check A-ABORT-RQ sent
@@ -1354,6 +1355,9 @@ class TestState03(TestStateBase):
         self.move_to_state(assoc, scp)
 
         scp.step()
+
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
+
         scp.step()
         scp.shutdown()
 
@@ -1386,10 +1390,12 @@ class TestState03(TestStateBase):
             pass
 
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
-
         self.move_to_state(assoc, scp)
 
         scp.step()
+
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
+
         scp.step()
         scp.shutdown()
 
@@ -1507,6 +1513,7 @@ class TestState03(TestStateBase):
 
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
         self.move_to_state(assoc, scp)
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
 
         scp.step()
         scp.shutdown()
@@ -1572,6 +1579,7 @@ class TestState03(TestStateBase):
 
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
         self.move_to_state(assoc, scp)
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
 
         scp.step()
         scp.shutdown()
@@ -1605,6 +1613,7 @@ class TestState03(TestStateBase):
 
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
         self.move_to_state(assoc, scp)
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
 
         scp.step()
         scp.shutdown()
@@ -1668,6 +1677,8 @@ class TestState03(TestStateBase):
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
         self.move_to_state(assoc, scp)
 
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
+
         scp.step()
         scp.shutdown()
 
@@ -1700,6 +1711,7 @@ class TestState03(TestStateBase):
 
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
         self.move_to_state(assoc, scp)
+        self.wait_on_state(assoc.dul.state_machine, "Sta1")
 
         scp.step()
         scp.shutdown()
@@ -1736,7 +1748,7 @@ class TestState03(TestStateBase):
         scp.step()
         scp.shutdown()
 
-        time.sleep(0.5)
+        self.wait_on_state(assoc.dul.state_machine, "Sta1")
 
         assert fsm._transitions[:2] == ["Sta2", "Sta3"]
         assert fsm._changes[:3] == [
@@ -1807,6 +1819,7 @@ class TestState03(TestStateBase):
 
         assoc.acse._negotiate_as_acceptor = _neg_as_acc
         self.move_to_state(assoc, scp)
+        self.wait_on_state(assoc.dul.state_machine, "Sta13")
 
         scp.step()
         scp.shutdown()
diff --git a/pynetdicom/tests/test_transport.py b/pynetdicom/tests/test_transport.py
index 00129a2d6..c0d9e8707 100644
--- a/pynetdicom/tests/test_transport.py
+++ b/pynetdicom/tests/test_transport.py
@@ -30,7 +30,7 @@ from pynetdicom.transport import (
     T_CONNECT,
 )
 from pynetdicom.sop_class import Verification, RTImageStorage
-from .encoded_pdu_items import p_data_tf_rq
+from .encoded_pdu_items import p_data_tf_rq, a_associate_rq
 from .hide_modules import hide_modules
 from .utils import wait_for_server_socket
 
@@ -364,6 +364,7 @@ class TestTLS:
 
     def test_tls_not_server_not_client(self):
         """Test associating with no TLS on either end."""
+
         self.ae = ae = AE()
         ae.add_supported_context("1.2.840.10008.1.1")
         server = ae.start_server(("localhost", 11112), block=False)
@@ -400,23 +401,25 @@ class TestTLS:
 
         assert len(server.active_associations) == 0
 
-    def test_tls_yes_server_not_client(self, server_context):
+    def test_tls_yes_server_not_client(self, server_context, caplog):
         """Test wrapping the acceptor socket with TLS (and not client)."""
-        self.ae = ae = AE()
-        ae.add_supported_context("1.2.840.10008.1.1")
-        server = ae.start_server(
-            ("localhost", 11112),
-            block=False,
-            ssl_context=server_context,
-        )
+        with caplog.at_level(logging.ERROR, logger="pynetdicom"):
+            self.ae = ae = AE()
+            ae.add_supported_context("1.2.840.10008.1.1")
+            server = ae.start_server(
+                ("localhost", 11112),
+                block=False,
+                ssl_context=server_context,
+            )
 
-        ae.add_requested_context("1.2.840.10008.1.1")
-        assoc = ae.associate("localhost", 11112)
-        assert assoc.is_aborted
+            ae.add_requested_context("1.2.840.10008.1.1")
+            assoc = ae.associate("localhost", 11112)
+            assert assoc.is_aborted
 
-        server.shutdown()
+            server.shutdown()
 
-        assert len(server.active_associations) == 0
+            assert len(server.active_associations) == 0
+            assert "Connection closed before the entire PDU was received" in caplog.text
 
     def test_tls_yes_server_yes_client(self, server_context, client_context):
         """Test associating with TLS on both ends."""
@@ -795,6 +798,62 @@ class TestAssociationServer:
         assoc.release()
         ae.shutdown()
 
+    def test_split_pdu_windows(self):
+        """Regression test for #653"""
+
+        events = []
+
+        def handle_echo(event):
+            events.append(event)
+            return 0x0000
+
+        req_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        req_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        req_sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVTIMEO, pack("ll", 6000, 0))
+
+        req_sock.bind(("localhost", 0))
+
+        self.ae = ae = AE()
+        ae.network_timeout = 1
+        ae.add_supported_context("1.2.840.10008.1.1")
+        ae.add_requested_context("1.2.840.10008.1.1")
+
+        server = ae.start_server(
+            ("localhost", 11112),
+            block=False,
+            evt_handlers=[(evt.EVT_C_ECHO, handle_echo)],
+        )
+
+        # Set AE requestor connection timeout
+        req_sock.settimeout(30)
+        req_sock.connect(("localhost", 11112))
+        req_sock.settimeout(None)
+
+        # Send data directly to the acceptor
+        req_sock.send(a_associate_rq)
+
+        # Give the acceptor time to send the A-ASSOCIATE-AC
+        while not server.active_associations:
+            time.sleep(0.0001)
+
+        assoc = server.active_associations[0]
+        while not assoc.is_established:
+            time.sleep(0.0001)
+
+        # Forcibly split the P-DATA PDU into two TCP segments
+        req_sock.send(p_data_tf_rq[:12])
+        time.sleep(0.5)
+        req_sock.send(p_data_tf_rq[12:])
+
+        # Give the acceptor time to process the C-ECHO-RQ
+        while assoc.is_established and not events:
+            time.sleep(0.0001)
+
+        server.shutdown()
+        req_sock.close()
+
+        assert 1 == len(events)
+
     def test_gc(self):
         """Test garbage collection."""
         self.ae = ae = AE()