107 lines
5.7 KiB
Python
107 lines
5.7 KiB
Python
import json
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import unittest
|
|
from http.client import HTTPConnection
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from app.main import create_server, hash_password
|
|
|
|
|
|
class ApiSmokeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = TemporaryDirectory()
|
|
self.old_env = {key: os.environ.get(key) for key in ("BOOTSTRAP_ADMIN_EMAIL", "BOOTSTRAP_ADMIN_PASSWORD")}
|
|
os.environ["BOOTSTRAP_ADMIN_EMAIL"] = "owner@example.test"
|
|
os.environ["BOOTSTRAP_ADMIN_PASSWORD"] = "development-password"
|
|
self.db_path = self.tmp.name + "/test.db"
|
|
self.server = create_server("127.0.0.1", 0, self.db_path)
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
self.conn = HTTPConnection("127.0.0.1", self.server.server_port, timeout=3)
|
|
self.cookie = None
|
|
status, _ = self.request("POST", "/api/v1/auth/login", {"email": "owner@example.test", "password": "development-password"})
|
|
self.assertEqual(status, 200)
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join(timeout=2)
|
|
for key, value in self.old_env.items():
|
|
if value is None: os.environ.pop(key, None)
|
|
else: os.environ[key] = value
|
|
self.tmp.cleanup()
|
|
|
|
def request(self, method, path, payload=None, cookie=True):
|
|
body = json.dumps(payload).encode() if payload is not None else None
|
|
headers = {"Content-Type": "application/json"} if body else {}
|
|
if cookie and self.cookie: headers["Cookie"] = self.cookie
|
|
self.conn.request(method, path, body, headers)
|
|
response = self.conn.getresponse()
|
|
set_cookie = response.getheader("Set-Cookie")
|
|
if set_cookie and "session=" in set_cookie:
|
|
self.cookie = set_cookie.split(";", 1)[0]
|
|
return response.status, json.loads(response.read() or b"{}")
|
|
|
|
def test_auth_login_me_logout_and_protected_route(self):
|
|
self.assertEqual(self.request("GET", "/api/v1/auth/me")[0], 200)
|
|
self.assertEqual(self.request("POST", "/api/v1/auth/logout")[0], 200)
|
|
self.assertEqual(self.request("GET", "/api/v1/dashboard/summary")[0], 401)
|
|
|
|
def test_viewer_cannot_mutate(self):
|
|
password_hash, salt = hash_password("viewer-password")
|
|
db = sqlite3.connect(self.db_path)
|
|
db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ("demo-tenant", "viewer@example.test", password_hash, salt, "viewer"))
|
|
db.commit(); db.close()
|
|
self.cookie = None
|
|
self.assertEqual(self.request("POST", "/api/v1/auth/login", {"email": "viewer@example.test", "password": "viewer-password"})[0], 200)
|
|
self.assertEqual(self.request("POST", "/api/v1/businesses", {"name": "Nope"})[0], 403)
|
|
|
|
def test_cross_organization_businesses_are_isolated(self):
|
|
password_hash, salt = hash_password("other-password")
|
|
db = sqlite3.connect(self.db_path)
|
|
db.execute("INSERT INTO organizations (id,name) VALUES (?,?)", ("other-tenant", "Other"))
|
|
db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ("other-tenant", "other@example.test", password_hash, salt, "owner"))
|
|
db.commit(); db.close()
|
|
self.cookie = None
|
|
self.assertEqual(self.request("POST", "/api/v1/auth/login", {"email": "other@example.test", "password": "other-password"})[0], 200)
|
|
self.assertEqual(self.request("POST", "/api/v1/businesses", {"name": "Other Co", "website": "https://other.test"})[0], 201)
|
|
self.assertEqual(self.request("GET", "/api/v1/businesses")[1]["organization_id"], "other-tenant")
|
|
self.cookie = None
|
|
self.assertEqual(self.request("POST", "/api/v1/auth/login", {"email": "owner@example.test", "password": "development-password"})[0], 200)
|
|
status, businesses = self.request("GET", "/api/v1/businesses")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(businesses["items"], [])
|
|
|
|
def test_missing_auth_is_rejected(self):
|
|
self.cookie = None
|
|
self.assertEqual(self.request("GET", "/api/v1/businesses", cookie=False)[0], 401)
|
|
|
|
def test_health_create_get_summary_and_import_preview(self):
|
|
self.assertEqual(self.request("GET", "/api/v1/health/live")[0], 200)
|
|
status, created = self.request("POST", "/api/v1/businesses", {"name": "Acme", "website": "https://acme.co.za", "email": "a@acme.co.za"})
|
|
self.assertEqual(status, 201)
|
|
self.assertEqual(created["organization_id"], "demo-tenant")
|
|
status, fetched = self.request("GET", "/api/v1/businesses/" + str(created["id"]))
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(fetched["name"], "Acme")
|
|
status, summary = self.request("GET", "/api/v1/dashboard/summary")
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(summary["businesses"], 1)
|
|
status, preview = self.request("POST", "/api/v1/imports/preview", {"rows": [{"name": "Acme", "website": "https://acme.co.za"}, {"name": "New Co"}]})
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(preview["accepted"], 1)
|
|
self.assertEqual(preview["duplicates"], 1)
|
|
|
|
def test_suppression_blocks_new_business(self):
|
|
status, _ = self.request("POST", "/api/v1/suppressions", {"kind": "domain", "value": "blocked.co.za"})
|
|
self.assertEqual(status, 201)
|
|
status, response = self.request("POST", "/api/v1/businesses", {"name": "Blocked", "website": "https://blocked.co.za"})
|
|
self.assertEqual(status, 409)
|
|
self.assertEqual(response["error"], "suppressed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|