refactor(test): reorganize into unit/integration tiers

CMakeLists.txt

@@ -1,8 +1,8 @@
 # Top-level build for PyFlow.
 #
 # All CMake content lives here at the repository root: the crypto_api C
-# library (sources under PyFlow/crypto_api, C tests under test/) plus its
-# install/export rules. The Python package is managed by uv/pytest instead.
+# library (sources under PyFlow/crypto_api, C tests under test/crypto_api)
+# plus its install/export rules. The Python package is managed by uv/pytest instead.
 cmake_minimum_required(VERSION 3.16)
 project(pyflow VERSION 0.1.0 LANGUAGES C)
 
@@ -90,10 +90,10 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/crypto_api.pc.in
 install(FILES ${CMAKE_CURRENT_BINARY_DIR}/crypto_api.pc
         DESTINATION ${CMAKE_INSTALL_LIBDIR}/pkgconfig)
 
-# ---- C tests (sources in test/) -------------------------------------------
+# ---- C tests (sources in test/crypto_api) ---------------------------------
 if(CRYPTO_API_BUILD_TESTS AND BUILD_TESTING)
   foreach(test_name IN ITEMS test_hkdf test_rsa test_ecdh)
-    add_executable(${test_name} test/${test_name}.c)
+    add_executable(${test_name} test/crypto_api/${test_name}.c)
     target_link_libraries(${test_name} PRIVATE crypto_api)
     target_compile_options(${test_name} PRIVATE -Wall -Wextra)
     add_test(NAME ${test_name} COMMAND ${test_name})

README.md

@@ -33,7 +33,7 @@ PyFlow/
 ├── __init__.py / __main__.py         package launcher entry (`python -m PyFlow`)
 ├── flow_setup.py                     launcher implementation
 └── setup.json                        default launcher configuration (generated)
-test/                        C tests (test_hkdf/test_rsa/test_ecdh) + Python tests
+test/                        Python tests (unit/ + integration/), C tests under test/crypto_api/
 docs/                        Sphinx documentation (multi-language)
 CMakeLists.txt               top-level build for the C library and C tests
 ```
@@ -166,7 +166,7 @@ uv run pytest                       # full Python suite
 ctest --test-dir build       # C library tests
 ```
 
-The encrypted-channel tests (`test/test_crypto_rsa.py`, `test/test_crypto_tcp.py`) are skipped automatically when `libcrypto_api` has not been built; everything else runs regardless. The suite passes on Python 3.10–3.14, including the free-threaded (no-GIL) 3.14 build.
+The encrypted-channel tests (`test/unit/network_api/test_crypto_rsa.py`, `test/integration/network_api/test_crypto_tcp.py`) are skipped automatically when `libcrypto_api` has not been built; everything else runs regardless. The suite passes on Python 3.10–3.14, including the free-threaded (no-GIL) 3.14 build.
 
 ## Documentation
 

docs/Crypto/Crypto.rst

@@ -38,7 +38,7 @@ Build it from the repository root:
     cmake --build build --parallel
     ctest --test-dir build --output-on-failure
 
-The C test suite lives in ``test/`` (``test_hkdf``, ``test_rsa``,
+The C test suite lives in ``test/crypto_api/`` (``test_hkdf``, ``test_rsa``,
 ``test_ecdh``) and is built and run together with the library.
 
 The library is built as a shared object (``libcrypto_api.so``) so it can

docs/locale/ja/LC_MESSAGES/Crypto/Crypto.po

@@ -91,12 +91,12 @@ msgstr ""
 msgid "Build it from the repository root:"
 msgstr "リポジトリのルートからビルドします。"
 
-#: ../../Crypto/Crypto.rst:41 04d983635b00498a8ec2d1844472d78e
+#: ../../Crypto/Crypto.rst:41 7ff0eaa67fde42f8a90908d100eda4fa
 msgid ""
-"The C test suite lives in ``test/`` (``test_hkdf``, ``test_rsa``, "
+"The C test suite lives in ``test/crypto_api/`` (``test_hkdf``, ``test_rsa``, "
 "``test_ecdh``) and is built and run together with the library."
 msgstr ""
-"C テストスイートは ``test/`` (``test_hkdf``、``test_rsa``、``test_ecdh``) "
+"C テストスイートは ``test/crypto_api/`` (``test_hkdf``、``test_rsa``、``test_ecdh``) "
 "に存在し、ライブラリと一緒に構築および実行されます。"
 
 #: ../../Crypto/Crypto.rst:44 2d8ebac8b06147e4983c07a6c69296ab

docs/locale/ko/LC_MESSAGES/Crypto/Crypto.po

@@ -91,12 +91,12 @@ msgstr ""
 msgid "Build it from the repository root:"
 msgstr "저장소 루트에서 빌드합니다."
 
-#: ../../Crypto/Crypto.rst:41 04d983635b00498a8ec2d1844472d78e
+#: ../../Crypto/Crypto.rst:41 7ff0eaa67fde42f8a90908d100eda4fa
 msgid ""
-"The C test suite lives in ``test/`` (``test_hkdf``, ``test_rsa``, "
+"The C test suite lives in ``test/crypto_api/`` (``test_hkdf``, ``test_rsa``, "
 "``test_ecdh``) and is built and run together with the library."
 msgstr ""
-"C 테스트 스위트는 ``test/``(``test_hkdf``, ``test_rsa``, ``test_ecdh``)에 있으며 라이브러리와"
+"C 테스트 스위트는 ``test/crypto_api/``(``test_hkdf``, ``test_rsa``, ``test_ecdh``)에 있으며 라이브러리와"
 " 함께 빌드되고 실행됩니다."
 
 #: ../../Crypto/Crypto.rst:44 2d8ebac8b06147e4983c07a6c69296ab

docs/locale/ru/LC_MESSAGES/Crypto/Crypto.po

@@ -94,12 +94,12 @@ msgstr ""
 msgid "Build it from the repository root:"
 msgstr "Соберите его из корня репозитория:"
 
-#: ../../Crypto/Crypto.rst:41 04d983635b00498a8ec2d1844472d78e
+#: ../../Crypto/Crypto.rst:41 7ff0eaa67fde42f8a90908d100eda4fa
 msgid ""
-"The C test suite lives in ``test/`` (``test_hkdf``, ``test_rsa``, "
+"The C test suite lives in ``test/crypto_api/`` (``test_hkdf``, ``test_rsa``, "
 "``test_ecdh``) and is built and run together with the library."
 msgstr ""
-"Набор тестов C находится в ``test/`` (``test_hkdf``, ``test_rsa``, "
+"Набор тестов C находится в ``test/crypto_api/`` (``test_hkdf``, ``test_rsa``, "
 "``test_ecdh``) и собирается и запускается вместе с библиотекой."
 
 #: ../../Crypto/Crypto.rst:44 2d8ebac8b06147e4983c07a6c69296ab

docs/locale/zh_CN/LC_MESSAGES/Crypto/Crypto.po

@@ -91,11 +91,11 @@ msgstr ""
 msgid "Build it from the repository root:"
 msgstr "从存储库根目录构建它："
 
-#: ../../Crypto/Crypto.rst:41 04d983635b00498a8ec2d1844472d78e
+#: ../../Crypto/Crypto.rst:41 7ff0eaa67fde42f8a90908d100eda4fa
 msgid ""
-"The C test suite lives in ``test/`` (``test_hkdf``, ``test_rsa``, "
+"The C test suite lives in ``test/crypto_api/`` (``test_hkdf``, ``test_rsa``, "
 "``test_ecdh``) and is built and run together with the library."
-msgstr "C 测试套件位于“test/”（“test_hkdf”、“test_rsa”、“test_ecdh”）中，并与库一起构建和运行。"
+msgstr "C 测试套件位于“test/crypto_api/”（“test_hkdf”、“test_rsa”、“test_ecdh”）中，并与库一起构建和运行。"
 
 #: ../../Crypto/Crypto.rst:44 2d8ebac8b06147e4983c07a6c69296ab
 #, fuzzy

docs/locale/zh_TW/LC_MESSAGES/Crypto/Crypto.po

@@ -91,11 +91,11 @@ msgstr ""
 msgid "Build it from the repository root:"
 msgstr "從儲存庫根目錄建立它："
 
-#: ../../Crypto/Crypto.rst:41 04d983635b00498a8ec2d1844472d78e
+#: ../../Crypto/Crypto.rst:41 7ff0eaa67fde42f8a90908d100eda4fa
 msgid ""
-"The C test suite lives in ``test/`` (``test_hkdf``, ``test_rsa``, "
+"The C test suite lives in ``test/crypto_api/`` (``test_hkdf``, ``test_rsa``, "
 "``test_ecdh``) and is built and run together with the library."
-msgstr "C 測試套件位於“test/”（“test_hkdf”、“test_rsa”、“test_ecdh”）中，並與程式庫一起建置和運行。"
+msgstr "C 測試套件位於“test/crypto_api/”（“test_hkdf”、“test_rsa”、“test_ecdh”）中，並與程式庫一起建置和運行。"
 
 #: ../../Crypto/Crypto.rst:44 2d8ebac8b06147e4983c07a6c69296ab
 #, fuzzy

pyproject.toml

@@ -66,4 +66,3 @@ exclude = ["test/"]
 
 [tool.pytest.ini_options]
 testpaths = ["test"]
-addopts = "--ignore=test/test_instance_setup"

test/README.md

@@ -0,0 +1,26 @@
+# Test layout
+
+Tests are split first by the real dependencies they touch, then mirrored
+against the package layout one level deep: `PyFlow/<pkg>/` maps to
+`test/<tier>/<pkg>/`, and root-level modules map to the tier root. A new
+subdirectory appears only when the source gains the matching package.
+
+```
+test/
+├── unit/           no network I/O, no subprocesses
+│   ├── test_flow_setup.py   launcher logic with mocked Popen (PyFlow/flow_setup.py)
+│   └── network_api/         object construction, pure logic, real crypto without sockets
+├── integration/    real sockets, but server + client live in the test process
+│   ├── conftest.py          shared server/client/udp fixtures
+│   ├── helpers.py           wait_until / server_ready polling helpers
+│   ├── test_forward_extension.py, test_command_handlers.py, ...
+│   │                        extension modules at the PyFlow package root
+│   └── network_api/         encrypted channel, file transfer, udp, event store, ...
+└── crypto_api/     C library tests (CMake/ctest, not pytest), mirroring PyFlow/crypto_api
+```
+
+The tree is illustrative, not an inventory — it shows the pattern, not
+every file.
+
+- **unit** may load the real crypto library through ctypes; that is not
+  network I/O even though it pulls in a shared object.

test/test_util.h → test/crypto_api/helpers.h

@@ -1,6 +1,6 @@
 /* Shared test harness: check macros with failure accounting. */
-#ifndef PF_CRYPTO_TEST_UTIL_H
-#define PF_CRYPTO_TEST_UTIL_H
+#ifndef PF_CRYPTO_HELPERS_H
+#define PF_CRYPTO_HELPERS_H
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -42,4 +42,4 @@ static int g_failures = 0;
         return 1;                                                               \
     }
 
-#endif /* PF_CRYPTO_TEST_UTIL_H */
+#endif /* PF_CRYPTO_HELPERS_H */

test/test_ecdh.c → test/crypto_api/test_ecdh.c

@@ -6,7 +6,7 @@
 #include <stdlib.h>
 #include <string.h>
 
-#include "test_util.h"
+#include "helpers.h"
 
 #define PRIV_A_FILE "test_ecdh_a_priv.pem"
 #define PASSPHRASE "ecdh passphrase"

test/test_hkdf.c → test/crypto_api/test_hkdf.c

@@ -5,7 +5,7 @@
 #include <stdlib.h>
 #include <string.h>
 
-#include "test_util.h"
+#include "helpers.h"
 
 /* Hex string -> bytes; returns byte count or -1 on error. */
 static int hex2bin(const char *hex, uint8_t *out, size_t out_cap) {

test/test_rsa.c → test/crypto_api/test_rsa.c

@@ -5,7 +5,7 @@
 #include <stdlib.h>
 #include <string.h>
 
-#include "test_util.h"
+#include "helpers.h"
 
 #define PUB_FILE "test_rsa_pub.pem"
 #define PRIV_FILE "test_rsa_priv.pem"

test/conftest.py → test/integration/conftest.py

@@ -1,13 +1,5 @@
-import importlib.util
-import os
-import sys
-
 import pytest
 
-# Only add project root to sys.path if the PyFlow package is not importable
-if importlib.util.find_spec("PyFlow") is None:
-    sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
 import PyFlow.command_control_extension_tcp as ctl
 from PyFlow.network_api.connect_tcp import TCP_Client_Base, TCP_Server_Base
 from PyFlow.network_api.connect_udp import UDP

test/test_util.py → test/integration/helpers.py

@@ -1,4 +1,4 @@
-"""Shared helpers for tests."""
+"""Shared helpers for the integration tests."""
 
 import time
 
@@ -15,9 +15,4 @@
 
 def server_ready(server, timeout=5.0):
     """Wait until a TCP server is running."""
-    deadline = time.monotonic() + timeout
-    while time.monotonic() < deadline:
-        if server.running:
-            return True
-        time.sleep(0.05)
-    return False
+    return wait_until(lambda: server.running, timeout=timeout)

test/test_crypto_tcp.py → test/integration/network_api/test_crypto_tcp.py

@@ -1,4 +1,4 @@
-"""End-to-end tests for RSA-encrypted TCP channels (connect_tcp.py).
+"""Integration tests for RSA-encrypted TCP channels (connect_tcp.py).
 
 A real server and client are connected over loopback with the crypto key
 directories redirected to a temporary location. These tests need the
@@ -15,7 +15,7 @@
 import pytest
 
 
-from test_util import server_ready, wait_until
+from helpers import server_ready, wait_until
 
 from PyFlow.network_api import rsa_crypto
 from PyFlow.network_api.connect_tcp import TCP_Client_Base, TCP_Server_Base
@@ -34,446 +34,446 @@
 _PORT_COUNTER = 65000  # per-test ports avoid cross-test interference
 
 
 
 def _next_port():
     global _PORT_COUNTER
     _PORT_COUNTER += 1
     return _PORT_COUNTER
 
 
 def _new_keypair(paths, lib=None, bits=None):
     """Generate a fresh RSA keypair into (pub_path, pvt_path)."""
     import ctypes
 
     if lib is None:
         lib = rsa_crypto.load_library()
     handle = ctypes.c_void_p()
     assert (
         lib.pf_rsa_keygen(bits or rsa_crypto.DEFAULT_KEY_BITS, ctypes.byref(handle))
         == rsa_crypto.PF_OK
     )
     key = rsa_crypto.RsaKey(handle.value, lib)
     pub_path, pvt_path = paths
     assert lib.pf_rsa_write_priv(key.handle, pvt_path.encode(), None) == 0
     assert lib.pf_rsa_write_pub(key.handle, pub_path.encode()) == 0
     del key
 
 
 @pytest.fixture
 def tcp_pair(tmp_path):
     """A running server plus a connected, handshaken client."""
     ssh_dir = tmp_path / "ssh"
     ssh_dir.mkdir()
     port = _next_port()
     server = TCP_Server_Base(
         host="127.0.0.1",
         port=port,
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=True,
     )
     _redirect_crypto(server.crypto, tmp_path, ssh_dir, "pub_key")
     threading.Thread(target=server.start_TCP_Server, daemon=True).start()
     assert server_ready(server), "server did not start"
 
     client = TCP_Client_Base(
         host="127.0.0.1",
         port=port,
         client_host="127.0.0.1",
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=True,
     )
     _redirect_crypto(client.crypto, tmp_path, ssh_dir, "pub_key_client")
     assert client.connect()
     yield server, client, tmp_path
 
     client.close()
     server.stop()
 
 
 def _redirect_crypto(crypto, tmp_path, ssh_dir, subdir="pub_key"):
     crypto.pvt_key_dir = str(tmp_path / "pvt_key")
     crypto.pub_key_dir = str(tmp_path / subdir)
     crypto.ssh_dir = str(ssh_dir)
     crypto.registry_path = os.path.join(crypto.pub_key_dir, "pub_key.json")
     os.makedirs(crypto.pvt_key_dir, exist_ok=True)
     os.makedirs(crypto.pub_key_dir, exist_ok=True)
 
 
 def _wait_flip(client, timeout=15):
     return wait_until(
         lambda: client.client_socket in client._encrypted_sockets,
         timeout=timeout,
     )
 
 
 def _wait_disconnected(client, timeout=15):
     """Wait until the client's connection is closed (rejected)."""
     return wait_until(
         lambda: not client.running or client.client_socket is None,
         timeout=timeout,
     )
 
 
 def _server_sock(server, client):
     """The server-side socket object for a connected client."""
     return server.clients[client.client_socket.getsockname()]["socket"]
 
 
 def _wait_server_reexchanged(server, client, timeout=20):
     """Wait until the server dropped the socket (decode failure) and
     re-flipped it (re-exchange complete), or the client died."""
     sock = _server_sock(server, client)
     if not wait_until(lambda: sock not in server._encrypted_sockets, timeout=timeout):
         return False
     if not client.running:
         return False
 
     return wait_until(
         lambda: sock in server._encrypted_sockets and client.running,
         timeout=timeout,
     )
 
 
 def _roundtrip(client, message, timeout=2.0):
     """Send a message and capture stdout until the server echoes it."""
     buf = io.StringIO()
     with contextlib.redirect_stdout(buf):
         client.send_message(client.client_socket, message)
         wait_until(lambda: f"msg send: {message}" in buf.getvalue(), timeout=timeout)
     return buf.getvalue()
 
 
 def _registry_entries(crypto):
     import json
 
     if not os.path.exists(crypto.registry_path):
         return {}
     with open(crypto.registry_path, "r", encoding="utf-8") as f:
         return json.load(f)
 
 
 def test_handshake_flips_both_sides_and_stores_keys(tcp_pair):
     server, client, tmp_path = tcp_pair
     assert _wait_flip(client)
     assert client.client_socket in client._encrypted_sockets
     assert wait_until(lambda: len(server._encrypted_sockets) == 1, timeout=5)
     assert len(server._encrypted_sockets) == 1
     for role in ("server", "client"):  # local keypairs generated under pvt_key
 
         assert os.path.exists(os.path.join(str(tmp_path / "pvt_key"), f"{role}_priv.pem"))
         assert os.path.exists(os.path.join(str(tmp_path / "pvt_key"), f"{role}_pub.pem"))
     server_reg = _registry_entries(server.crypto)  # the TOFU registry was written on both sides
 
     client_reg = _registry_entries(client.crypto)
     assert len(server_reg) == 1  # this client, keyed by (ip, client port)
     assert len(client_reg) == 1  # the server, keyed by (ip, server port)
     names = os.listdir(str(tmp_path / "pub_key"))  # exchanged peer keys cached as pem files under each side's pub_key
 
     assert "pub_key.json" in names
     assert any(name.startswith("client_") and name.endswith(".pem") for name in names)
     client_names = os.listdir(str(tmp_path / "pub_key_client"))
     assert "pub_key.json" in client_names
     assert any(name.startswith("server_") and name.endswith(".pem") for name in client_names)
 
 
 def test_encrypted_roundtrip(tcp_pair):
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     out = _roundtrip(client, "hello encrypted")
     assert "[server] msg send: hello encrypted" in out
 
 
 def test_concurrent_encrypted_sends_not_dropped(tcp_pair):
     """Concurrent encrypted sends must not overtake each other on the
     wire: every message must be echoed back exactly once (seq allocation
     and the send are serialised per connection)."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     n_threads, per_thread = 8, 25
     total = n_threads * per_thread
     errors = []
 
     def sender(t):
         for i in range(per_thread):
             try:
                 client.send_message(client.client_socket, f"msg-{t}-{i}")
             except Exception as e:  # noqa: BLE001
                 errors.append(e)
 
     threads = [threading.Thread(target=sender, args=(t,)) for t in range(n_threads)]
     buf = io.StringIO()
     with contextlib.redirect_stdout(buf):
         for th in threads:
             th.start()
         for th in threads:
             th.join()
         echoed = wait_until(
             lambda: buf.getvalue().count("msg send:") >= total, timeout=30
         )
     assert not errors
     assert echoed, (
         f"dropped {total - buf.getvalue().count('msg send:')} of {total} messages"
     )
 
 
 def test_slow_sender_does_not_reorder_frames(tcp_pair):
     """A thread preempted between seq allocation and the wire write must
     not let a later sender overtake it (the peer would drop the earlier
     frame as out-of-order). Deterministic: the first sender's write is
     delayed under the per-connection send lock."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     gate = threading.Event()
     stalled = [0]
 
     class _StallingSocket(socket.socket):
         __slots__ = ()  # keep the exact socket layout so __class__ assignment works
 
         def sendall(self, data):
             if stalled[0] == 0:
                 stalled[0] = 1  # stall exactly one (the first) sender after seq allocation
                 gate.wait()
             super().sendall(data)
 
     client.client_socket.__class__ = _StallingSocket
     buf = io.StringIO()
     with contextlib.redirect_stdout(buf):
         t1 = threading.Thread(target=client.send_message, args=(client.client_socket, "first"))
         t1.start()
         time.sleep(0.05)  # t1 allocates seq 0 and stalls inside sendall
         t2 = threading.Thread(target=client.send_message, args=(client.client_socket, "second"))
         t2.start()
         time.sleep(0.1)  # t2 attempts its send while t1 is still stalled
         gate.set()
         t1.join()
         t2.join()
         echoed = wait_until(
             lambda: "msg send: first" in buf.getvalue()
             and "msg send: second" in buf.getvalue(),
             timeout=10,
         )
     assert echoed, (
         f"frame reordering dropped a message; echoes: {buf.getvalue()!r}"
     )
 
 
 def test_second_connection_passes_tofu(tcp_pair, tmp_path):
     """A second connection presenting the same key is accepted (known
     key, endpoint updated), and the registry gains no duplicate."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     client.close()
     assert wait_until(lambda: client.client_socket not in server._encrypted_sockets, timeout=5)
     client2 = TCP_Client_Base(
         host="127.0.0.1",
         port=server.port,
         client_host="127.0.0.1",
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=True,
     )
     _redirect_crypto(client2.crypto, tmp_path, tmp_path / "ssh", "pub_key_client")
     assert client2.connect()
     try:
         assert _wait_flip(client2)
         assert len(_registry_entries(server.crypto)) == 1  # same key from a (possibly new) source port: accepted, still one entry per peer
 
         assert len(_registry_entries(client2.crypto)) == 1
         out = _roundtrip(client2, "second connection")
         assert "[server] msg send: second connection" in out
     finally:
         client2.close()
 
 
 def test_rotated_client_key_rejected(tcp_pair, tmp_path):
     """A client that rotates its keypair is rejected: on the next
     re-exchange the same endpoint presents a different public key."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     _new_keypair((client.crypto.pub_path, client.crypto.priv_path))  # rotate the client's keypair (as if ~/.ssh was replaced) and reload it
 
     client.crypto.reload_own_key()
     client.send_message(client.client_socket, "trigger rotation")
     assert _wait_disconnected(client)  # the server rejects the changed key and drops the connection
     assert len(_registry_entries(server.crypto)) == 1  # the rejected key must not have been recorded
 
 
 
 def test_rotated_server_key_rejected(tcp_pair, tmp_path):
     """A server that rotates its keypair is rejected by the client: the
     known server endpoint now presents a different public key."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     _new_keypair((server.crypto.pub_path, server.crypto.priv_path))  # rotate the server's keypair and reload it in the running server
 
     server.crypto.reload_own_key()
     client.send_message(client.client_socket, "trigger rotation")
     assert _wait_disconnected(client)  # the client rejects the changed key and closes the connection
 
     assert len(_registry_entries(client.crypto)) == 1
 
 
 def test_custom_keys_pair_used(tmp_path):
     """The server honours a valid user-supplied keypair."""
     ssh_dir = tmp_path / "ssh"
     ssh_dir.mkdir()
     pub_path = str(tmp_path / "custom_pub.pem")
     pvt_path = str(tmp_path / "custom_pvt.pem")
     _new_keypair((pub_path, pvt_path))
     port = _next_port()
     server = TCP_Server_Base(
         host="127.0.0.1",
         port=port,
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=True,
         is_custom_keys=[pub_path, pvt_path],
     )
     _redirect_crypto(server.crypto, tmp_path, ssh_dir, "pub_key")
     threading.Thread(target=server.start_TCP_Server, daemon=True).start()
     assert server_ready(server), "server did not start"
     client = TCP_Client_Base(
         host="127.0.0.1",
         port=port,
         client_host="127.0.0.1",
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=True,
     )
     _redirect_crypto(client.crypto, tmp_path, ssh_dir, "pub_key_client")
     assert client.connect()
     try:
         assert _wait_flip(client)
         assert server.crypto.priv_path == pvt_path
         assert server.crypto.pub_path == pub_path
         out = _roundtrip(client, "custom key channel")
         assert "[server] msg send: custom key channel" in out
     finally:
         client.close()
         server.stop()
 
 
 def test_encryption_disabled_is_plaintext(tmp_path):
     ssh_dir = tmp_path / "ssh"
     ssh_dir.mkdir()
     port = _next_port()
     server = TCP_Server_Base(
         host="127.0.0.1",
         port=port,
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=False,
     )
     threading.Thread(target=server.start_TCP_Server, daemon=True).start()
     assert server_ready(server), "server did not start"
     client = TCP_Client_Base(
         host="127.0.0.1",
         port=port,
         client_host="127.0.0.1",
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=False,
     )
     assert client.connect()
     try:
         assert not client._encrypted_sockets
         assert not server._encrypted_sockets
         out = _roundtrip(client, "plain hello")
         assert "[server] msg send: plain hello" in out
     finally:
         client.close()
         server.stop()
 
 
 def test_replay_of_old_ciphertext_rejected(tcp_pair):
     """Replaying an old ciphertext (correct session nonce, stale seq) is
     dropped by the receiver and does not disturb the connection."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     out = _roundtrip(client, "first message")  # consume seq 0 with a normal message
 
     assert "[server] msg send: first message" in out
     nonce = client._crypto_my_nonce  # replay a ciphertext carrying seq 0 (already consumed) under the current session nonce
 
     body = client.crypto.encrypt_for_peer(client._crypto_server_pem_path, "REPLAY_MARKER")
     buf = io.StringIO()
     with contextlib.redirect_stdout(buf):
         client._send_raw(client.client_socket, f"{nonce}|0|{body}")
         client.send_message(client.client_socket, "after replay")
         wait_until(lambda: "msg send: after replay" in buf.getvalue(), timeout=2.0)
     assert "REPLAY_MARKER" not in buf.getvalue()
 
 
 def test_replay_across_sessions_rejected(tcp_pair):
     """A ciphertext from a previous session (stale nonce) is rejected
     even at seq 0: the session nonce changes every handshake."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     old_nonce = client._crypto_my_nonce
     body = client.crypto.encrypt_for_peer(client._crypto_server_pem_path, "OLD_SESSION")
     client._send_raw(client.client_socket, f"{'a'*32}|0|AAAA")  # force a re-exchange (new nonce): trigger a decode failure on the server with garbage
     assert _wait_server_reexchanged(server, client)
     assert client.running
     assert client._crypto_my_nonce != old_nonce
     buf = io.StringIO()  # replay the old-session ciphertext: nonce mismatch -> dropped
 
     with contextlib.redirect_stdout(buf):
         client._send_raw(client.client_socket, f"{old_nonce}|0|{body}")
         client.send_message(client.client_socket, "after old session")
         wait_until(lambda: "msg send: after old session" in buf.getvalue(), timeout=2.0)
     assert "OLD_SESSION" not in buf.getvalue()
 
 
 def test_unauthenticated_pub_push_ignored(tmp_path, capsys):
     """A /crypto_pub_key push from a connection that never started the
     handshake is ignored: it cannot poison the TOFU registry."""
     ssh_dir = tmp_path / "ssh"
     ssh_dir.mkdir()
     port = _next_port()
     server = TCP_Server_Base(
         host="127.0.0.1",
         port=port,
         is_extend_command=True,
         is_input_command_in_console=False,
         is_enable_encrypto=True,
     )
     _redirect_crypto(server.crypto, tmp_path, ssh_dir, "pub_key")
     threading.Thread(target=server.start_TCP_Server, daemon=True).start()
     assert server_ready(server), "server did not start"
     raw = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     raw.settimeout(3)
     raw.connect(("127.0.0.1", port))
     raw.sendall(b"/crypto_pub_key 0\n")
 
     output = ""
 
     def _saw_ignore_log():
         nonlocal output
         out, _ = capsys.readouterr()
         output += out
         return "ignoring /crypto_pub_key from non-handshaking peer" in output
 
     assert wait_until(_saw_ignore_log, timeout=5.0), "server did not report ignoring the push"
     raw.close()
     try:
         assert len(_registry_entries(server.crypto)) == 0  # no registry entry and no cached key file were created
 
         names = os.listdir(str(tmp_path / "pub_key"))
         assert not any(name.endswith(".pem") for name in names)
     finally:
         server.stop()
 
 
 def test_decode_failure_burst_closes_connection(tcp_pair):
     """After MAX_DECODE_FAILURES consecutive decode failures the
     circuit breaker closes the connection instead of re-exchanging
     forever."""
     server, client, _ = tcp_pair
     assert _wait_flip(client)
     for i in range(3):
         client._send_raw(client.client_socket, f"{'a'*32}|{i}|AAAA")
         if not _wait_server_reexchanged(server, client):
             break  # breaker fired, connection closed
     assert _wait_disconnected(client)
@@ -554,12 +554,8 @@
         is_enable_encrypto=True,
     )
     _redirect_crypto(server.crypto, tmp_path, ssh_dir, "pub_key")
-    recv_dir = os.path.join(
-        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
-        "PyFlow",
-        "network_api",
-        "received_files",
-    )
+    recv_dir = server.file_transfer_dir
+    os.makedirs(recv_dir, exist_ok=True)
     # a failed earlier run may have left stale key files behind; the
     # assertion below checks this run leaves nothing behind, so start clean
     for leftover_name in os.listdir(recv_dir):

test/test_message_event_store.py → test/integration/network_api/test_message_event_store.py

@@ -13,7 +13,7 @@
 
 import pytest
 
-from test_util import server_ready, wait_until
+from helpers import server_ready, wait_until
 
 from PyFlow.network_api.connect_tcp import TCP_Client_Base, TCP_Server_Base
 

test/test_tcp_file_transfer.py → test/integration/network_api/test_tcp_file_transfer.py

@@ -1,4 +1,4 @@
-"""End-to-end file-transfer tests (plain channel, no crypto)."""
+"""Integration tests for file transfer (plain channel, no crypto)."""
 
 import os
 import threading
@@ -8,7 +8,7 @@
 import pytest
 
 
-from test_util import server_ready, wait_until
+from helpers import server_ready, wait_until
 
 from PyFlow.network_api.connect_tcp import (
     TCP_Client_Base,

test/test_forward_extension.py → test/integration/test_forward_extension.py

@@ -246,7 +246,7 @@ def fake_server(**kwargs):
 def test_server_dispatches_forward_send_msg_internally(server, capsys):
     """/forward_send_msg arriving over the wire is routed by handle_command's
     built-in chain (no extension registration involved)."""
-    from test_util import wait_until
+    from helpers import wait_until
 
     server.running = True
     dest = DummySocket()

test/test_forward_msg_attribution.py → test/integration/test_forward_msg_attribution.py

@@ -26,7 +26,7 @@
 
 import pytest
 
-from test_util import server_ready, wait_until
+from helpers import server_ready, wait_until
 
 from PyFlow.network_api.connect_tcp import (
     TCP_Client_Base,