1import requests_unixsocket
2import urllib
3import os
4import os.path
5
6from lib import BaseTest
7
8
9class SystemdAPIHandoverTest(BaseTest):
10    aptly_server = None
11    socket_path = "/tmp/_aptly_systemdapihandovertest.sock"
12
13    def prepare(self):
14        # On Debian they use /lib on other systems /usr/lib.
15        systemd_activate = "/usr/lib/systemd/systemd-activate"
16        if not os.path.exists(systemd_activate):
17            systemd_activate = "/lib/systemd/systemd-activate"
18        if not os.path.exists(systemd_activate):
19            print("Could not find systemd-activate")
20            return
21        self.aptly_server = self._start_process("%s -l %s aptly api serve -no-lock" %
22                                                (systemd_activate, self.socket_path),)
23        super(SystemdAPIHandoverTest, self).prepare()
24
25    def shutdown(self):
26        if self.aptly_server is not None:
27            self.aptly_server.terminate()
28            self.aptly_server.wait()
29            self.aptly_server = None
30        if os.path.exists(self.socket_path):
31            os.remove(self.socket_path)
32        super(SystemdAPIHandoverTest, self).shutdown()
33
34    def run(self):
35        pass
36
37    """
38    Verify we can listen on a unix domain socket.
39    """
40    def check(self):
41        if self.aptly_server is None:
42            print("Skipping test as we failed to setup a listener.")
43            return
44        session = requests_unixsocket.Session()
45        r = session.get('http+unix://%s/api/version' % urllib.quote(self.socket_path, safe=''))
46        self.check_equal(r.json(), {'Version': os.environ['APTLY_VERSION']})
47