90 lines
3.9 KiB
Python
90 lines
3.9 KiB
Python
import concurrent.futures
|
|
import hashlib
|
|
import http.client
|
|
import pathlib
|
|
import secrets
|
|
import sqlite3
|
|
import struct
|
|
import tempfile
|
|
import threading
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from server import Server, database
|
|
|
|
|
|
class SyncTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.path = str(pathlib.Path(self.temp.name) / "test.sqlite3")
|
|
self.token = secrets.token_urlsafe(32)
|
|
self.server = Server(("127.0.0.1", 0), self.path, hashlib.sha256(self.token.encode()).hexdigest())
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join()
|
|
self.temp.cleanup()
|
|
|
|
def request(self, method, body=None, token=True, **headers):
|
|
c = http.client.HTTPConnection(*self.server.server_address, timeout=5)
|
|
if token:
|
|
headers["Authorization"] = "Bearer " + self.token
|
|
c.request(method, "/v1/vault", body=body, headers=headers)
|
|
r = c.getresponse()
|
|
result = r.status, dict(r.getheaders()), r.read()
|
|
c.close()
|
|
return result
|
|
|
|
def envelope(self):
|
|
return b"QSAFE01!" + struct.pack(">I", 600_000) + secrets.token_bytes(80)
|
|
|
|
def test_requires_credential_and_preconditions(self):
|
|
self.assertEqual(self.request("GET", token=False)[0], 401)
|
|
self.assertEqual(self.request("GET")[0], 404)
|
|
self.assertEqual(self.request("PUT", self.envelope())[0], 412)
|
|
self.assertEqual(self.request("PUT", b"plaintext", **{"If-None-Match": "*"})[0], 413)
|
|
|
|
def test_roundtrip_and_versions(self):
|
|
a, b = self.envelope(), self.envelope()
|
|
status, headers, _ = self.request("PUT", a, **{"If-None-Match": "*"})
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(self.request("GET")[2], a)
|
|
self.assertEqual(self.request("PUT", b, **{"If-Match": '"stale"'})[0], 412)
|
|
self.assertEqual(self.request("PUT", b, **{"If-Match": headers["ETag"]})[0], 200)
|
|
self.assertEqual(self.request("GET")[2], b)
|
|
with database(self.path) as db:
|
|
self.assertEqual(db.execute("SELECT body FROM history").fetchone()[0], a)
|
|
self.assertEqual(db.execute("PRAGMA integrity_check").fetchone()[0], "ok")
|
|
|
|
def test_competing_writes_only_one_wins(self):
|
|
_, headers, _ = self.request("PUT", self.envelope(), **{"If-None-Match": "*"})
|
|
etag = headers["ETag"]
|
|
with concurrent.futures.ThreadPoolExecutor(2) as pool:
|
|
jobs = [pool.submit(self.request, "PUT", self.envelope(), **{"If-Match": etag}) for _ in range(2)]
|
|
self.assertEqual(sorted(j.result()[0] for j in jobs), [200, 412])
|
|
|
|
def test_rejects_invalid_envelope(self):
|
|
body = b"not-a-qsafe-file" + secrets.token_bytes(80)
|
|
self.assertEqual(self.request("PUT", body, **{"If-None-Match": "*"})[0], 400)
|
|
|
|
def test_backup_is_standalone_and_contains_latest_revision(self):
|
|
a, b = self.envelope(), self.envelope()
|
|
_, headers, _ = self.request("PUT", a, **{"If-None-Match": "*"})
|
|
self.assertEqual(self.request("PUT", b, **{"If-Match": headers["ETag"]})[0], 200)
|
|
target = pathlib.Path(self.temp.name) / "backups"
|
|
subprocess.run([sys.executable, str(pathlib.Path(__file__).with_name("backup.py")), self.path, str(target)], check=True)
|
|
backups = list(target.glob("*.sqlite3"))
|
|
self.assertEqual(len(backups), 1)
|
|
self.assertEqual(list(target.glob("*.tmp*")), [])
|
|
with sqlite3.connect(backups[0]) as db:
|
|
self.assertEqual(db.execute("SELECT body FROM vault").fetchone()[0], b)
|
|
self.assertEqual(db.execute("SELECT body FROM history").fetchone()[0], a)
|
|
self.assertEqual(db.execute("PRAGMA integrity_check").fetchone()[0], "ok")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|