Merge ~jocave/checkbox-support:add-snap-connect-script into checkbox-support:master

[Jonathan Cave]

Create reusable version of snap_connect.py script that has previously been copied throughout project specific providers.

Tested in plano project.

[Paul Larson (pwlars)]

+1

review: Approve

[Devices Certification Bot (ce-certification-qa)]

The merge was fine but running tests failed.

[trusty] starting container
[trusty] (timing) 0.00user 0.03system 0:30.34elapsed 0%CPU (0avgtext+0avgdata 5020maxresident)k
[trusty] (timing) 0inputs+80outputs (0major+1163minor)pagefaults 0swaps
[trusty] provisioning container
[trusty] (timing) 72.01user 104.30system 8:40.70elapsed 33%CPU (0avgtext+0avgdata 68700maxresident)k
[trusty] (timing) 8inputs+1609704outputs (16major+1279757minor)pagefaults 0swaps
[trusty-testing] Starting tests...
Found a test script: ./requirements/container-tests-checkbox-support
[trusty-testing] container-tests-checkbox-support: PASS
[trusty-testing] (timing) 14.23user 0.10system 0:44.61elapsed 32%CPU (0avgtext+0avgdata 129388maxresident)k
[trusty-testing] (timing) 0inputs+904outputs (0major+127407minor)pagefaults 0swaps
[trusty-testing] Fixing file permissions in source directory
[trusty-testing] Destroying container
Name: trusty-testing
State: STOPPED
[xenial] starting container
[xenial] Unable to start ephemeral container!
[xenial] stdout:
[xenial] stderr: https://paste.ubuntu.com/p/BBnS2RfMXN/
[xenial] NOTE: unable to execute tests, marked as failed
[xenial] Destroying failed container to reclaim resources
Destroyed container xenial-testing
[bionic] starting container
[bionic] (timing) 0.00user 0.01system 0:30.40elapsed 0%CPU (0avgtext+0avgdata 4996maxresident)k
[bionic] (timing) 0inputs+80outputs (0major+1163minor)pagefaults 0swaps
[bionic] provisioning container
[bionic] (timing) 37.15user 16.74system 6:51.66elapsed 13%CPU (0avgtext+0avgdata 89416maxresident)k
[bionic] (timing) 0inputs+2193096outputs (0major+1956724minor)pagefaults 0swaps
[bionic-testing] Starting tests...
Found a test script: ./requirements/container-tests-checkbox-support
[bionic-testing] container-tests-checkbox-support: PASS
[bionic-testing] (timing) 12.57user 0.18system 0:45.42elapsed 28%CPU (0avgtext+0avgdata 130904maxresident)k
[bionic-testing] (timing) 0inputs+2832outputs (0major+134360minor)pagefaults 0swaps
[bionic-testing] Fixing file permissions in source directory
[bionic-testing] Destroying container
Name: bionic-testing
State: STOPPED

review: Needs Fixing

diff --git a/checkbox_support/scripts/snap_connect.py b/checkbox_support/scripts/snap_connect.py
new file mode 100644
index 0000000..7e8fac0
--- /dev/null
+++ b/checkbox_support/scripts/snap_connect.py
@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+# Copyright 2017 Canonical Ltd.
+# All rights reserved.
+#
+# Written by:
+#   Jonathan Cave <[e-mail]>
+#   Maciej Kisielewski <[e-mail]>
+
+"""
+This program connects snap plugs and lists connections between snaps.
+
+The syntax of program parameters is A:B:C, where :
+
+    A - plug to be connected
+    B - target snap (snap the plug will connect to)
+    C - target slot (slot to connect to)
+
+    Note that originating snap is implied. $SNAP_NAME is used.
+
+Example:
+    $ sudo ./snap_connect.py udisks2:udisks2:service
+
+running
+    $ sudo ./snap_connect.py A:B:C
+is equivalent to:
+    $ snap connect $SNAP_NAME:A B:C
+
+    Note that the program needs sudo if asked to connect plugs.
+"""
+
+import argparse
+import json
+import logging
+import requests_unixsocket
+import os
+import sys
+import time
+from collections import namedtuple
+
+
+Connection = namedtuple(
+    'Connection',
+    ['target_snap', 'target_slot', 'plug_snap', 'plug_plug'])
+
+
+class BadRequest(Exception):
+    pass
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        'connections', nargs='+', default=[],
+        metavar='plug:target_snap:target_slot')
+    args = parser.parse_args()
+    api = SnapdAPI()
+
+    for conn in [spec.split(':') for spec in args.connections]:
+        if len(conn) != 3:
+            raise SystemExit("Bad connection description")
+        assert os.environ['SNAP_NAME']
+        snap = os.environ['SNAP_NAME']
+        existing_connections = api.get_connections()
+        new_connection = Connection(
+            target_snap=conn[1], target_slot=conn[2],
+            plug_snap=snap, plug_plug=conn[0])
+        if new_connection not in existing_connections:
+            try:
+                api.connect(new_connection)
+            except BadRequest as exc:
+                logging.warning("Failed to connect %s to %s. %s" % (
+                    conn[0], conn[1], exc))
+
+
+class SnapdAPI():
+    """Based on https://github.com/snapcore/snapd/wiki/REST-API"""
+    SNAPD_BASE_URL = 'http+unix://%2Frun%2Fsnapd.socket'
+
+    def __init__(self):
+        self.session = requests_unixsocket.Session()
+
+    def get_connections(self):
+        data = self._get('/v2/interfaces')
+        connections = []
+        if 'plugs' in data:
+            for plug in data['plugs']:
+                if 'connections' in plug:
+                    for con in plug['connections']:
+                        connections.append(Connection(
+                            con['snap'], con['slot'],
+                            plug['snap'], plug['plug']))
+        return connections
+
+    def connect(self, con):
+        json_data = json.dumps({
+            'action': 'connect',
+            'slots': [{'snap': con.target_snap, 'slot': con.target_slot}],
+            'plugs': [{'snap': con.plug_snap, 'plug': con.plug_plug}]
+        })
+        res = self._post('/v2/interfaces', json_data)
+        ready = False
+        while not ready:
+            # busy wait until snapd reports connection job as finised
+            time.sleep(0.5)
+            con_res = self._get('/v2/changes/{}'.format(res['change']))
+            ready = con_res['ready']
+
+    def _get(self, path):
+        res = self.session.get(self.SNAPD_BASE_URL + path)
+        if not res.ok:
+            logging.error("Got error %i attempting to access %s",
+                          res.status_code, path)
+            sys.exit(1)
+        return res.json()['result']
+
+    def _post(self, path, data=None):
+        res = self.session.post(self.SNAPD_BASE_URL + path, data=data)
+        if not res.ok:
+            full_res = json.loads(res.text)
+            if res.status_code == 400:
+                raise BadRequest(full_res['result']['message'])
+            logging.error("Got error %i attempting to access %s",
+                          res.status_code, path)
+            sys.exit(1)
+        return res.json()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/setup.py b/setup.py
index df2923a..c2517fd 100755
--- a/setup.py
+++ b/setup.py
@@ -88,6 +88,8 @@ setup(
                 "checkbox_support.scripts.snap_configuration:main"),
             ("checkbox-support-nmea_test="
                 "checkbox_support.scripts.nmea_test:main"),
+            ("checkbox-support-snap_connect="
+                "checkbox_support.scripts.snap_connect:main"),
         ],
     },
 )