create the live files and test that they exist
# See the License for the specific language governing permissions and
# limitations under the License.
+import lzma
+import multiprocessing
+import os
+import pathlib
+import socket
+import subprocess
+import tempfile
+
+from pymedphys._data import download
+
def test_icom_cli():
- assert False
+ icom_server_process = multiprocessing.Process(target=_mock_icom_server)
+ icom_server_process.start()
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ icom_listen_cli = subprocess.Popen(
+ ["pymedphys", "icom", "listen", "127.0.0.1", temp_dir], env=os.environ
+ )
+ icom_server_process.join()
+ icom_listen_cli.terminate()
+
+ live_files = list(pathlib.Path(temp_dir).glob("**/*.txt"))
+ assert len(live_files) == 256
+
+
+def _mock_icom_server():
+ paths = download.zip_data_paths("metersetmap-gui-e2e-data.zip")
+ icom_paths = [path for path in paths if path.suffix == ".xz"]
+
+ def load_icom_stream(icom_path):
+ with lzma.open(icom_path, "r") as f:
+ contents = f.read()
+
+ return contents
+
+ data = load_icom_stream(icom_paths[0])
+
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ s.bind(("127.0.0.1", 1706))
+ s.listen()
+ conn, _addr = s.accept()
+ with conn:
+ conn.sendall(data)
