(self)
| 110 | self.assertIn("usage: cactus", err) |
| 111 | |
| 112 | def test_serve(self): |
| 113 | cactus = self.find_cactus() |
| 114 | |
| 115 | ret, _, _ = self.run_cli(["create", self.path]) |
| 116 | self.assertEqual(0, ret) |
| 117 | |
| 118 | port = 12345 |
| 119 | |
| 120 | p = subprocess.Popen([cactus, "serve", "-p", str(port)], cwd=self.path, stdout=subprocess.DEVNULL, |
| 121 | stderr=subprocess.DEVNULL) |
| 122 | |
| 123 | srv = "http://127.0.0.1:{0}".format(port) |
| 124 | s = requests.Session() |
| 125 | s.mount(srv, HTTPAdapter(max_retries=Retry(backoff_factor=0.2))) |
| 126 | |
| 127 | r = s.post("{0}/_cactus/shutdown".format(srv)) |
| 128 | r.raise_for_status() |
| 129 | |
| 130 | # We'd love to use p.wait(n) here, but that doesn't work on |
| 131 | # some of the versions of Python we support. |
| 132 | for _ in range(5): |
| 133 | if p.poll() != None: |
| 134 | break |
| 135 | time.sleep(1) |
| 136 | else: |
| 137 | self.fail("Server did not exit!") |
| 138 | |
| 139 | self.assertEqual(0, p.returncode) |
nothing calls this directly
no test coverage detected