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