python: ensure basic behaviour of all channel types

Make sure every channel type sends either "ready" (if expecting data) or
"close" (if it sent data and is done) on open.

Make sure every channel type also sends a "close" message on close.

src/cockpit/channel.py

@@ -262,6 +262,7 @@ def create_transport_done(self, task):
 
         self.connection_made(transport)
         self.thaw_endpoint()
+        self.ready()
 
     def connection_made(self, transport: asyncio.BaseTransport):
         assert isinstance(transport, asyncio.Transport)

src/cockpit/channels/dbus.py

@@ -491,3 +491,6 @@ def do_data(self, data):
 
         self.tasks.add(task)
         task.add_done_callback(self.tasks.discard)
+
+    def do_close(self):
+        self.close()

src/cockpit/channels/filesystem.py

@@ -145,6 +145,7 @@ def unlink_temppath(self):
     def do_open(self, options):
         self._path = options.get('path')
         self._tag = options.get('tag')
+        self.ready()
 
     def do_data(self, data):
         if self._tempfile is None:

src/cockpit/channels/http.py

@@ -154,6 +154,8 @@ def do_open(self, options):
         self.options = options
         self.body = b''
 
+        self.ready()
+
     def do_data(self, data):
         self.body += data
 

src/cockpit/channels/packages.py

@@ -66,6 +66,7 @@ def do_done(self):
 
         self.router.packages.serve_file(path, self)
         self.done()
+        self.close()
 
     def do_data(self, data):
         self.post += data

src/cockpit/channels/trivial.py

@@ -41,3 +41,6 @@ class NullChannel(Channel):
 
     def do_open(self, options):
         self.ready()
+
+    def do_close(self):
+        self.close()

test/pytest/test_bridge.py

@@ -2,6 +2,7 @@
 import asyncio
 import json
 import os
+import pytest
 import unittest
 import unittest.mock
 import sys
@@ -12,6 +13,7 @@
 
 import systemd_ctypes
 from cockpit.bridge import Bridge
+from cockpit.channels import CHANNEL_TYPES
 
 MOCK_HOSTNAME = 'mockbox'
 
@@ -313,6 +315,7 @@ async def verify_root_bridge_running(self):
 
         # close up
         self.transport.send_close(channel=root_dbus)
+        await self.transport.assert_msg('', command='close', channel=root_dbus)
 
     async def test_superuser_dbus(self):
         await self.start()
@@ -553,3 +556,108 @@ async def test_fslist1_notexist(self):
             'fslist1', path='/nonexisting', watch=False,
             problem='not-found',
             reply_keys={'message': "[Errno 2] No such file or directory: '/nonexisting'"})
+
+
+@pytest.mark.parametrize('channeltype', CHANNEL_TYPES)
+def test_channel(channeltype, tmp_path):
+    async def run() -> None:
+        bridge = Bridge(argparse.Namespace(privileged=False))
+        transport = MockTransport(bridge)
+        await transport.assert_msg('', command='init')
+        transport.send_init()
+
+        payload = channeltype.payload
+        args = dict(channeltype.restrictions)
+
+        async def serve_page(reader, writer):
+            while True:
+                line = await reader.readline()
+                if line.strip():
+                    print('HTTP Request line', line)
+                else:
+                    break
+
+            print('Sending HTTP reply')
+            writer.write(b'HTTP/1.1 200 OK\r\n\r\nA document\r\n')
+            await writer.drain()
+            writer.close()
+
+        srv = str(tmp_path / 'sock')
+        await asyncio.start_unix_server(serve_page, srv)
+
+        if payload == 'fslist1':
+            args = {'path': '/', 'watch': False}
+        elif payload == 'fsread1':
+            args = {'path': '/etc/passwd'}
+        elif payload == 'fsreplace1':
+            args = {'path': 'tmpfile'}
+        elif payload == 'fswatch1':
+            args = {'path': '/etc'}
+        elif payload == 'http-stream1':
+            args = {'internal': 'packages', 'method': 'GET', 'path': '/manifests.js',
+                    'headers': {'X-Forwarded-Proto': 'http', 'X-Forwarded-Host': 'localhost'}}
+        elif payload == 'http-stream2':
+            args = {'method': 'GET', 'path': '/bzzt', 'unix': srv}
+        elif payload == 'stream':
+            if 'spawn' in args:
+                args = {'spawn': ['cat']}
+            else:
+                args = {'unix': srv}
+        elif payload == 'metrics1':
+            args['metrics'] = [{'name': 'memory.free'}]
+        elif payload == 'dbus-json3':
+            if not os.path.exists('/run/dbus/system_bus_socket'):
+                pytest.skip('no dbus')
+        else:
+            args = {}
+
+        print('sending open', payload, args)
+        ch = transport.send_open(payload, **args)
+        saw_data = False
+
+        while True:
+            channel, msg = await transport.next_frame()
+            print(channel, msg)
+            if channel == '':
+                control = json.loads(msg)
+                assert control['channel'] == ch
+                command = control['command']
+                if command == 'done':
+                    saw_data = True
+                elif command == 'ready':
+                    # If we get ready, it's our turn to send data first.
+                    # Hopefully we didn't receive any before.
+                    assert isinstance(bridge.open_channels[ch], channeltype)
+                    assert not saw_data
+                    break
+                elif command == 'close':
+                    # If we got an immediate close message then it should be
+                    # because the channel sent data and finished, without error.
+                    assert 'problem' not in control
+                    assert saw_data
+                    return
+                else:
+                    assert False, (payload, args, control)
+            else:
+                saw_data = True
+
+        # If we're here, it's our turn to talk.  Say nothing.
+        print('sending done')
+        transport.send_done(ch)
+
+        if payload in ['dbus-json3', 'fswatch1', 'null']:
+            transport.send_close(ch)
+
+        while True:
+            channel, msg = await transport.next_frame()
+            print(channel, msg)
+            if channel == '':
+                control = json.loads(msg)
+                command = control['command']
+                if command == 'done':
+                    continue
+                elif command == 'close':
+                    assert 'problem' not in control
+                    return
+
+    asyncio.run(run())