53 lines
3.2 KiB
Python
53 lines
3.2 KiB
Python
|
import sqlite3
|
||
|
|
try:
|
||
|
|
from test_api import ApiSmokeTests
|
||
|
|
except ModuleNotFoundError:
|
||
|
|
from tests.test_api import ApiSmokeTests
|
||
|
|
|
||
|
|
|
||
|
|
class OperatorWorkflowTests(ApiSmokeTests):
|
||
|
|
def test_saved_filter_round_trip_update_delete_and_bounds(self):
|
||
|
|
status, saved = self.request("POST", "/api/v1/saved-filters", {"name": "Hot", "filters": {"score_min": 70, "website_state": "business_site"}})
|
||
|
|
self.assertEqual(status, 201)
|
||
|
|
self.assertEqual(saved["filters"]["score_min"], 70)
|
||
|
|
fid = saved["id"]
|
||
|
|
status, listing = self.request("GET", "/api/v1/saved-filters")
|
||
|
|
self.assertEqual(status, 200)
|
||
|
|
self.assertEqual(listing["items"][0]["name"], "Hot")
|
||
|
|
self.assertEqual(self.request("PATCH", f"/api/v1/saved-filters/{fid}", {"filters": {"priority": "high"}})[0], 200)
|
||
|
|
self.assertEqual(self.request("DELETE", f"/api/v1/saved-filters/{fid}")[0], 200)
|
||
|
|
self.assertEqual(self.request("POST", "/api/v1/saved-filters", {"name": "Bad", "filters": {"x": ["a"] * 51}})[0], 400)
|
||
|
|
|
||
|
|
def test_review_queue_filters_flags_suppression_and_bulk_is_atomic(self):
|
||
|
|
_, first = self.request("POST", "/api/v1/businesses", {"name": "Queue One", "website": "https://queue-one.test", "email": "one@queue.test"})
|
||
|
|
_, second = self.request("POST", "/api/v1/businesses", {"name": "Queue Two", "website": "https://queue-two.test"})
|
||
|
|
self.assertEqual(self.request("POST", "/api/v1/suppressions", {"kind": "domain", "value": "queue-one.test"})[0], 201)
|
||
|
|
status, queue = self.request("GET", "/api/v1/review-queue?review_status=pending&website_state=business_site&page_size=10")
|
||
|
|
self.assertEqual(status, 200)
|
||
|
|
flagged = next(x for x in queue["items"] if x["id"] == first["id"])
|
||
|
|
self.assertTrue(flagged["suppressed"])
|
||
|
|
self.assertFalse(flagged["outreach_eligible"])
|
||
|
|
self.assertEqual(self.request("POST", "/api/v1/businesses/bulk-review", {"ids": [first["id"], second["id"]], "action": "verify"})[0], 200)
|
||
|
|
status, queue = self.request("GET", "/api/v1/review-queue?review_status=verified&page_size=10")
|
||
|
|
self.assertEqual(status, 200)
|
||
|
|
self.assertEqual({x["id"] for x in queue["items"]}, {first["id"], second["id"]})
|
||
|
|
self.assertFalse(next(x for x in queue["items"] if x["id"] == first["id"])["outreach_eligible"])
|
||
|
|
self.assertEqual(self.request("POST", "/api/v1/businesses/bulk-review", {"ids": [first["id"]] * 101, "action": "reject"})[0], 400)
|
||
|
|
db = sqlite3.connect(self.db_path)
|
||
|
|
self.assertEqual(db.execute("SELECT COUNT(*) FROM audit_log WHERE action='businesses.bulk_review'").fetchone()[0], 1)
|
||
|
|
db.close()
|
||
|
|
|
||
|
|
def test_dashboard_operator_counts_and_clickable_metadata(self):
|
||
|
|
self.request("POST", "/api/v1/businesses", {"name": "Dashboard New", "website": "https://dash.test"})
|
||
|
|
status, summary = self.request("GET", "/api/v1/dashboard/summary")
|
||
|
|
self.assertEqual(status, 200)
|
||
|
|
self.assertEqual(summary["counts"]["new"], 1)
|
||
|
|
for key in ("new", "hot", "review", "source_health", "active_jobs"):
|
||
|
|
self.assertIn(key, summary["clickable_filters"])
|
||
|
|
self.assertEqual(summary["clickable_filters"][key]["count"], summary["counts"][key])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
import unittest
|
||
|
|
unittest.main()
|