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 class BrowserOnlyAppTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.NamedTemporaryFile(delete=False) self.tmp.close() server.DB_PATH = self.tmp.name 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): 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) 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_nous_provider_exposes_models_and_tests_connection(self): 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_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.")) if __name__ == "__main__": unittest.main()