103 lines
5.1 KiB
Python
103 lines
5.1 KiB
Python
import http.client
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from http.server import ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
import server # noqa: E402
|
|
|
|
|
|
class QuarryTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmp = tempfile.TemporaryDirectory()
|
|
cfgp = Path(cls.tmp.name) / "config.json"
|
|
cfgp.write_text(json.dumps({"name": "Test Board", "data_dir": "./data", "access": "code",
|
|
"view_code_sha256": server.sha("peek"), "agents": []}))
|
|
cls.k = {a: server.add_agent(str(cfgp), a, a.title()) for a in ("alice", "bob", "carol")}
|
|
server.setup(server.load_config(cfgp), cfgp)
|
|
cls.srv = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
|
|
threading.Thread(target=cls.srv.serve_forever, daemon=True).start()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.srv.shutdown(); cls.tmp.cleanup()
|
|
|
|
def req(self, method, path, body=None, agent=None, headers=None):
|
|
h = dict(headers or {})
|
|
if agent:
|
|
h["Authorization"] = "Bearer " + self.k[agent]
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
if data is not None:
|
|
h["Content-Type"] = "application/json"
|
|
c = http.client.HTTPConnection("127.0.0.1", self.srv.server_port, timeout=10)
|
|
c.request(method, path, body=data, headers=h)
|
|
r = c.getresponse()
|
|
raw = r.read()
|
|
try:
|
|
return r.status, json.loads(raw), dict(r.getheaders())
|
|
except ValueError:
|
|
return r.status, raw, dict(r.getheaders())
|
|
|
|
def test_flow_and_permissions(self):
|
|
s, b, _ = self.req("POST", "/v1/tasks", {"title": "Ship it", "detail": "d", "assignee": "bob"}, agent="alice")
|
|
self.assertEqual(s, 201)
|
|
tid = b["task"]["id"]
|
|
self.assertEqual(b["task"]["owner_id"], "alice")
|
|
# assignee may move, not edit
|
|
self.assertEqual(self.req("PATCH", f"/v1/tasks/{tid}", {"status": "doing"}, agent="bob")[0], 200)
|
|
self.assertEqual(self.req("PATCH", f"/v1/tasks/{tid}", {"title": "x"}, agent="bob")[0], 403)
|
|
# others: comment only
|
|
self.assertEqual(self.req("PATCH", f"/v1/tasks/{tid}", {"status": "done"}, agent="carol")[0], 403)
|
|
self.assertEqual(self.req("POST", f"/v1/tasks/{tid}/comments", {"text": "nice"}, agent="carol")[0], 201)
|
|
self.assertEqual(self.req("DELETE", f"/v1/tasks/{tid}", agent="bob")[0], 403)
|
|
# owner edits everything
|
|
s, b, _ = self.req("PATCH", f"/v1/tasks/{tid}", {"title": "Shipped", "status": "done", "assignee": None}, agent="alice")
|
|
self.assertEqual((s, b["task"]["title"], b["task"]["assignee"]), (200, "Shipped", None))
|
|
s, b, _ = self.req("GET", f"/v1/tasks/{tid}", agent="carol")
|
|
self.assertEqual(b["task"]["comments"][0]["author_name"], "Carol")
|
|
self.assertEqual(self.req("DELETE", f"/v1/tasks/{tid}", agent="alice")[0], 200)
|
|
self.assertEqual(self.req("GET", f"/v1/tasks/{tid}", agent="alice")[0], 404)
|
|
|
|
def test_validation(self):
|
|
self.assertEqual(self.req("POST", "/v1/tasks", {"title": ""}, agent="alice")[0], 400)
|
|
self.assertEqual(self.req("POST", "/v1/tasks", {"title": "t", "status": "later"}, agent="alice")[0], 400)
|
|
self.assertEqual(self.req("POST", "/v1/tasks", {"title": "t", "assignee": "nobody"}, agent="alice")[0], 400)
|
|
self.assertEqual(self.req("POST", "/v1/tasks", {"title": "x" * 121}, agent="alice")[0], 400)
|
|
|
|
def test_people_read_only(self):
|
|
self.assertEqual(self.req("GET", "/v1/tasks")[0], 401)
|
|
s, _, h = self.req("GET", "/")
|
|
self.assertEqual((s, h["Location"]), (302, "/view"))
|
|
self.assertEqual(self.req("GET", "/v1/tasks", headers={"X-View-Code": "peek"})[0], 200)
|
|
# a view code never writes
|
|
self.assertEqual(self.req("POST", "/v1/tasks", {"title": "t"}, headers={"X-View-Code": "peek"})[0], 401)
|
|
self.assertEqual(self.req("POST", "/v1/tasks", {"title": "t"}, headers={"Authorization": "Bearer wrong"})[0], 401)
|
|
c = http.client.HTTPConnection("127.0.0.1", self.srv.server_port, timeout=10)
|
|
c.request("POST", "/view", body="code=peek", headers={"Content-Type": "application/x-www-form-urlencoded"})
|
|
r = c.getresponse(); r.read()
|
|
self.assertEqual(r.status, 303)
|
|
cookie = r.getheader("Set-Cookie").split(";")[0]
|
|
s, body, _ = self.req("GET", "/", headers={"Cookie": cookie})
|
|
self.assertEqual(s, 200)
|
|
self.assertIn(b"Test Board", body)
|
|
|
|
def test_keys_stored_hashed(self):
|
|
cfg = json.loads((Path(self.tmp.name) / "config.json").read_text())
|
|
blob = json.dumps(cfg)
|
|
for key in self.k.values():
|
|
self.assertNotIn(key, blob)
|
|
|
|
def test_whoami_and_agents(self):
|
|
self.assertEqual(self.req("GET", "/v1/whoami", agent="bob")[1]["agent"], "bob")
|
|
self.assertEqual([a["id"] for a in self.req("GET", "/v1/agents", agent="bob")[1]["agents"]], ["alice", "bob", "carol"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|