fix(oci): end a pasted definition with a single Ctrl+D

oci/src/proxmenux_oci/custom.py

@@ -3,6 +3,7 @@ is translated into the same template the catalog applications use."""
 from __future__ import annotations
 
 import json
+import os
 import re
 import shlex
 import subprocess
@@ -455,12 +456,24 @@ def describe_oci_compose(template: dict[str, Any]) -> str:
 
 
 def _read_pasted(ui, title: str | None = None, limit: int = MAX_COMPOSE_BYTES, missing: str | None = None) -> str:
-    """The definition is pasted in the terminal: dialog cannot take it."""
+    """The definition is pasted in the terminal: dialog cannot take it. On a
+    terminal one Ctrl+D ends it, also at the end of a line that has text."""
     console.show_logo()
     console.msg_title(title or translate('Compose file of the application'))
-    console.msg_info2(translate('Paste it here and press Ctrl+D on an empty line.'))
+    console.msg_info2(translate('Paste it here and press Ctrl+D.'))
     print()
-    text = sys.stdin.read(limit + 1)
+    if sys.stdin.isatty():
+        data = b''
+        while len(data) <= limit:
+            chunk = os.read(sys.stdin.fileno(), 65536)
+            data += chunk
+            # The terminal hands over whole lines; a chunk without its newline
+            # is the line Ctrl+D sent before the end of the line.
+            if not chunk.endswith(b'\n'):
+                break
+        text = data.decode('utf-8', errors='replace')
+    else:
+        text = sys.stdin.read(limit + 1)
     if not text.strip():
         raise UserCancelled(missing or (translate('No docker run command was given') if title is not None
                                         else translate('No Compose file was given')))

oci/tests/test_paste_until_ctrl_d.py

@@ -0,0 +1,55 @@
+"""A pasted definition ends with one Ctrl+D, also when it is pressed at the
+end of the last line instead of on an empty one."""
+
+import io
+import os
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from proxmenux_oci import custom
+
+
+def read_pasted(limit):
+    with patch.object(custom.console, "show_logo"), patch.object(custom.console, "msg_title"), \
+            patch.object(custom.console, "msg_info2"), patch("builtins.print"):
+        return custom._read_pasted(None, limit=limit)
+
+COMPOSE = "services:\n  app:\n    image: nginx:latest\n"
+
+
+@unittest.skipUnless(hasattr(os, "openpty"), "needs a pseudo terminal")
+class PasteUntilCtrlDTests(unittest.TestCase):
+    def read(self, typed: bytes, limit: int = 1024) -> str:
+        controller, terminal = os.openpty()
+        with os.fdopen(terminal, "r") as stdin, patch("sys.stdin", stdin):
+            os.write(controller, typed)
+            try:
+                return read_pasted(limit)
+            finally:
+                os.close(controller)
+
+    def test_ctrl_d_on_an_empty_line(self):
+        self.assertEqual(self.read(COMPOSE.encode() + b"\x04"), COMPOSE)
+
+    def test_ctrl_d_at_the_end_of_the_last_line(self):
+        self.assertEqual(self.read(COMPOSE.rstrip("\n").encode() + b"\x04"), COMPOSE.rstrip("\n"))
+
+    def test_reading_stops_past_the_limit(self):
+        text = self.read(b"a: 1\n" * 20 + b"\x04", limit=10)
+        self.assertGreater(len(text), 10)
+        self.assertLess(len(text), 100)
+
+
+class PastedFromAPipeTests(unittest.TestCase):
+    def test_a_pipe_is_read_to_its_end(self):
+        with patch("sys.stdin", io.StringIO(COMPOSE)):
+            self.assertEqual(read_pasted(1024), COMPOSE)
+
+
+if __name__ == "__main__":
+    unittest.main()