591 lines
42 KiB
Python
591 lines
42 KiB
Python
import json
|
||
import os
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import unittest
|
||
from http.client import HTTPConnection
|
||
from unittest.mock import patch
|
||
|
||
from api import server
|
||
from api import browser_investigator
|
||
|
||
|
||
class BrowserOnlyAppTests(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.NamedTemporaryFile(delete=False)
|
||
self.tmp.close()
|
||
server.DB_PATH = self.tmp.name
|
||
self.auth_required = server.AUTH_REQUIRED
|
||
server.AUTH_REQUIRED = False
|
||
self.scan_patcher = patch("api.server.scan_website", return_value={"website_status": "healthy", "website_http_status": 200, "website_final_url": "", "website_latency_ms": 50, "website_https": 1, "website_title": "", "website_technology": "", "website_has_viewport": 1, "website_tls_expires_at": "", "website_checked_at": server.now(), "website_error": "", "website_emails": [], "email_evidence": []})
|
||
self.scan_patcher.start()
|
||
self.http = server.ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
|
||
self.thread = threading.Thread(target=self.http.serve_forever, daemon=True)
|
||
self.thread.start()
|
||
self.conn = HTTPConnection("127.0.0.1", self.http.server_port)
|
||
|
||
def tearDown(self):
|
||
server.AUTH_REQUIRED = self.auth_required
|
||
self.scan_patcher.stop()
|
||
self.http.shutdown(); self.http.server_close(); self.thread.join(1)
|
||
try: os.unlink(self.tmp.name)
|
||
except FileNotFoundError: pass
|
||
|
||
def request(self, method, path, payload=None):
|
||
body = json.dumps(payload).encode() if payload is not None else None
|
||
self.conn.request(method, path, body, {"Content-Type": "application/json"} if body else {})
|
||
response = self.conn.getresponse()
|
||
return response.status, json.loads(response.read() or b"{}")
|
||
|
||
def wait_for_run(self, run_id):
|
||
for _ in range(300):
|
||
status, body = self.request("GET", f"/api/runs/{run_id}")
|
||
self.assertEqual(status, 200)
|
||
if body["status"] not in {"queued", "running"}:
|
||
return body
|
||
time.sleep(0.02)
|
||
self.fail("run did not finish")
|
||
|
||
def test_search_redirect_links_are_unwrapped(self):
|
||
wrapped = "https://duckduckgo.com/l/?uddg=https%3A%2F%2Fwww.example.co.za%2Fservices"
|
||
self.assertEqual(server.normalize_result_href(wrapped), "https://www.example.co.za/services")
|
||
self.assertEqual(server.normalize_result_href("//example.co.za/about"), "https://example.co.za/about")
|
||
|
||
@patch.dict(os.environ, {"DISCOVERY_BACKEND": "browser"})
|
||
@patch("api.server.google_search", return_value=[{"name": "Acme Plumbing", "website": "https://www.acme.co.za", "domain": "acme.co.za"}, {"name": "Other", "website": "https://other.co.za", "domain": "other.co.za"}])
|
||
def test_discovery_filters_catalogued_root_domains(self, search):
|
||
status, first = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "Cape Town", "max_results": 5})
|
||
self.assertEqual(status, 202)
|
||
first = self.wait_for_run(first["id"])
|
||
self.assertEqual(first["added"], 2)
|
||
saved = self.request("GET", "/api/prospects")[1]["items"]
|
||
self.assertTrue(all(item["search_category"] == "plumbers" for item in saved))
|
||
status, second = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "Cape Town", "max_results": 5})
|
||
self.assertEqual(status, 202)
|
||
second = self.wait_for_run(second["id"])
|
||
self.assertEqual(second["found"], 2); self.assertEqual(second["added"], 0)
|
||
self.assertEqual(self.request("GET", "/api/prospects")[1]["total"], 2)
|
||
|
||
@patch.dict(os.environ, {"DISCOVERY_BACKEND": "browser"})
|
||
def test_map_without_website_gets_stable_catalogue_key(self):
|
||
mapped = server.map_gmaps_result({"title": "A Local Plumber", "place_id": "abc123", "address": "Cape Town", "phone": "+27 10", "review_rating": 4.5, "review_count": 12}, "plumbers Cape Town")
|
||
self.assertEqual(mapped["domain"], "maps:abc123")
|
||
self.assertEqual(mapped["website_status"], "no_detected_website")
|
||
self.assertGreaterEqual(mapped["opportunity_score"], 70)
|
||
|
||
@patch.dict(os.environ, {"DISCOVERY_BACKEND": "browser"})
|
||
def test_review_updates_only_selected_prospect(self):
|
||
with patch("api.server.google_search", return_value=[{"name": "Acme", "website": "https://acme.co.za", "domain": "acme.co.za"}]):
|
||
_, run = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "South Africa"})
|
||
self.wait_for_run(run["id"])
|
||
status, item = self.request("PATCH", "/api/prospects/1/review", {"status": "approved", "notes": "Looks relevant"})
|
||
self.assertEqual(status, 200); self.assertEqual(item["review_status"], "approved")
|
||
|
||
def test_standard_maps_worker_uses_jobs_api_without_key(self):
|
||
csv_body = b"title,address,website,phone,latitude,longitude,review_rating,review_count,link\nAcme Plumbing,Cape Town,https://acme.co.za,+2710, -33.9,18.4,4.5,12,https://maps.google.com/?cid=1\n"
|
||
with patch.object(server, "GMAPS_URL", "http://gmaps:8080"), \
|
||
patch.object(server, "GMAPS_API_MODE", "standard"), \
|
||
patch.object(server, "GMAPS_API_KEY", ""), \
|
||
patch("api.server.http_json", side_effect=[{"id": "job-1"}, {"ID": "job-1", "Status": "pending"}, {"ID": "job-1", "Status": "working"}, {"ID": "job-1", "Status": "ok"}]) as http_json, \
|
||
patch("api.server.http_bytes", return_value=csv_body) as http_bytes:
|
||
job_id = server.submit_gmaps_job("plumbers Cape Town")
|
||
results = server.poll_gmaps_job(job_id)
|
||
self.assertEqual(job_id, "job-1")
|
||
self.assertEqual(results[0]["title"], "Acme Plumbing")
|
||
first_call = http_json.call_args_list[0]
|
||
self.assertEqual(first_call.args[0:2], ("POST", "http://gmaps:8080/api/v1/jobs"))
|
||
self.assertNotIn("X-API-Key", first_call.args[3])
|
||
self.assertTrue(first_call.args[2]["email"])
|
||
self.assertEqual(http_bytes.call_args.args[1], "http://gmaps:8080/api/v1/jobs/job-1/download")
|
||
|
||
def test_degoog_search_normalises_candidates_and_keeps_provenance(self):
|
||
class Response:
|
||
def __enter__(self): return self
|
||
def __exit__(self, *args): return None
|
||
def read(self, _limit=None):
|
||
return json.dumps({"results": [{"url": "https://www.acme.co.za/", "title": "Acme Plumbing", "content": "Acme Plumbing Cape Town"}]}).encode()
|
||
prospect = {"name": "Acme Plumbing", "address": "Cape Town", "category": "Plumbers", "phone": ""}
|
||
with patch.dict(os.environ, {"DEGOOG_ENABLED": "true", "DEGOOG_URL": "http://degoog:4444"}, clear=False), patch("api.browser_investigator.urllib.request.urlopen", return_value=Response()):
|
||
results = browser_investigator._degoog_search_candidates(prospect, ['"Acme Plumbing" Cape Town'], requested=3, timeout_seconds=10)
|
||
self.assertEqual(results[0]["source"], "degoog")
|
||
self.assertEqual(results[0]["domain"], "acme.co.za")
|
||
|
||
free_models = [{"id": "meituan/longcat-2.0:free", "label": "Meituan · LongCat 2.0", "free": True}]
|
||
with patch.object(server, "nous_models", return_value=free_models):
|
||
status, body = self.request("GET", "/api/ai/models?provider=nous")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["models"], free_models)
|
||
status, body = self.request("POST", "/api/settings", {"ai": {"AI_PROVIDER": "nous", "AI_MODEL": "meituan/longcat-2.0:free", "AI_API_KEY": "nous-test-key"}})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["ai"]["provider"], "nous")
|
||
status, body = self.request("GET", "/api/settings")
|
||
self.assertEqual(status, 200)
|
||
self.assertTrue(body["ai"]["configured"])
|
||
status, body = self.request("GET", "/api/config")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["ai_provider"], "nous")
|
||
self.assertNotIn("nous-test-key", json.dumps(body))
|
||
|
||
with patch("api.server.http_json", return_value={"choices": [{"message": {"content": "OK"}}]}) as http_json:
|
||
status, body = self.request("POST", "/api/ai/test", {})
|
||
self.assertEqual(status, 200)
|
||
self.assertTrue(body["ok"])
|
||
self.assertEqual(http_json.call_args.args[0:2], ("POST", "https://inference-api.nousresearch.com/v1/chat/completions"))
|
||
|
||
def test_deepseek_provider_uses_fixed_endpoint_and_default_models(self):
|
||
status, body = self.request("GET", "/api/ai/models?provider=deepseek")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["base_url"], "https://api.deepseek.com/v1")
|
||
self.assertEqual([item["id"] for item in body["models"]], ["deepseek-chat", "deepseek-reasoner"])
|
||
status, body = self.request("POST", "/api/settings", {"ai": {"AI_PROVIDER": "deepseek", "AI_API_KEY": "deepseek-test-key"}})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["ai"]["base_url"], "https://api.deepseek.com/v1")
|
||
self.assertEqual(body["ai"]["model"], "deepseek-chat")
|
||
self.assertTrue(body["ai"]["configured"])
|
||
status, body = self.request("GET", "/api/config")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["ai_provider"], "deepseek")
|
||
self.assertNotIn("deepseek-test-key", json.dumps(body))
|
||
with patch("api.server.http_json", return_value={"choices": [{"message": {"content": "OK"}}]}) as http_json:
|
||
status, body = self.request("POST", "/api/ai/test", {})
|
||
self.assertEqual(status, 200)
|
||
self.assertTrue(body["ok"])
|
||
self.assertEqual(http_json.call_args.args[0:2], ("POST", "https://api.deepseek.com/v1/chat/completions"))
|
||
self.assertEqual(http_json.call_args.args[2]["thinking"], {"type": "disabled"})
|
||
|
||
def test_provider_errors_are_not_mislabelled_as_maps_errors(self):
|
||
class CloudflareResponse:
|
||
def read(self):
|
||
return b'{"error_code":1010,"detail":"blocked"}'
|
||
|
||
def close(self):
|
||
return None
|
||
|
||
code = 403
|
||
|
||
with patch("api.server.urllib.request.urlopen", side_effect=urllib.error.HTTPError("https://inference-api.nousresearch.com/v1/chat/completions", 403, "Forbidden", {}, CloudflareResponse())):
|
||
with self.assertRaisesRegex(RuntimeError, r"^ai_cloudflare_1010:"):
|
||
server.http_json("POST", "https://inference-api.nousresearch.com/v1/chat/completions", {}, error_prefix="ai")
|
||
|
||
def test_provider_requests_send_explicit_user_agent(self):
|
||
response = type("Response", (), {"__enter__": lambda self: self, "__exit__": lambda self, *args: None, "read": lambda self: b"{}"})()
|
||
with patch("api.server.urllib.request.urlopen", return_value=response) as open_url:
|
||
server.http_json("GET", "https://example.test/v1/models", error_prefix="ai")
|
||
self.assertEqual(open_url.call_args.args[0].headers.get("User-agent"), server.OUTBOUND_USER_AGENT)
|
||
|
||
def test_ai_assessment_is_saved_for_a_prospect(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,website_status,opportunity_score,discovered_at) VALUES(?,?,?,?,?,?)", ("Acme Plumbing", "acme.co.za", "plumber", "no_detected_website", 82, server.now()))
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", ("AI_PROVIDER", "nous", server.now()))
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", ("AI_MODEL", "meituan/longcat-2.0:free", server.now()))
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", ("AI_API_KEY", "test-key", server.now()))
|
||
conn.commit(); conn.close()
|
||
response = {"choices": [{"message": {"content": '{"probability":84,"recommendation":"contact","reasoning":"Active business with a clear website opportunity."}'}}]}
|
||
with patch("api.server.http_json", return_value=response):
|
||
status, body = self.request("POST", "/api/prospects/1/assess", {})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["ai_probability"], 84)
|
||
self.assertEqual(body["ai_recommendation"], "contact")
|
||
saved = server.db().execute("SELECT ai_probability,ai_recommendation,ai_reasoning FROM prospects WHERE id=1").fetchone()
|
||
self.assertEqual(tuple(saved), (84.0, "contact", "Active business with a clear website opportunity."))
|
||
|
||
def test_live_site_snapshot_captures_design_signals_for_generation(self):
|
||
class Response:
|
||
headers = {"Content-Type": "text/html; charset=utf-8"}
|
||
def __enter__(self): return self
|
||
def __exit__(self, *args): return None
|
||
def read(self, limit): return b'''<html class="brand-site"><head><title>Acme</title><style>:root{--brand:#123456} body{font-family: Georgia}</style></head><body><nav><a href="/services">Services</a></nav><main><h1>Acme Plumbing</h1><section class="hero dark"><img src="https://example.com/hero.jpg"><h2>Fast help</h2></section></main></body></html>'''
|
||
def geturl(self): return "https://acme.example/"
|
||
with patch("api.server._safe_url", return_value=True), patch("api.server.urllib.request.urlopen", return_value=Response()):
|
||
snapshot = server.fetch_company_page("https://acme.example/")
|
||
self.assertIn("hero", snapshot["website_design_brief"])
|
||
self.assertIn("#123456", snapshot["website_design_brief"])
|
||
self.assertIn("Georgia", snapshot["website_design_brief"])
|
||
self.assertIn("https://example.com/hero.jpg", snapshot["website_design_brief"])
|
||
|
||
def test_rendered_design_variants_are_not_the_same_template(self):
|
||
prospect = {"name": "Acme Plumbing", "category": "plumbers", "address": "Cape Town"}
|
||
first = server.premium_render_pages(prospect, {}, mode="onepage", design={"variant": "midnight", "layout": "editorial", "radius": "sharp"})["home"]
|
||
second = server.premium_render_pages(prospect, {}, mode="onepage", design={"variant": "warm", "layout": "bento", "radius": "soft"})["home"]
|
||
self.assertNotEqual(first, second)
|
||
self.assertIn("layout-editorial", first)
|
||
self.assertIn("layout-bento", second)
|
||
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,address,phone,emails,website,website_status,discovered_at) VALUES(?,?,?,?,?,?,?,?,?)", ("Acme Plumbing", "maps:acme", "plumbers", "Cape Town", "+27101234567", "info@acme.co.za", "", "no_detected_website", server.now()))
|
||
conn.commit(); conn.close()
|
||
status, created = self.request("POST", "/api/prospects/1/mockup", {"instructions": "Keep it friendly"})
|
||
self.assertEqual(status, 201)
|
||
self.assertIn("/mockups/", created["share_url"])
|
||
self.assertEqual(created["generation_source"], "fallback")
|
||
status, listed = self.request("GET", "/api/prospects/1/mockups")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(listed["items"][0]["id"], created["id"])
|
||
preview_path = "/mockups/" + created["token"]
|
||
self.conn.request("GET", preview_path)
|
||
response = self.conn.getresponse()
|
||
body = response.read().decode("utf-8")
|
||
self.assertEqual(response.status, 200)
|
||
self.assertIn("text/html", response.getheader("Content-Type"))
|
||
self.assertIn("Acme Plumbing", body)
|
||
self.assertNotIn("<script", body.lower())
|
||
|
||
def test_mockup_sanitizer_removes_active_markup(self):
|
||
html = server.sanitize_mockup_html('<html><body onload="bad"><script>alert(1)</script><a href="javascript:bad()">Hi</a></body></html>')
|
||
self.assertIn("<html>", html)
|
||
self.assertNotIn("script", html.lower())
|
||
self.assertNotIn("onload", html.lower())
|
||
self.assertNotIn("javascript:", html.lower())
|
||
|
||
def test_multipage_mockup_exposes_real_page_routes(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Brightspark Electrical", "maps:bright", "electrician", server.now()))
|
||
conn.commit(); conn.close()
|
||
status, created = self.request("POST", "/api/prospects/1/mockup", {"mode": "multipage"})
|
||
self.assertEqual(status, 201)
|
||
self.assertEqual(created["mode"], "multipage")
|
||
self.assertEqual({item["slug"] for item in created["pages"]}, {"home", "services", "about", "contact"})
|
||
self.conn.request("GET", "/mockups/" + created["token"] + "/services")
|
||
response = self.conn.getresponse()
|
||
body = response.read().decode("utf-8")
|
||
self.assertEqual(response.status, 200)
|
||
self.assertIn("Services", body)
|
||
self.assertIn("images.unsplash.com", body)
|
||
self.assertIn("https://images.unsplash.com", response.getheader("Content-Security-Policy"))
|
||
self.assertIn("frame-ancestors 'none'", response.getheader("Content-Security-Policy"))
|
||
|
||
def test_mockup_editor_creates_a_new_version_and_keeps_original(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Brightspark Electrical", "maps:bright-editor", "electrician", server.now()))
|
||
conn.commit(); conn.close()
|
||
status, original = self.request("POST", "/api/prospects/1/mockup", {"mode": "multipage"})
|
||
self.assertEqual(status, 201)
|
||
self.assertEqual(original.get("version"), None) # v1 rows remain backwards-compatible
|
||
conn = server.db()
|
||
for key, value in (("AI_PROVIDER", "nous"), ("AI_MODEL", "meituan/longcat-2.0:free"), ("AI_API_KEY", "test-key")):
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
|
||
conn.commit(); conn.close()
|
||
ai_response = {"choices": [{"message": {"content": json.dumps({
|
||
"tagline": "Powerful work. Presented with confidence.",
|
||
"intro": "A sharper digital home for Brightspark Electrical.",
|
||
"services": [{"name": "Electrical services", "description": "A clear way to understand the work available."}],
|
||
"about": "A refined concept based on the public business listing.",
|
||
"cta": "Let’s make the next step easy.",
|
||
"hero_image": server.STOCK_IMAGE_LIBRARY["electric"][0],
|
||
"gallery_image": server.STOCK_IMAGE_LIBRARY["electric"][1],
|
||
})}}]}
|
||
with patch("api.server.http_json", return_value=ai_response) as request:
|
||
status, edited = self.request("POST", f"/api/mockups/{original['id']}/edit", {"instruction": "Make the whole site darker and put more emphasis on services.", "page": "all"})
|
||
self.assertEqual(status, 201)
|
||
self.assertEqual(edited["version"], 2)
|
||
self.assertEqual(edited["parent_id"], original["id"])
|
||
self.assertEqual(edited["mode"], "multipage")
|
||
self.assertIn("Applied your request", edited["change_summary"])
|
||
self.assertIn("max_tokens", request.call_args.args[2])
|
||
self.assertEqual(request.call_args.args[2]["max_tokens"], 1800)
|
||
self.conn.request("GET", "/mockups/" + edited["token"] + "/services?embed=1")
|
||
response = self.conn.getresponse(); body = response.read().decode("utf-8")
|
||
self.assertEqual(response.status, 200)
|
||
self.assertIn("frame-ancestors 'self'", response.getheader("Content-Security-Policy"))
|
||
self.assertIn("--paper:#091321", body)
|
||
self.assertIn("Electrical services", body)
|
||
self.conn.request("GET", "/mockups/" + edited["token"])
|
||
response = self.conn.getresponse(); edited_home = response.read().decode("utf-8")
|
||
self.assertEqual(response.status, 200)
|
||
self.assertIn("Powerful work", edited_home)
|
||
self.conn.request("GET", "/mockups/" + original["token"] + "/services")
|
||
response = self.conn.getresponse(); original_body = response.read().decode("utf-8")
|
||
self.assertEqual(response.status, 200)
|
||
self.assertIn("Brightspark Electrical", original_body)
|
||
status, metadata = self.request("GET", f"/api/mockups/{edited['id']}")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(metadata["version"], 2)
|
||
status, listed = self.request("GET", "/api/prospects/1/mockups")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual([item["version"] for item in listed["items"]], [2, 1])
|
||
|
||
def test_multipage_ai_uses_compact_content_and_renders_all_pages(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Brightspark Electrical", "maps:bright-ai", "electrician", server.now()))
|
||
for key, value in (("AI_PROVIDER", "nous"), ("AI_MODEL", "meituan/longcat-2.0:free"), ("AI_API_KEY", "test-key")):
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
|
||
conn.commit(); conn.close()
|
||
response = {"choices": [{"message": {"content": json.dumps({
|
||
"tagline": "Powerful work. Presented with clarity.",
|
||
"intro": "A considered digital home for Brightspark Electrical.",
|
||
"services": [{"name": "Electrical services", "description": "A clear way to understand the work available."}],
|
||
"about": "A refined concept grounded in the public business listing.",
|
||
"cta": "Let’s make the next step easy.",
|
||
"hero_image": server.STOCK_IMAGE_LIBRARY["electric"][0],
|
||
"gallery_image": server.STOCK_IMAGE_LIBRARY["electric"][1],
|
||
})}}]}
|
||
with patch("api.server.http_json", return_value=response) as request:
|
||
status, created = self.request("POST", "/api/prospects/1/mockup", {"mode": "multipage"})
|
||
self.assertEqual(status, 201)
|
||
self.assertEqual(created["generation_source"], "ai")
|
||
self.assertEqual({item["slug"] for item in created["pages"]}, {"home", "services", "about", "contact"})
|
||
payload = request.call_args.args[2]
|
||
self.assertEqual(payload["max_tokens"], 1800)
|
||
self.assertNotIn("pages", json.dumps(payload["messages"]))
|
||
stored = server.db().execute("SELECT pages_json FROM mockups WHERE id=?", (created["id"],)).fetchone()
|
||
pages = json.loads(stored[0])
|
||
self.assertIn("Powerful work", pages["home"])
|
||
self.assertTrue(all("Brightspark Electrical" in html and "@keyframes" in html and "<footer" in html for html in pages.values()))
|
||
|
||
def test_onepage_mockup_uses_in_page_navigation(self):
|
||
html = server.premium_render_pages({"name": "Acme", "category": "plumber"}, mode="onepage")["home"]
|
||
self.assertIn('href="#services"', html)
|
||
self.assertIn('id="services"', html)
|
||
self.assertIn('id="about"', html)
|
||
self.assertIn('id="contact"', html)
|
||
|
||
def test_browser_investigator_normalises_and_filters_candidates(self):
|
||
raw = '{"candidates":[{"url":"https://www.acme.co.za/about","confidence":88,"evidence":["name match"]},{"url":"https://facebook.com/acme","confidence":99},{"domain":"other.co.za","confidence":"42","evidence":"not a list"}]}'
|
||
candidates = browser_investigator._parse_agent_result(raw)
|
||
self.assertEqual([item["domain"] for item in candidates], ["acme.co.za", "other.co.za"])
|
||
self.assertEqual(candidates[0]["confidence"], 88)
|
||
self.assertEqual(candidates[1]["evidence"], [])
|
||
|
||
def test_direct_search_scoring_recognises_compact_business_domain(self):
|
||
prospect = {"name": "BRIGHTSPARK ELECTRICAL Services", "address": '{"city":"Cape Town"}', "phone": "067 732 2090"}
|
||
confidence, evidence = browser_investigator._candidate_score(
|
||
prospect, "https://brightsparkelec.co.za/", "Brightspark Electrical Services", "Cape Town 067 732 2090"
|
||
)
|
||
self.assertGreaterEqual(confidence, 80)
|
||
self.assertTrue(any("domain" in item for item in evidence))
|
||
|
||
def test_website_candidates_are_limited_to_coza_and_com(self):
|
||
raw = '{"candidates":[{"url":"https://bright.example.org","confidence":99},{"url":"https://brightsparkelec.co.za","confidence":80},{"url":"https://brightspark.com","confidence":75}]}'
|
||
candidates = browser_investigator._parse_agent_result(raw)
|
||
self.assertEqual([item["domain"] for item in candidates], ["brightsparkelec.co.za", "brightspark.com"])
|
||
|
||
def test_direct_search_rejects_generic_result_without_identity_signal(self):
|
||
prospect = {"name": "BRIGHTSPARK ELECTRICAL Services", "address": '{"city":"Cape Town"}'}
|
||
confidence, evidence = browser_investigator._candidate_score(
|
||
prospect, "https://unrelated-electrical.co.za/", "Electrical services in Cape Town", "Directory listing"
|
||
)
|
||
self.assertEqual(confidence, 0)
|
||
self.assertEqual(evidence, [])
|
||
|
||
def test_dom_search_candidates_are_saved_before_browser_agent(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,address,discovered_at) VALUES(?,?,?,?,?)", ("Brightspark Electrical Services", "maps:bright", "Electrician", "Cape Town", server.now()))
|
||
conn.commit(); conn.close()
|
||
found = [{"url": "https://brightsparkelec.co.za/", "domain": "brightsparkelec.co.za", "confidence": 88, "evidence": ["visible on bing search results"], "source": "browser_dom"}]
|
||
with patch("api.server.direct_search_candidates", return_value=found) as direct, patch("api.server.investigate_websites") as agent:
|
||
status, body = self.request("POST", "/api/prospects/1/website-candidates", {})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["source"], "browser_dom")
|
||
self.assertEqual(body["items"][0]["domain"], "brightsparkelec.co.za")
|
||
self.assertEqual(body["items"][0]["source"], "browser_dom")
|
||
direct.assert_called_once()
|
||
agent.assert_not_called()
|
||
|
||
def test_browser_website_candidates_can_be_found_and_accepted(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,address,discovered_at) VALUES(?,?,?,?,?)", ("Acme Plumbing", "maps:abc", "plumber", "Cape Town", server.now()))
|
||
conn.commit(); conn.close()
|
||
found = [{"url": "https://acmeplumbing.co.za", "domain": "acmeplumbing.co.za", "confidence": 91, "evidence": ["name match", "location match"]}]
|
||
with patch("api.server.investigate_websites", return_value=found):
|
||
status, body = self.request("POST", "/api/prospects/1/website-candidates", {})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["count"], 1)
|
||
candidate_id = body["items"][0]["id"]
|
||
status, body = self.request("GET", "/api/prospects/1/website-candidates")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["items"][0]["domain"], "acmeplumbing.co.za")
|
||
status, body = self.request("PATCH", f"/api/prospects/1/website-candidates/{candidate_id}", {"status": "accepted"})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["candidate"]["status"], "accepted")
|
||
self.assertEqual(body["prospect"]["domain"], "acmeplumbing.co.za")
|
||
self.assertEqual(body["prospect"]["website"], "https://acmeplumbing.co.za")
|
||
|
||
def test_email_draft_returns_structured_review_notes(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,search_category,category,address,maps_url,website_status,emails,opportunity_score,score_reasons,discovered_at) VALUES(?,?,?,?,?,?,?,?,?,?,?)", ("Acme Plumbing", "maps:abc", "plumbers", "Plumber", "Cape Town", "https://maps.example/acme", "no_detected_website", "acme@gmail.com", 88, '["No detected website","Public email found"]', server.now()))
|
||
for key, value in (("AI_PROVIDER", "custom"), ("AI_BASE_URL", "https://ai.example/v1"), ("AI_MODEL", "test-model"), ("AI_API_KEY", "test-key")):
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
|
||
conn.commit(); conn.close()
|
||
content = json.dumps({"subject": "A website idea for Acme Plumbing", "message": "Good day Acme Plumbing,\n\nI noticed your public listing does not show a dedicated website. We can help with a professional online presence. Would you like me to send a few details?\n\nKind regards", "rationale": "Uses the missing-website signal and the supplied offer.", "recommended_offer": "Gold Plus hosting and a free one-page website", "unknowns": ["Confirm the preferred contact person"], "evidence_used": ["prospect_record", "public_maps_listing", "public_email_listing"], "follow_up": "Confirm the offer and recipient before sending."})
|
||
with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}) as http_json:
|
||
status, body = self.request("POST", "/api/prospects/1/message", {"offer": "Gold Plus hosting at R119/month with a free one-page website.", "tone": "consultative", "goal": "review"})
|
||
self.assertEqual(status, 200)
|
||
request_payload = http_json.call_args.args[2]
|
||
prompt = json.loads(request_payload["messages"][1]["content"])
|
||
self.assertIn('"tone": "consultative"', request_payload["messages"][1]["content"])
|
||
self.assertIn("client_specific_strategy", prompt["evidence"])
|
||
self.assertIn("private planning notes", request_payload["messages"][0]["content"])
|
||
self.assertIn("Do not mention Google", request_payload["messages"][0]["content"])
|
||
self.assertEqual(request_payload["temperature"], 0.65)
|
||
self.assertIn("consultative and helpful", request_payload["messages"][0]["content"])
|
||
self.assertEqual(body["unknowns"], ["Confirm the preferred contact person"])
|
||
self.assertEqual(body["evidence_used"], ["prospect_record", "public_maps_listing", "public_email_listing"])
|
||
saved = server.db().execute("SELECT ai_unknowns,ai_evidence_ids,ai_follow_up FROM prospects WHERE id=1").fetchone()
|
||
self.assertEqual(json.loads(saved[0]), body["unknowns"])
|
||
self.assertEqual(json.loads(saved[1]), body["evidence_used"])
|
||
|
||
def test_message_style_pass_removes_stock_openers_without_rewriting_facts(self):
|
||
text = server._clean_generated_message(
|
||
"Good day Acme Plumbing,\n\nI hope you are well. I noticed your business listing does not show a website.\n\nKind regards",
|
||
"Acme Plumbing",
|
||
)
|
||
self.assertTrue(text.startswith("Hi Acme Plumbing,"))
|
||
self.assertNotIn("I hope you are well", text)
|
||
self.assertNotIn("I noticed your business", text)
|
||
self.assertIn("does not show a website", text)
|
||
self.assertIn("Kind regards", text)
|
||
|
||
def test_message_strategy_uses_concise_greeting_and_missing_site_shape(self):
|
||
prospect = {
|
||
"id": 7,
|
||
"name": "Ash Electrical Cape Town",
|
||
"address": "Salt River, Cape Town",
|
||
"search_category": "Electricians",
|
||
"website_status": "no_detected_website",
|
||
}
|
||
self.assertEqual(server._message_display_name(prospect), "Ash Electrical")
|
||
strategy = server._message_strategy_context(prospect)
|
||
self.assertEqual(strategy["business_name_for_greeting"], "Ash Electrical")
|
||
self.assertIn("looking at local Electricians in Cape Town", strategy["preferred_message_shape"])
|
||
self.assertIn("dedicated website", strategy["preferred_message_shape"])
|
||
|
||
def test_email_draft_can_include_latest_mockup_link(self):
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Acme", "maps:acme", "plumber", server.now()))
|
||
conn.execute("INSERT INTO mockups(organization_id,prospect_id,token,title,html,created_at,updated_at) VALUES(?,?,?,?,?,?,?)", (1, 1, "a" * 24, "Acme · Website concept", "<html><body>Acme</body></html>", server.now(), server.now()))
|
||
for key, value in (("AI_PROVIDER", "custom"), ("AI_BASE_URL", "https://ai.example/v1"), ("AI_MODEL", "test-model"), ("AI_API_KEY", "test-key")):
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
|
||
conn.commit(); conn.close()
|
||
content = json.dumps({"subject": "A concept for Acme", "message": "Good day Acme,\n\nWe can help.", "rationale": "Uses the supplied record.", "recommended_offer": "Website", "unknowns": [], "evidence_used": ["prospect_record"], "follow_up": "Review before sending."})
|
||
with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}):
|
||
status, body = self.request("POST", "/api/prospects/1/message", {"offer": "A modern website"})
|
||
self.assertEqual(status, 200)
|
||
self.assertIn("/mockups/", body["message"])
|
||
self.assertIn("mockup_preview", body["evidence_used"])
|
||
self.assertTrue(body["mockup_url"])
|
||
|
||
def test_email_draft_chat_revision_uses_current_text_and_persists(self):
|
||
conn = server.db()
|
||
conn.execute(
|
||
"INSERT INTO prospects(name,domain,search_category,category,address,maps_url,website_status,emails,opportunity_score,ai_subject,ai_message,ai_offer,discovered_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
("Acme Plumbing", "maps:chat", "plumbers", "Plumber", "Cape Town", "https://maps.example/acme", "no_detected_website", "acme@gmail.com", 88,
|
||
"A practical website idea for Acme", "Good day Acme Plumbing,\n\nYour public listing does not show a dedicated website. We can help.\n\nKind regards", "Gold Plus hosting at R119/month with a free rebuild.", server.now()),
|
||
)
|
||
for key, value in (("AI_PROVIDER", "custom"), ("AI_BASE_URL", "https://ai.example/v1"), ("AI_MODEL", "test-model"), ("AI_API_KEY", "test-key")):
|
||
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?)", (key, value, server.now()))
|
||
conn.commit(); conn.close()
|
||
content = json.dumps({
|
||
"subject": "A softer website idea for Acme",
|
||
"message": "Good day Acme Plumbing,\n\nYour public listing does not show a dedicated website. We can help with Gold Plus hosting at R119/month and a free rebuild. Would a short chat be useful?\n\nKind regards",
|
||
"rationale": "Shortened the opening and softened the call to action as requested.",
|
||
"recommended_offer": "Gold Plus hosting at R119/month with a free rebuild.",
|
||
"unknowns": ["Confirm the preferred contact person"],
|
||
"evidence_used": ["prospect_record", "public_maps_listing", "public_email_listing"],
|
||
"follow_up": "Check the wording and offer before sending.",
|
||
})
|
||
with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}) as http_json:
|
||
status, body = self.request("POST", "/api/prospects/1/message/edit", {
|
||
"instruction": "Make it warmer, shorter, and use a softer call to action.",
|
||
"current_subject": "A practical website idea for Acme",
|
||
"current_message": "Good day Acme Plumbing,\n\nYour public listing does not show a dedicated website. We can help.\n\nKind regards",
|
||
"offer": "Gold Plus hosting at R119/month with a free rebuild.",
|
||
"tone": "warm",
|
||
"goal": "conversation",
|
||
"include_mockup": False,
|
||
})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["subject"], "A softer website idea for Acme")
|
||
self.assertIn("softened", body["rationale"])
|
||
request_payload = http_json.call_args.args[2]
|
||
prompt = json.loads(request_payload["messages"][1]["content"])
|
||
self.assertEqual(prompt["current_draft"]["subject"], "A practical website idea for Acme")
|
||
self.assertIn("Make it warmer", prompt["edit_instruction"])
|
||
self.assertEqual(request_payload["max_tokens"], 1000)
|
||
self.assertIn("cannot add facts", request_payload["messages"][0]["content"])
|
||
saved = server.db().execute("SELECT ai_subject,ai_message,ai_offer FROM prospects WHERE id=1").fetchone()
|
||
self.assertEqual(saved[0], body["subject"])
|
||
self.assertEqual(saved[1], body["message"])
|
||
self.assertEqual(saved[2], "Gold Plus hosting at R119/month with a free rebuild.")
|
||
status, logs = self.request("GET", "/api/logs?kind=ai")
|
||
self.assertEqual(status, 200)
|
||
self.assertTrue(any(item["event"] == "message_revision_completed" for item in logs["items"]))
|
||
|
||
def test_logs_capture_api_actions_and_redact_credentials(self):
|
||
server.record_log("ai", "test_ai_call", details={"api_key": "do-not-store", "model": "test", "prompt": "hello"})
|
||
status, body = self.request("POST", "/api/logs", {"action": "ui_test", "details": {"button": "test"}})
|
||
self.assertEqual(status, 202)
|
||
status, body = self.request("GET", "/api/logs?kind=ai")
|
||
self.assertEqual(status, 200)
|
||
item = next(row for row in body["items"] if row["event"] == "test_ai_call")
|
||
self.assertEqual(item["details"]["api_key"], "[redacted]")
|
||
self.assertEqual(item["details"]["model"], "test")
|
||
status, body = self.request("GET", "/api/logs?kind=action")
|
||
self.assertEqual(status, 200)
|
||
self.assertTrue(any(row["event"] == "ui_test" for row in body["items"]))
|
||
|
||
def test_runtime_settings_persist_and_change_live_limits(self):
|
||
status, body = self.request("POST", "/api/settings", {
|
||
"browser_use": {"enabled": True, "max_steps": 22, "timeout_seconds": 900, "llm_timeout_seconds": 120},
|
||
"maps": {"request_timeout_seconds": 1800, "poll_seconds": 4, "max_depth": 3},
|
||
"limits": {"website_scan_timeout_seconds": 15, "ai_mockup_timeout_seconds": 75},
|
||
})
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(body["browser_use"]["timeout_seconds"], 900)
|
||
self.assertEqual(body["maps"]["request_timeout_seconds"], 1800)
|
||
self.assertEqual(body["limits"]["website_scan_timeout_seconds"], 15)
|
||
self.assertEqual(body["limits"]["ai_mockup_timeout_seconds"], 75)
|
||
status, settings = self.request("GET", "/api/settings")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(settings["browser_use"]["max_steps"], 22)
|
||
self.assertEqual(settings["browser_use"]["llm_timeout_seconds"], 120)
|
||
self.assertEqual(settings["maps"]["poll_seconds"], 4.0)
|
||
self.assertEqual(settings["maps"]["max_depth"], 3)
|
||
self.assertEqual(settings["limits"]["ai_mockup_timeout_seconds"], 75)
|
||
self.assertEqual(server.browser_use_limits(), (22, 900))
|
||
self.assertEqual(server.website_scan_timeout_seconds(), 15)
|
||
|
||
def test_company_profile_and_promotions_feed_email_draft(self):
|
||
status, profile = self.request("POST", "/api/settings", {"company_profile": {
|
||
"name": "Marco Web Studio", "description": "A South African hosting studio.",
|
||
"services": "Domains, hosting and website rebuilds.", "packages": "Gold Plus hosting at R119/month.",
|
||
}})
|
||
self.assertEqual(status, 200)
|
||
self.assertTrue(profile["company_profile"]["configured"])
|
||
self.request("POST", "/api/settings", {"ai": {"AI_PROVIDER": "custom", "AI_BASE_URL": "https://ai.example/v1", "AI_MODEL": "test-model", "AI_API_KEY": "test-key"}})
|
||
status, promotion = self.request("POST", "/api/promotions", {"name": "Free rebuild", "description": "Free website rebuild when hosting moves to Gold Plus.", "terms": "New hosting customers only."})
|
||
self.assertEqual(status, 201)
|
||
status, listed = self.request("GET", "/api/promotions")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(listed["items"][0]["name"], "Free rebuild")
|
||
conn = server.db()
|
||
conn.execute("INSERT INTO prospects(name,domain,category,discovered_at) VALUES(?,?,?,?)", ("Acme", "maps:acme", "plumber", server.now()))
|
||
conn.commit(); conn.close()
|
||
content = json.dumps({"subject": "A free website rebuild", "message": "Good day Acme,\n\nWe can help.", "rationale": "Uses the supplied promotion.", "recommended_offer": "Free rebuild", "unknowns": [], "evidence_used": ["prospect_record"], "follow_up": "Review before sending."})
|
||
with patch("api.server.http_json", return_value={"choices": [{"message": {"content": content}}]}) as http_json:
|
||
status, body = self.request("POST", "/api/prospects/1/message", {"offer": "Please mention our offer.", "promotion_id": promotion["id"]})
|
||
self.assertEqual(status, 200)
|
||
prompt = http_json.call_args.args[2]["messages"][1]["content"]
|
||
self.assertIn("Free website rebuild", prompt)
|
||
self.assertIn("Gold Plus hosting at R119/month", prompt)
|
||
saved = server.db().execute("SELECT ai_promotion_id FROM prospects WHERE id=1").fetchone()
|
||
self.assertEqual(saved[0], promotion["id"])
|
||
|
||
@patch("api.server.ai_assess_prospect", return_value={"ai_probability": 88, "ai_recommendation": "contact", "ai_reasoning": "Active business with a broken site."})
|
||
def test_workspace_workflow_and_assessment_history(self, assess):
|
||
with patch("api.server.google_search", return_value=[{"name": "Acme", "website": "https://acme.co.za", "domain": "acme.co.za"}]):
|
||
_, run = self.request("POST", "/api/discover", {"category": "plumbers", "keywords": "plumbing", "location": "Cape Town"})
|
||
self.wait_for_run(run["id"])
|
||
conn = server.db(); conn.execute("INSERT INTO users(organization_id,email,name,password_hash,role,active,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)", (1, "reviewer@example.com", "Reviewer", "", "reviewer", 1, server.now(), server.now())); conn.commit(); user_id = conn.execute("SELECT id FROM users WHERE email='reviewer@example.com'").fetchone()[0]; conn.close()
|
||
status, note = self.request("POST", "/api/prospects/1/notes", {"body": "Call after reviewing the website."}); self.assertEqual(status, 201)
|
||
status, task = self.request("POST", "/api/prospects/1/tasks", {"title": "Send tailored draft", "due_at": "2030-01-01"}); self.assertEqual(status, 201)
|
||
status, assigned = self.request("POST", "/api/prospects/1/assignment", {"assigned_user_id": user_id}); self.assertEqual(status, 200); self.assertEqual(assigned["assigned_user_id"], user_id)
|
||
status, body = self.request("POST", "/api/prospects/1/assess", {}); self.assertEqual(status, 200); self.assertEqual(body["ai_probability"], 88)
|
||
status, history = self.request("GET", "/api/prospects/1/history"); self.assertEqual(status, 200); self.assertTrue(history["score_history"]); self.assertTrue(history["ai_history"])
|
||
status, tasks = self.request("GET", "/api/prospects/1/tasks"); self.assertEqual(status, 200); self.assertEqual(tasks["items"][0]["title"], "Send tailored draft")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|